GitHubじゃ!Pythonじゃ!

GitHubからPython関係の優良リポジトリを探したかったのじゃー、でも英語は出来ないから日本語で読むのじゃー、英語社会世知辛いのじゃー

ReactiveX

RxPY – Reactive Extensions for Python

投稿日:

Reactive Extensions for Python

Reactive Extensions for Python(RxPY)は、

Pythonで観測可能なコレクションとLINQ形式のクエリ演算子を使用して非同期およびイベントベースのプログラムを作成するためのライブラリ

ReactiveXについて

Reactive Extensions for Python(RxPY)は、Pythonで観測可能なシーケンスとLINQスタイルのクエリ演算子を使用して、非同期およびイベントベースのプログラムを作成するためのライブラリセットです。 Rxを使用することで、開発者はObservablesで非同期データ・ストリームを表現し、演算子を使用して非同期データ・ストリームを照会し、スケジューラーを使用してデータ/イベント・ストリームの並行性をパラメータ化できます。

Rxを使用すると、複数の非同期データストリーム(株価、ツイート、コンピュータイベント、Webサービスリクエストなどのさまざまなソースに由来)を表示し、Observerオブジェクトを使用してイベントストリームを購読することができます。 Observableは、イベントが発生するたびにサブスクライブしたObserverインスタンスに通知します。ソースObservableと消費オブザーバの間にさまざまな変換を加えることができます。

観測可能なシーケンスはデータストリームなので、Observable型で実装された標準のLINQライクなクエリ演算子を使用してクエリを実行できます。 したがって、これらの静的LINQ演算子を使用することで、複数のイベントに対して時間ベースの操作を簡単にフィルター、マップ、縮小、作成および実行することができます。 さらに、強力なクエリを記述できるようにするための、多数の他のリアクティブ・ストリーム固有の演算子があります。 取り消し、例外、および同期もObservableオブジェクトのメソッドを使用して正常に処理されます。

コミュニティ

スラックの会話に参加しよう!

PySlackersの優雅な人々は、私たちに#rxpy Slackチャンネルの家を与えました。 質問、会話、およびRxPyに関連するすべてのことについて、私たちに参加してください。

https://pyslackers.com/

参加するには、上記のページをナビゲートして、Eメールの招待を受け取ります。 申し込み後、#rxpyチャンネルに参加してください。

コミュニティのガイドラインと利用規約に従ってください。

インストール

RxPYは、 Python 2.7,3.4、 PyPyIronPythonで動作します。

RxPYをインストールするには:

pip install rx

pip3を使用している場合、 pippip3と呼ばれることに注意してください。

チュートリアル

基礎

ObservableはReactiveXのコアタイプです。 これは、 排出物として知られている品目を一連のオペレーターを介して最終的にObserverに到着して消費されるまで、連続して押し出します。

プッシュベース(プルベースではなく)の反復処理により、コードと並行処理をはるかに迅速に表現する強力な新しい可能性が開かれます。 Observableはイベントをデータとデータとしてイベントとして扱うため、2つのイベントを一緒に構成することは簡単になります。

Observerアイテムを渡すObservableを作成するには、多くの方法があります。 Observable.create()ファクトリを使用してObservable.create()アイテムを渡すファンクションを渡すことができます。 Observeron_next()on_completed() 、およびon_error()関数を実装しています。 on_next()はアイテムを渡すために使用されます。 on_completed()はそれ以上アイテムが来ないことを通知し、 on_error()はエラーを通知します。

たとえば、これらの3つの方法でObserverを実装し、これらのイベントを単純に印刷することができます。 Observable.create()は、5つの文字列をObserver渡す関数を利用して、これらのイベントを呼び出すことができます。

from rx import Observable, Observer


def push_five_strings(observer):
    observer.on_next("Alpha")
    observer.on_next("Beta")
    observer.on_next("Gamma")
    observer.on_next("Delta")
    observer.on_next("Epsilon")
    observer.on_completed()


class PrintObserver(Observer):

    def on_next(self, value):
        print("Received {0}".format(value))

    def on_completed(self):
        print("Done!")

    def on_error(self, error):
        print("Error Occurred: {0}".format(error))

source = Observable.create(push_five_strings)

source.subscribe(PrintObserver())

出力:

Received Alpha
Received Beta
Received Gamma
Received Delta
Received Epsilon
Done!

しかし、一般的な排出源のためのObservable工場が数多く存在する。 5つのアイテムを単にプッシュするには、 Observable.create()およびそのバッキング関数を取り除き、 Observable.create()を使用します。 このファクトリは引数リストを受け取り、各エミッションをon_next()として反復し、繰り返しが完了したらon_completed()呼び出します。 したがって、これらの5つのStringを引数として渡すことができます。

