RxJavaでセンサ値を扱うと便利

## TL;DR

* センサ値にフィルタかけたい
* それ，[RxJava][rxjava]でも同じことできるよ
* `rx.Observable`から値を取り出すことになるので，`flatMap`で色んな処理に繋げられて便利

## Filter
Androidで加速度を取得するとき，大量にノイズ拾ってすごいギザギザしたグラフになっ たりする．そこで，普通はセンサ値にフィルタかけてノイズを除去する．よくあるのはMedian Filter，Low Pass Filter，High Pass Filterなど（参考: [SensorEvent | Android Developers][sensorevent]）．

### Median Filter
複数サンプリングしたデータのうち，中央値（median）を採用する．スパイクノイズに強い．

### Low Pass Filter
データから高周波成分を取り除き，平滑化する．こまかいギザギザしたノイズを除去できる．

### High Pass Filter
データから低周波成分を取り除く．加速度なら重力の影響を除去できる．

## Naive implementation

普通にやるとこんな感じ（Low Pass Filter + Median Filter）．

```java
private final List<Vector3d> samples = new ArrayList<>();
private Vector3d acc;

public void onSensorChanged(SensorEvent event) {
    samples.add(new Vector3d(event.values.clone());

    if (samples.size() == SAMPLE_COUNT) {
        final List<Float> xSamples = new ArrayList<>();
        final List<Float> ySamples = new ArrayList<>();
        final List<Float> zSamples = new ArrayList<>();

        for (Vector3d sample : samples) {
            xSamples.add(sample.x);
            ySamples.add(sample.y);
            zSamples.add(sample.z);
        }

        Collections.sort(xSamples);
        Collections.sort(ySamples);
        Collections.sort(zSamples);

        float x = ALPHA * acc.x + xSamples.get(SAMPLE_COUNT / 2);
        float y = ALPHA * acc.y + ySamples.get(SAMPLE_COUNT / 2);
        float z = ALPHA * acc.z + zSamples.get(SAMPLE_COUNT / 2);

        acc = new Vector3d(x, y, z);

        samples.remove(0);
    }
}

public Vector3d getAcceleration() {
    return acc;
}
```

共通のクラスとか定数は以下:


```java
class Vector3d {
    private final float x, y, z;

    public Vector3d(float[] values) {
        this(values[0], values[1], values[2]);
    }

    public Vector3d(float x, float y, float z) {
        this.x = x;
        this.y = y;
        this.z = z;
    }

    public float getX() {
        return x;
    }

    public float getY() {
        return y;
    }

    public float getZ() {
        return z;
    }
}

private final float ALPHA = 0.8;
private final int SAMPLE_COUNT = 10;
```

## Implementation with RxJava

[**RxJava**][rxjava]で同じことをやってみる．

```java
public Observable<Vector3d> observe() {
    return getObservable()
            .takeLast(SAMPLE_COUNT)
            .toList()
            .filter(vectors -> vectors.size() == SAMPLE_COUNT)
            .map(vectors -> new Vector3d(
                    getMedian(vectors, Vector3d::getX),
                    getMedian(vectors, Vector3d::getY),
                    getMedian(vectors, Vector3d::getZ)
            ))
            .scan((current, next) -> new Vector3d(
                    lpf(current, next, Vector3d::getX),
                    lpf(current, next, Vector3d::getY),
                    lpf(current, next, Vector3d::getZ)
            ));
}

private <T, R> R getMedian(List<T> list, Func1<T, R> func) {
    return Observable.from(list).map(func).toSortedList()
            .map(values -> values.get(SAMPLE_COUNT / 2)).toBlocking().single();
}

private <T, R extends Number> float lpf(T current, T next, Func1<T, R> func) {
    return ALPHA * func.call(current).floatValue() + (1 - ALPHA) * func.call(next).floatValue();
}

// 以下，センサ値をobservableに変換する処理

private final List<Listener> listeners = new ArrayList();

private Observable<Vector3d> getObservable() {
    final PublishSubject<Vector3d> subject = PublishSubject.create();
    final Listener listener = event -> subject.onNext(new Vector3d(event.values.clone()));
    subject.doOnUnsubscribe(() -> listeners.remove(listener));
    listeners.add(listener);
    return subject;
}

@Override
public void onSensorChanged(SensorEvent event) {
    for (Listener listener : mListeners) {
        listener.onSensorChanged(event);
    }
}

private interface Listener {
    void onSensorChanged(SensorEvent event);
}
```

~~あれ，もとより複雑になってる？~~ コードの好みは人それぞれかもしれない．

もうちょい綺麗に書く方法あるかも？

## RxJavaで実装することによるメリット
`rx.Observable`が返るので，たとえば末尾に`map`とかチェインして，値を保存するようにした上で`notifyPropertyChanged()`とか呼べば[Data Binding][databinding]に繋げることも可能．あとは[**RxAndroid**][rxandroid]の`observeOn(AndroidSchedulers.mainThread())`を繋げてUIに反映させたりとか．

この**`rx.Observable`が返る**というのがポイントで，`rx.Observable`を世界標準共通インタフェースとして扱うことが出来ればどんな処理でも`flatMap`でチェイン出来るようになる．たとえば，[Retrofit][retrofit]でAPIクライアント作っておけば，適当に溜め込んだセンサ値をサーバに送りつける みたいなのもさくっと書けるようになって便利．

```java
// `client`はRetrofitでいいカンジに実装したAPIクライアント
observe().buffer(100).flatMap(client::post);
```

## References

* [SensorEvent | Android Developers][sensorevent]
* [ReactiveX/RxJava][rxjava]
* [ReactiveX/RxAndroid][rxandroid]
* [Data Binding Guide | Android Developers][databinding]
* [Retrofit][retrofit]

[sensorevent]: http://developer.android.com/reference/android/hardware/SensorEvent.html
[rxjava]: https://github.com/ReactiveX/RxJava
[rxandroid]: https://github.com/ReactiveX/RxAndroid
[databinding]: http://developer.android.com/tools/data-binding/guide.html
[retrofit]: http://square.github.io/retrofit/