, Observer.create에 다른 매개 변수를 전달하려면 어떻게해야합니까?

Question

설명을 위해 RxPY 사용.

함수에서 관측 가능 함수를 만들고 싶지만이 함수는 매개 변수를 가져야합니다. 이 특정 예제는 임의의 간격으로, 내가 보내고 자하는 미리 정의 된 많은 시세 중 하나를 반환해야합니다. 지금까지의 해결 방법은 클로저를 사용하는 것입니다.

from __future__ import print_function 

from rx import Observable 
import random 
import string 
import time 

def make_tickers(n = 300, s = 123): 
    """ generates up to n unique 3-letter strings geach makde up of uppsercase letters""" 
    random.seed(s) 
    tickers = [''.join(random.choice(string.ascii_uppercase) for _ in range(3)) for y in range(n)] 
    tickers = list(set(tickers)) # unique 
    print(len(tickers)) 
    return(tickers) 

def spawn_prices_fn(tickers): 
    """ returns a function that will return a random element 
    out of tickers every 20-100 ms, and takes an observable parameter """ 

    def spawner(observer): 
     while True: 
      next_tick = random.choice(tickers) 
      observer.on_next(next_tick) 
      time.sleep(random.randint(20, 100)/1000.0) 

    return(spawner) 


if __name__ == "__main__": 
    spawned = spawn_prices_fn(make_tickers()) 
    xx = Observable.create(spawned) 
    xx.subscribe(lambda s: print(s)) 

더 간단한 방법이 있습니까? 클로저가 필요없는 Observable.create의 첫 번째 매개 변수 함수에 추가 매개 변수를 보낼 수 있습니까? 정식 조언은 무엇입니까?

Answer 1

여러 가지 방법으로 수행 할 수 있습니다. 코드를 너무 많이 변경하지 않는 솔루션 중 하나입니다. 세대를 시세 참고도 나는 또한 약간 만들기 위해 코드를 조정 rx -like

더 할 몇 가지 rx 마법과 결합 된 하나의 문자열을 생성하는 기능으로 분할 할 수 flake8 행복

from __future__ import print_function 

import random 
import string 
import time 

from rx import Observable 


def make_tickers(n=300, s=123): 
    """ 
    Generates up to n unique 3-letter strings each made up of uppercase letters 
    """ 
    random.seed(s) 
    tickers = [''.join(random.choice(string.ascii_uppercase) for _ in range(3)) 
       for y in range(n)] 
    tickers = list(set(tickers)) # unique 
    print(len(tickers)) 
    return(tickers) 


def random_picker(tickers): 
    ticker = random.choice(tickers) 
    time.sleep(random.randint(20, 100)/1000.0) 
    return ticker 


if __name__ == "__main__": 
    xx = Observable\ 
     .repeat(make_tickers())\ 
     .map(random_picker)\ 
     .subscribe(lambda s: print(s)) 

또는 make_tickers없이 솔루션 :

from __future__ import print_function 

import random 
import string 
import time 

from rx import Observable 


def random_picker(tickers): 
    ticker = random.choice(tickers) 
    time.sleep(random.randint(20, 100)/1000.0) 
    return ticker 


if __name__ == "__main__": 
    random.seed(123) 
    Observable.range(1, 300)\ 
     .map(lambda _: ''.join(random.choice(string.ascii_uppercase) 
           for _ in range(3)))\ 
     .reduce(lambda x, y: x + [y], [])\ 
     .do_while(lambda _: True)\ 
     .map(random_picker)\ 
     .subscribe(lambda s: print(s)) 

time.sleep는에서 멀리 이동 될 수 있습니다 0이지만 코드가 조금 복잡해집니다