from rx import Observable, Observer

class PrintObserver(Observer):

    def on_next(self, value):
        print("Received {0}".format(value))

    def on_completed(self):
        print("Done!")

    def on_error(self, error):
        print("Error Occurred: {0}".format(error))

source = Observable.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon")

source.subscribe(PrintObserver())

ほとんどの場合、 Observerを実装することの冗長さを嫌うことはほとんどありませObserver 代わりに、 on_nexton_complete 、およびon_errorアクションを指定するsubscribe() 1から3のλ引数を渡すことができます。

from rx import Observable

source = Observable.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon")

source.subscribe_(on_next=lambda value: print("Received {0}".format(value)),
                 on_completed=lambda: print("Done!"),
                 on_error=lambda error: print("Error Occurred: {0}".format(error))
                 )

3つのイベントタイプをすべて指定する必要はありません。 名前付き引数を使用して監視するイベントを選択して選択するか、 on_next単一のラムダを指定するon_nextです。 通常、本番on_errorは、 on_errorを指定して、エラーが明示的にサブスクライバによって処理されるようにします。

from rx import Observable

source = Observable.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon")

source.subscribe_(lambda value: print("Received {0}".format(value)))

出力:

Received Alpha
Received Beta
Received Gamma
Received Delta
Received Epsilon

演算子と連鎖

RxPYで利用可能な130以上の演算子を使用して、新しいObservableを導出することもできます。 各オペレータは、何らかの形で供給源からの排出を変換する新しいObservableObservableます。 たとえば、各Stringを長さにmap()てから、少なくとも5の長さのfilter()を実行すると、2つのObservableが互いに組み立てられます。

from rx import Observable

source = Observable.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon")

lengths = source.map(lambda s: len(s))

filtered = lengths.filter(lambda i: i >= 5)

filtered.subscribe_(lambda value: print("Received {0}".format(value)))

出力:

Received 5
Received 5
Received 5
Received 7

通常、Observablesを各オペレータの中間変数に保存したくない場合は、その時点で複数のサブスクライバを使用する必要があります。 代わりに、インライン化して操作の「観察可能な連鎖」を作成しようとします。 そうすれば、コードが読みやすくなり、ストーリーをもっと簡単に伝えることができます。

from rx import Observable

Observable.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon") \
    .map(lambda s: len(s)) \
    .filter(lambda i: i >= 5) \
    .subscribe_(lambda value: print("Received {0}".format(value)))

イベントの発生

Observablesはデータの上にイベントを出すこともできます。 データとイベントを同じ方法で扱うことで、強力なコンポジションを作成して2つのデータを連携させることができます。 以下では、1000ミリ秒ごとに連続する整数を出力するObservableがあります。 このObservableは無限に動作し、 on_completeを決して呼び出しon_complete

from rx import Observable

Observable.interval(1000) \
    .map(lambda i: "{0} Mississippi".format(i)) \
    .subscribe_(lambda s: print(s))

input("Press any key to quit\n")

出力:

0 Mississippi
1 Mississippi
2 Mississippi
3 Mississippi
4 Mississippi
5 Mississippi
6 Mississippi
...

Observable.interval()は( TimeoutSchedulerを介して)別のスレッドで動作するため、起動される前にアプリケーションが早期に終了しないようにする必要があります。 キーが押されるまで、 input()を使ってメインスレッドを止めることができます。 オブザーバブルは、ボタンイベント、リクエスト、タイマー、さらにライブのTwitterフィードに対して作成することができます

マルチキャスティング

Observableへの各加入者は、しばしば別々の排出ストリームを受信します。 たとえば、 Observable 2人の加入者を割り当てて3つの乱数を出すと、両方の加入者が異なる数になるでしょう。

from rx import Observable
from random import randint


three_emissions = Observable.range(1, 3)

three_random_ints = three_emissions.map(lambda i: randint(1, 100000))

three_random_ints.subscribe_(lambda i: print("Subscriber 1 Received: {0}".format(i)))
three_random_ints.subscribe_(lambda i: print("Subscriber 2 Received: {0}".format(i)))

出力:

Subscriber 1 Received: 79262
Subscriber 1 Received: 20892
Subscriber 1 Received: 69197
Subscriber 2 Received: 66574
Subscriber 2 Received: 41177
Subscriber 2 Received: 47445

Observableチェーン内の特定のポイントに、各サブスクライバごとに別々の排出ストリームを生成するのではなく、すべてのサブスクライバに同じ排出をプッシュするには、 publish()を呼び出してConnectableObservableを返すことができます。 次に、サブスクライバを設定し、 connect()を呼び出して同じ排出ストリームを受信する準備ができたら、呼び出します。

