将 RxPy 从 1.x 升级到 3.x 后订阅不起作用
subscribe not work after upgrading RxPy from 1.x to 3.x
我正在使用 Python 3.7.3.
我尝试将 RxPy 从 1.6.1 (1.x) 升级到 3.0.0a3 (3.x)。
旧代码使用 RxPy 1.x
from rx import Observable
import psutil
import numpy as np
import pylab as plt
cpu_data = (Observable
.interval(100) # Each 100 milliseconds
.map(lambda x: psutil.cpu_percent())
.publish())
cpu_data.connect()
def monitor_cpu(npoints):
lines, = plt.plot([], [])
plt.xlim(0, npoints)
plt.ylim(0, 100)
cpu_data_window = cpu_data.buffer_with_count(npoints, 1)
def update_plot(cpu_readings):
lines.set_xdata(np.arange(len(cpu_readings)))
lines.set_ydata(np.array(cpu_readings))
plt.draw()
alertpoints = 4
high_cpu = (cpu_data
.buffer_with_count(alertpoints, 1)
.map(lambda readings: all(r > 20 for r in readings)))
label = plt.text(1, 1, "normal")
def update_warning(is_high):
if is_high:
label.set_text("high")
else:
label.set_text("normal")
high_cpu.subscribe(update_warning)
cpu_data_window.subscribe(update_plot)
plt.show()
if __name__ == '__main__':
monitor_cpu(10)
如果你运行代码你可以看到实时CPU监控图表。
然而,在我通过
安装了新的 RxPy 之后
pip3 install --pre rx
下面的新代码,只显示白色,没有任何动态图表。
而函数update_plot
实际上从来没有运行。有什么想法吗?
使用 RxPy 的新代码3.x
from rx import interval, operators as op
import psutil
import numpy as np
import pylab as plt
cpu_data = interval(100).pipe( # Each 100 milliseconds
op.map(lambda x: psutil.cpu_percent()),
op.publish())
cpu_data.connect()
def monitor_cpu(npoints):
lines, = plt.plot([], [])
plt.xlim(0, npoints)
plt.ylim(0, 100)
cpu_data_window = cpu_data.pipe(
op.buffer_with_count(npoints, 1))
def update_plot(cpu_readings):
print('update') # here never runs
lines.set_xdata(np.arange(len(cpu_readings)))
lines.set_ydata(np.array(cpu_readings))
plt.draw()
alertpoints = 4
high_cpu = cpu_data.pipe(
op.buffer_with_count(alertpoints, 1),
op.map(lambda readings: all(r > 20 for r in readings)))
label = plt.text(1, 1, "normal")
def update_warning(is_high):
if is_high:
label.set_text("high")
else:
label.set_text("normal")
high_cpu.subscribe(update_warning)
cpu_data_window.subscribe(update_plot)
plt.show()
if __name__ == '__main__':
monitor_cpu(10)
时间单位现在以秒为单位
cpu_data = interval(0.1).pipe( # Each 100 milliseconds
op.map(lambda x: psutil.cpu_percent()),
op.publish())
cpu_data.connect()
我正在使用 Python 3.7.3.
我尝试将 RxPy 从 1.6.1 (1.x) 升级到 3.0.0a3 (3.x)。
旧代码使用 RxPy 1.x
from rx import Observable
import psutil
import numpy as np
import pylab as plt
cpu_data = (Observable
.interval(100) # Each 100 milliseconds
.map(lambda x: psutil.cpu_percent())
.publish())
cpu_data.connect()
def monitor_cpu(npoints):
lines, = plt.plot([], [])
plt.xlim(0, npoints)
plt.ylim(0, 100)
cpu_data_window = cpu_data.buffer_with_count(npoints, 1)
def update_plot(cpu_readings):
lines.set_xdata(np.arange(len(cpu_readings)))
lines.set_ydata(np.array(cpu_readings))
plt.draw()
alertpoints = 4
high_cpu = (cpu_data
.buffer_with_count(alertpoints, 1)
.map(lambda readings: all(r > 20 for r in readings)))
label = plt.text(1, 1, "normal")
def update_warning(is_high):
if is_high:
label.set_text("high")
else:
label.set_text("normal")
high_cpu.subscribe(update_warning)
cpu_data_window.subscribe(update_plot)
plt.show()
if __name__ == '__main__':
monitor_cpu(10)
如果你运行代码你可以看到实时CPU监控图表。
然而,在我通过
安装了新的 RxPy 之后pip3 install --pre rx
下面的新代码,只显示白色,没有任何动态图表。
而函数update_plot
实际上从来没有运行。有什么想法吗?
使用 RxPy 的新代码3.x
from rx import interval, operators as op
import psutil
import numpy as np
import pylab as plt
cpu_data = interval(100).pipe( # Each 100 milliseconds
op.map(lambda x: psutil.cpu_percent()),
op.publish())
cpu_data.connect()
def monitor_cpu(npoints):
lines, = plt.plot([], [])
plt.xlim(0, npoints)
plt.ylim(0, 100)
cpu_data_window = cpu_data.pipe(
op.buffer_with_count(npoints, 1))
def update_plot(cpu_readings):
print('update') # here never runs
lines.set_xdata(np.arange(len(cpu_readings)))
lines.set_ydata(np.array(cpu_readings))
plt.draw()
alertpoints = 4
high_cpu = cpu_data.pipe(
op.buffer_with_count(alertpoints, 1),
op.map(lambda readings: all(r > 20 for r in readings)))
label = plt.text(1, 1, "normal")
def update_warning(is_high):
if is_high:
label.set_text("high")
else:
label.set_text("normal")
high_cpu.subscribe(update_warning)
cpu_data_window.subscribe(update_plot)
plt.show()
if __name__ == '__main__':
monitor_cpu(10)
时间单位现在以秒为单位
cpu_data = interval(0.1).pipe( # Each 100 milliseconds
op.map(lambda x: psutil.cpu_percent()),
op.publish())
cpu_data.connect()