[Rx] 効率の良いストリームの合成方法

Question

ご質問させていただきます。よろしくお願いいたします。
Unity用のRxプラグインを使用させていただき開発しております。その中でストリームの合成がすごく難しくコードが複雑になってしまいこれで正しいのか分からないのでご質問させていただきたいです。

###環境
・Unity 2017.2.0p4
・UniRx 5.5.0

###実装したい処理

1. (初期化) Vector2のデータをランダムに1つ生成
2. (初期化) ランダムに1度00秒遅延させる
3. (開始) UpdateAsObservableでループ開始
4. (判定) 1で生成したデータとオブジェクトの距離が一定の距離になるまで処理を続ける
5. (OnNext) Subscribeで1で生成したデータに徐々に近づける
6. (Oncompleted) 完了後1から繰り返し

###実現したいこと
コードが複雑になっているので簡潔な方法がありましたらご教授いただきたいです。プロパティまたはフィールド(クロージャ)などの外部からの参照無しで実装したいです。言語は問いません。大変お手数ですがよろしくお願いいたします。

###現状のコード

static readonly float WALK_SPEED = .7f;
static readonly float RAND_INTERVAL_MIN = 4;
static readonly float RAND_INTERVAL_MAX = 10;
static readonly float DISTANCE = .05f;
static readonly Vector2 RAND_POS_LEFT_TOP = new Vector2(-1, 3);
static readonly Vector2 RAND_POS_RIGHT_BOTTOM = new Vector2(-2, 4);
Transform MyTransform;

void Start ()
{
  MyTransform = gameObject.transform;

  Observable.Empty<Vector2>().StartWith(() => GetRandPos())
    .SelectMany(pos => 
      Observable.Timer(TimeSpan.FromSeconds(UnityEngine.Random.Range(RAND_INTERVAL_MIN, RAND_INTERVAL_MAX)))
      .Select(_ => pos)
    )
    .SelectMany(pos => this.UpdateAsObservable().Select(_ => pos))
    .TakeWhile(pos => Vector2.Distance(pos, MyTransform.position) >= DISTANCE)
    .Select(pos => GetNextPos(pos, MyTransform.position))
    .RepeatUntilDestroy(this)
    .Subscribe(pos => MyTransform.position = pos);
}

Vector2 GetNextPos(Vector2 target,Vector2 me)
{
  var rad = Mathf.Atan2(target.y - me.y, target.x - me.x);

  me.x += WalkSpeed * Time.deltaTime * Mathf.Cos(rad);
  me.y += WalkSpeed * Time.deltaTime * Mathf.Sin(rad);
  return me;
}

Vector2 GetRandPos()
{
  var x = UnityEngine.Random.Range(
    RAND_POS_LEFT_TOP.x,
    RAND_POS_RIGHT_BOTTOM.x
  );

  var y = UnityEngine.Random.Range(
    RAND_POS_LEFT_TOP.y,
    RAND_POS_RIGHT_BOTTOM.y
  );

  return new Vector2(x, y);
}

Answer 1

遅延抜きで繰り返し1回部分のストリーム構築を考えてみます。

求めたい座標のストリーム(一回分)をys_、
元となる座標のストリームをxsとします。

ys_はxsの最新値とys_の最後の値から構築されるのでScan()オペレータを使います
ys_はある条件で終了するのでTakeWhile()オペレータを使います

ys_ = xs.Scan(...).TakeWhile(...)

説明のためys_の初期値y0を与えてys_を構築する関数をfと置きます。
f = y0 => ys_

次に遅延部分を考えます。

Observable.Timer()を使い一つだけ値をランダムな遅延させて流します。
流れてきた値は捨てて、y0を生成して差し替えます。
これにSelectMany(f)をくっつければ遅延してys_を生成して流してくれるはず

Observable.Timer(...).Select(...).SelectMany(f)

後は全体をRepeat()すれば繰り返すはず

ys = Observable.Timer(...).Select(...).SelectMany(f).RepeatUntilDestroy(...)

UniRxではなくRxで、Vector2でなく1次元でとりあえずやってみましたが
あまり簡潔になってないですね。すみません。

// テスト用にもととなる座標ストリームをでっち上げる
var xs = Observable.Interval(TimeSpan.FromSeconds(1))
    .Select(x => (double)x)
    .Take(60).Publish().RefCount();


var ys = Observable.Timer(TimeSpan.FromMilliseconds(rand.Next(1500, 4500)))
    .Select(_ => GetRandPos())
    .SelectMany(y0 =>
        xs.Scan(new { Y = y0, Dist = 10.0 }, // 初っ端でTakeWhileに引っかからないように(雑)
            (acc, x) =>
            {
                var y = GetNextPos(x, acc.Y);
                return new { Y = y, Dist = Math.Abs(x - y) };
            })
        .TakeWhile(acc => acc.Dist > 1.0)
        .Select(acc => acc.Y))
    .Repeat();


xs.Timestamp().Subscribe(x => Console.WriteLine($"x , {x.Timestamp.Ticks} , {x.Value}"));
ys.Timestamp().Subscribe(y => Console.WriteLine($"y , {y.Timestamp.Ticks} , {y.Value}"));

適当にGetHOGE()を設定して動かしてみました:

横軸:時間
縦軸:座標

---

回答ではなく質問です。追記依頼覧が狭いので。
(聞いたところで難しそうなので回答できるかはちょっと...すみません)

• ある移動体Pの座標(ストリーム)から、ある規則で動く別の移動体Qの座標(ストリーム)を求めたい。
• Qの座標はPの座標とQの前回座標から求められる。(GetNextPos())
• Qの初期座標はランダムに与えられる。(GetRandPos())
• PとQの距離が一定以下になったら、Qの座標はランダムに与え直される。

ということでよろしいでしょうか?

ランダムな遅延が読み解けなかったんですが、
目的のストリームはどちらですか?

1,2,...,5 座標の出力タイミング
R リセット

1)

|-------(Delay)------ 1 --- 2 --- 3 --- 4 --- R -----(Delay)---- 5 --->

2)

|1 ------(Delay)------ 2 --- 3 --- 4 --- R5 ---(Delay)--->

Comments

[IShix]

説明不足ですみません。ご返答ありがとうございます。箇条書きの内容に間違いありません。図にしていただいているものは1)が目的のストリームとなります。

[IShix]

具体的にご教授いただきありがとうございます。
Scanを使用したことがないので、いただいたコードを元に勉強させていただます。Timestampはこういう風に使うんですね。大変勉強になりました。Rx凄く難しいですが便利なので理解を更に深めていきたいと思います。

[ozwk]

今更ですが気になっていたので

ラムダ式のアンダースコアは「捨てられる引数」など、
どうでもいい値に使うという暗黙の了解があります(私だけ?)
なもので、
.TakeWhile(_ => Vector2.Distance(_, MyTransform.position) >= DISTANCE)
のように、ラムダ式の本体側に_が出てくるとびっくりします。

[IShix]

確かにそうですね。ご指摘ありがとうございます。コードも修正しました。