from rx import Observable
from random import randint


three_emissions = Observable.range(1, 3)

three_random_ints = three_emissions.map(lambda i: randint(1, 100000)).publish()

three_random_ints.subscribe_(lambda i: print("Subscriber 1 Received: {0}".format(i)))
three_random_ints.subscribe_(lambda i: print("Subscriber 2 Received: {0}".format(i)))

three_random_ints.connect()

出力:

Subscriber 1 Received: 90994
Subscriber 2 Received: 90994
Subscriber 1 Received: 91213
Subscriber 2 Received: 91213
Subscriber 1 Received: 42335
Subscriber 2 Received: 42335

これは寒いObservable (各サブスクライバのための操作を「再生する」)をとり、すべてのオブザーバをライブタイムで放送される同じストリームのストリームに置くことによって熱いものにします。 connect()が呼び出された後にサブスクライブする遅刻オブザーバは以前の放出を見逃すconnect()connect()呼び出す前にすべてのオブザーバを設定してください。

mutastastingを実装する別の方法は、 ConnectableObservable auto_connect()演算子を使用することauto_connect() これは、購読者になると同時に発射を開始し、加入者が出入りしても発射を続けます。 整数の引数を指定した場合は、その数のサブスクライバがサブスクライブされるまで、起動しません。

from rx import Observable
from random import randint


three_emissions = Observable.range(1, 3)

three_random_ints = three_emissions.map(lambda i: randint(1, 100000)).publish().auto_connect(2)

three_random_ints.subscribe_(lambda i: print("Subscriber 1 Received: {0}".format(i)))
three_random_ints.subscribe_(lambda i: print("Subscriber 2 Received: {0}".format(i))) # second subscriber triggers firing

オブザーバブルの結合

Observable.merge()Observable.concat()Observable.zip() 、およびObservable.combine_latest()ようObservable.merge()ファクトリを使用して、異なるObservableを一緒に作成できます。 Observablesが( subscribe_on()observe_on()演算子を使用してsubscribe_on()異なるスレッドで作業していても、それらは安全に結合されます。 たとえば、 Observable.zip()を使用すると、 Observable.zip()を使用して5つのStringを圧縮して処理を遅くすることができます。 各ソースから1つのエミッションを取り出し、タプルにまとめます。

from rx import Observable

letters = Observable.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon")

intervals = Observable.interval(1000)

Observable.zip(letters, intervals, lambda s, i: (s, i)) \
    .subscribe_(lambda t: print(t))

input("Press any key to quit\n")

出力:

('Alpha', 0)
('Beta', 1)
('Gamma', 2)
('Delta', 3)
('Epsilon', 4)

あなたはObservablesを事実上あらゆるものから作成することができます。また、カスタマイズされたObservablesを返すAPIやヘルパー関数を作成すると役立つことがよくあります。 たとえば、 SQLAlchemyを使用してObservableSQLクエリから作成し、特定の顧客IDのCUSTOMERテーブルレコードを返すことができます。 flat_map()を使用して、各排出量をObservableにマッピングし、その排出量を1つのObservableマージすることもできます。 これにより、以下に示すように3人の異なる顧客を照会することができます。

from sqlalchemy import create_engine, text
from rx import Observable

engine = create_engine('sqlite:///rexon_metals.db')
conn = engine.connect()


def customer_for_id(customer_id):
    stmt = text("SELECT * FROM CUSTOMER WHERE CUSTOMER_ID = :id")
    return Observable.from_(conn.execute(stmt, id=customer_id))


# Query customers with IDs 1, 3, and 5
Observable.of(1, 3, 5) \
    .flat_map(lambda id: customer_for_id(id)) \
    .subscribe_(lambda r: print(r))

出力:

(1, 'LITE Industrial', 'Southwest', '729 Ravine Way', 'Irving', 'TX', 75014)
(3, 'Re-Barre Construction', 'Southwest', '9043 Windy Dr', 'Irving', 'TX', 75032)
(5, 'Marsh Lane Metal Works', 'Southeast', '9143 Marsh Ln', 'Avondale', 'LA', 79782)

並行性

並行処理を実現するには、 subscribe_on()observe_on() 2つの演算子を使用します。 両方とも、作業を行うために各サブスクリプションにスレッドを提供するSchedulerが必要です(下記のスケジューラのセクションを参照)。 ThreadPoolSchedulerは、再利用可能なワーカースレッドのプールを作成するのに適しています。

PythonのGILは、複数のスレッドが同じコード行に同時にアクセスするのを防ぐので、同時実行性のパフォーマンスを損なう可能性があることに注意してください。 NumPyのようなライブラリは、GILを解放するので、並列集約型計算のためにこれを軽減することができます。 RxPyはスレッドのオーバーラップをある程度抑えることもできます。 アプリケーションを並行性でテストし、パフォーマンスが向上していることを確認してください。

subscribe_on()は、使用するスケジューラのチェーン開始時にObservableソースに指示します(そして、この演算子をどこに置くかは関係ありません)。 ただし、 observe_on()Observableチェーンのその時点で別のScheduler切り替わり、あるスレッドから別のスレッドへ効率よく移動します。 Observable.interval()delay()ようなObservable.interval() Observableファクトリや演算子には既定のSchedulerが既に存在するため、指定したsubscribe_on()は無視されます(通常はScheduler引数として渡すことができます)。

以下では、 observe_on()と同様にsubscribe_on()を順に使用するのではなく、3つの異なるプロセスを同時に実行します。

import multiprocessing
import random
import time
from threading import current_thread

from rx import Observable
from rx.concurrency import ThreadPoolScheduler


def intense_calculation(value):
    # sleep for a random short duration between 0.5 to 2.0 seconds to simulate a long-running calculation
    time.sleep(random.randint(5, 20) * .1)
    return value


# calculate number of CPU's, then create a ThreadPoolScheduler with that number of threads
optimal_thread_count = multiprocessing.cpu_count()
pool_scheduler = ThreadPoolScheduler(optimal_thread_count)

# Create Process 1
Observable.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon") \
    .map(lambda s: intense_calculation(s)) \
    .subscribe_on(pool_scheduler) \
    .subscribe_(on_next=lambda s: print("PROCESS 1: {0} {1}".format(current_thread().name, s)),
               on_error=lambda e: print(e),
               on_completed=lambda: print("PROCESS 1 done!"))

