rx-py

0熱度

1回答

我正在嘗試創建一個分析股票價格的工具。我得到了不同股票的價格數據流，我希望有一個可觀察的事件，當它收到一個新的，獨特的和完整的價格集合時發出事件。我的計劃：將流分組爲不同股票的不同子流，並重新組合它們的最新值。比方說，我有一個事件流是這樣的： from rx import Observable stock_events = [ {'stock': 'A', 'price': 1

0熱度

1回答

RxPY - 如何使用stop_and_wait？

我試圖在Python翻譯下面的JavaScript代碼示例： import Rx from "rx" let source = Rx.Observable.interval(1000) .timestamp() .controlled(); source.stopAndWait().subscribe( (result) => console.log("onNe

0熱度

1回答

配料結果可觀察

說我有一個觀察到的，看起來像這樣（這是Python，但應該是通用於所有語言）： rx.Observable.from_iterable([[1],[],[2],[3],[],[],[4],[5,6],[7],[8,9],[10]]) 我希望最終能批量整數到長度爲5的列表和能夠傳遞到那些功能，所以這樣的事情： batch_function([1,2,3,4,5]) batch_function([

0熱度

1回答

RxPy將值傳遞給觀察者

有沒有一種方法可以根據用戶輸入將值傳遞給觀察者（這意味着所傳遞的值並非始終是固定的）？ from rx import Observable, Observer def push_five_strings(observer,value): observer.on_next(value) #observer.on_next("Alpha") observer.o

1熱度

1回答

RxPy - 將Live Twitter流轉爲Rx Observable？

我沿着這個great tutorial跟着使用tweepy在Python中利用實時Twitter流。這將在提供RxJava，RxPy，RxScala或ReactiveX的現場時間內打印Tweets。 from tweepy.streaming import StreamListener from tweepy import OAuthHandler from tweepy import Str

0熱度

2回答

總結反應pyx中的可觀測數據

在Python中使用ReactiveX，我該如何總結一個Observable流？我有一個{「user」：「...」，「date」：...}字典流。我想創建一個函數，我可以應用這個函數，爲每個用戶累積最新日期的字典，然後在流結束時發出累積的觀察值（這就像max，但必須查看用戶字段，並且會發出多個值）。示例 - 輸入流： { "user": "a", "date": "2017-02-14" }

1熱度

1回答

通過rxpy從可觀察值返回值

將rx.Observable對象轉換爲函數中的「普通」對象的優雅方式是什麼？如： def foo(): return rx.Observable.just('value').subscribe(<some magic here>) >>> print(foo()) # expected: # value # however get: # <rx.disposables.a

0熱度

1回答

當一個對象的任何屬性隨RxPy變化時觸發一個函數

在RxPy中，在.NET框架here中有沒有類似於INotifyPropertyChanged？我試圖添加一個觀察者到一個對象，以便該對象的任何屬性改變，一個函數將被調用。

0熱度

1回答

rxpy將物品注入可觀察區域

此問題與有關。我想構建一個處理來自可觀察源消息的反應系統。除此之外，我正試圖將它與基於zookeeper的領導者選舉系統相結合。這種組合只允許一個進程場中的一個領導者處理消息流。以下是我正在嘗試構建的代碼的要點。 # event_source is an observable of messages # manager.leaders is an observable of leader

0熱度

1回答

從RxPy可見性更新pyqtgraph圖

在以下腳本中，創建了一個繪圖窗口並將values正確傳遞給plot.update。但情節並未更新。我究竟做錯了什麼？ import sys import time import numpy from numpy import pi import rx from rx.concurrency import QtScheduler import pyqtgraph as pg fr