# Create Process 2
Observable.range(1, 10) \
    .map(lambda s: intense_calculation(s)) \
    .subscribe_on(pool_scheduler) \
    .subscribe_(on_next=lambda i: print("PROCESS 2: {0} {1}".format(current_thread().name, i)),
               on_error=lambda e: print(e), on_completed=lambda: print("PROCESS 2 done!"))

# Create Process 3, which is infinite
Observable.interval(1000) \
    .map(lambda i: i * 100) \
    .observe_on(pool_scheduler) \
    .map(lambda s: intense_calculation(s)) \
    .subscribe_(on_next=lambda i: print("PROCESS 3: {0} {1}".format(current_thread().name, i)),
               on_error=lambda e: print(e))

input("Press any key to exit\n")

出力:

Press any key to exit
PROCESS 1: Thread-1 Alpha
PROCESS 2: Thread-2 1
PROCESS 3: Thread-4 0
PROCESS 2: Thread-2 2
PROCESS 1: Thread-1 Beta
PROCESS 3: Thread-7 100
PROCESS 3: Thread-7 200
PROCESS 2: Thread-2 3
PROCESS 1: Thread-1 Gamma
PROCESS 1: Thread-1 Delta
PROCESS 2: Thread-2 4
PROCESS 3: Thread-7 300
...

より詳細なチュートリアルについては、「 Reactive Python for Data Science」を参照してください。これは、 O’ReillyストアO’Reilly Safariの両方で利用できます。

Pythonの配置

Disposablesはコンテキストマネージャを実装しているためwithステートメントwith使用することができます。

観測可能なシーケンスは、 +を使って連結することができるので、以下のように書くことができます:

xs = Observable.of(1,2,3)
ys = Observable.of(4,5,6)
zs = xs + ys  # Concatenate observables

観測可能なシーケンスは*=を使用して繰り返すことができるので、次のように書くことができます:

xs = Observable.of(1,2,3)
ys = xs * 4

観察可能なシーケンスは、 [start:stop:step]を使用してスライスすることができるので、次のように書くことができます。

xs = Observable.of(1,2,3,4,5,6)
ys = xs[1:-1]

観測可能なシーケンスはイテレータに変換されるので、ジェネレータ式を使用したり、反復処理(キューイングとブロッキングを使用)することができます。

xs = Observable.of(1,2,3,4,5,6)
ys = xs.to_blocking()
zs = (x*x for x in ys if x > 3)
for x in zs:
    print(x)

.NETとRxJSとの違い

RxPYは、 Rx v2.2を実装しています。これには、 134以上のクエリ演算子1100以上の渡し単位テストがあります。 RxPYはほとんどの場合RxJSの直接ポートですが、RxNETとRxJavaからスレッドとブロックのオペレータに関しても少し借りています。

RxPYはPEP 8に従っているので、すべての関数とメソッド名は小文字で、可読性を向上させるために必要に応じてアンダースコアで区切られた単語を使用します。

したがって、次のような.NETコード: c# var group = source.GroupBy(i => i % 3);

Pythonで_をつけて書く必要があります: python group = source.group_by(lambda i: i % 3)

RxPYでは、オペレータが複数のオプションの引数を持つ場合、位置引数の代わりに名前付きキーワード引数を使用する必要があります。 RxPYは、あなたがオペレータに与えている(またはしていない)引数を検出しようとしません。

res = Observable.timer(5000) # Yes
res = Observable.timer(5000, 1000) # Yes
res = Observable.timer(5000, 1000, Scheduler.timeout) # Yes
res = Observable.timer(5000, scheduler=Scheduler.timeout) # Yes, but must name

res = Observable.timer(5000, Scheduler.timeout) # No, this is an error

したがって、 Observable.timerようObservable.timer演算子に複数のオプションの引数がある場合は、引数に名前を付ける必要があります。 少なくともオプションはオプションとマークされています。

スケジューラ

RxPYでは、完全に非同期に実行するか、スレッドを使用して作業やタイムアウトをスケジュールするかを決めることができます。

時間とスケジューラのハンドリングには、絶対時間値のdatetimeと相対時間のtimedeltaを指定する必要があります。 ミリ秒を表すにはintを使用することもできます。

RxPYにはバッテリーが付属しています。お気に入りのPythonフレームワークでRxPYを使いやすくするために、いくつかのPython固有のメインループスケジューラーがあります。

  • ThreadPoolSchedulerを使用してスケジューラの固定サイズのプールを作成します。
  • NewThreadSchedulerを使用してサブスクリプションごとに新しいスレッドを作成する
  • AsyncIOSchedulerで使用するAsyncIOScheduler (Python 3.4またはtrollius 、Python 2.6-3.5と互換性のあるasyncioポートが必要です)。
  • EventLetEventSchedulerで使用するEventLetEventScheduler
  • IOLoopSchedulerで使用するためのIOLoopScheduler あなたのTornadoアプリケーションでRxPYを使用する方法については、 オートコンプリートkonamicodeの例を参照してください。
  • GEventSchedulerで使用するGEventScheduler (Python 2.7のみ)。
  • TwistedSchedulerTwistedで使用するために使用します。
  • Tkinterで使用するためのTkinterScheduler TkinterアプリケーションでRxPYを使用する方法については、 タイムフライの例を参照してください。
  • PyGameSchedulerで使用するためのPyGameScheduler PyGameアプリケーションでRxPYを使用する方法については、 チェスの例を参照してください。
  • PyQt4PyQt5 、およびPySideで使用するためのQtScheduler QtアプリケーションでRxPYを使用する方法については、 タイムフライの例を参照してください。
  • Python GTK + 3で使用するGtkScheduler GTK +アプリケーションでRxPYを使用する方法については、 タイムフライの例を参照してください。
  • wxPythonで使用するWxScheduler wxアプリケーションでRxPYを使用する方法については、 タイムフライの例を参照してください。

貢献する

コードチェックインのフィードバックを確認して送信したり、実装されている新しい機能を提案して試したり、問題を登録したり、修正が確定したことを確認したり、自分自身のコード修正やコード投稿を提出することで貢献できます。

メインリポジトリはReactiveX / RxPYです。 現在のところ、 Reactive-Extensions / RxPyCodePlexには旧式のミラーがあります。 ReactiveX / RxPY / issuesに問題を登録してください。

ReactiveX / RxPYの マスターブランチに対してプルリクエストを提出してください。

ライセンス

Copyright(c)Microsoft Open Technologies、Inc.すべての権利は留保されています。 Microsoft Open Technologiesは寄稿者に感謝したいと思います。その一覧はhttp://rx.codeplex.com/wikipage?title=Contributorsにあります。

Apache License、Version 2.0(以下「ライセンス」)の下でライセンスされています。 ライセンスに従わない限り、このファイルを使用することはできません。 あなたはライセンスのコピーを

http://www.apache.org/licenses/LICENSE-2.0

適用法または書面による合意が必要な場合を除き、本ライセンスに基づいて配布されるソフトウェアは、明示的または黙示的にいかなる種類の保証または条件もなく「現状有姿」で配布されます。 ライセンスに基づいて許可および制限を規定する特定の言語については、ライセンスを参照してください。







-ReactiveX
-, , , , ,

執筆者: