sha*_*yan 5 python reactive-programming py-shiny
我正在开发一个基于 python 的闪亮应用程序，用于通过串行传输线驱动流体泵。配置具有设定幅度和运行时间的流量曲线后，可以通过按操作按钮“p1”启动串行传输。
我面临的问题是与按钮“p1”相关的内部缺乏反应性reactive.event(input.p1)：我想确保通过单击“p1”开始传输，从而触发可以reactive.event(input.p1)随时通过单击“p2”终止传输。
然而，当前的实现导致停止消息 inreactive.event(input.p2)排队并在传输reactive.event(input.p1)结束后发送。
我怎样才能解决这个问题？有什么方法可以确保我的程序仍然对其他输入做出反应？我希望单击“p2”后立即停止传输。两个按钮的实现都放在下面。
@reactive.Effect
@reactive.event(input.p1)
def _():
y = yg.get() # fetch np.array y from reactive value yg, numbers in y correspond to driving voltages
for e in y: # iterate over array
msg = "1:1:"+str(e)+":100" # formatted string containing the driving voltage
#print(msg) # print to console in debug mode
ser.write(bytes(msg,'utf-8')) # send the formatted string
t0 = time.time() # time stamp
while(((time.time()-t0)<=2)): # next driving voltage should be transmitted after 2 seconds
pass
ser.write(bytes("0:1",'utf-8')) # stops the pump after transmission has ended
@reactive.Effect
@reactive.event(input.p2)
def _():
#print("1:0")
ser.write(bytes("0:1",'utf-8')) # Stop the pump
好吧，我通过线程创建一个计时器来解决这个问题。就我个人而言，我认为答案是，就其本身而言，不可能在循环内保留反应性。我尝试了各种实现，包括@phili_b 建议的实现。但没有什么真正迫使程序退出 while 循环，该循环是通过单击按钮 p2 在与按钮 p1 相关的反应事件中调用的。
这是对我有用的解决方案：
# This function may very well be included into the one below,
# but in this case this structure serves me well
# function takes a number, forms the message string and puts it on the serial port
def transmit(e):
msg = "1:1:"+str(e)+":100"
#print(msg)
ser.write(bytes(msg,'utf-8'))
# This is the main function that is being threaded. Loop iterates over array y
# every 2 seconds until either end of y or the threading Event sflag is triggered
def rtimer(y,sflag): # Timer calls this function
i = 0
while i<np.size(y) and not sflag.is_set():
transmit(y[i])
i+=1
time.sleep(2) # 2 second interval between transmissions
# p1-button calls this function
@reactive.Effect()
@reactive.event(input.p1)
def _():
y = yg.get()
sflag.clear() # clear slfag in case it has been triggered prior
timer_thread = th.Thread(target=rtimer,args=[y,sflag]) # threading is imported as th
timer_thread.start()
# p2-button calls this function
@reactive.Effect()
@reactive.event(input.p2)
def stop():
#print("1:0")
sflag.set()
ser.write(bytes("1:0",'utf-8'))
我知道这是一种解决方法，而不是对我在 OP 中建议的实现的修复。尽管代码现在可以实现我想要的功能，但我仍然很好奇是否有办法使原始代码能够工作。我很想知道本质上是否有可能处理循环似乎对程序造成的限制。
根据请求添加了完整代码：请注意，按钮位于左下角。GUI 的状态非常粗糙，看起来很丑陋。这仅用于复制目的。
from shiny import App, render, ui, reactive
import serial
import time
import matplotlib.pyplot as plt
import numpy as np
import shinyswatch
import threading as th
import asyncio
#import reprex
# class RepeatTimer(th.Timer):
# def run(self):
# while not self.finished.wait(self.interval):
# self.function(*self.args,**self.kwargs)
#x = np.empty(1)
#y = np.empty(1)
#
# Pump PowerMode Amplitude Frequenz Amplitude2
# Pump
# 1 Erste Pumpe
# 2 Zweite Pumpe
# 6 Einschalten beider Pumpen
# 9 Ausschalten beider Pumpen
# powerMode
# 0 Ausschalten der Pumpe
# 1 Einschalten der Pumpe
# Amplitude/Amplitude2
# Ganzzahl zwischen 0-250 Amplitudenwert in Vpp
# Frequenz
# 100 Festgelegte Frequenz von 100Hz
# Format
# 1 : 1 : 250 : 100 : 150
# Pump : PowerMode : Amplitude : Frequenz : Amplitude2
ser = serial.Serial("COM6",115200)
#print(ser.name)
app_ui = ui.page_fluid(
shinyswatch.theme.minty(),
ui.row(
ui.column(8,
ui.h2("Header 1"),
ui.row(
ui.column(2,
ui.h5("Profile"),
ui.input_radio_buttons("M1","",{"A":"Constant","B":"Linear","C":"Periodic"})
),
ui.column(2,ui.h5("Parameter"),
ui.panel_conditional("input.M1 === 'A'",
ui.input_numeric("AK","Amplitude [V]",value=100),ui.input_numeric("TK","Runtime [s]",value=10)),
ui.panel_conditional("input.M1 === 'B'",
ui.input_switch("lin","Rising",True),ui.input_numeric("SL","Start-Amplitude [V]",value=50,min=10,max=250),ui.input_numeric("EL","End-Amplitude [V]",value=100,min=10,max=250),ui.input_numeric("TL","Runtime [s]",value=10,min=1)),
ui.panel_conditional("input.M1 === 'C'",
ui.input_switch("per","Sine",True),ui.input_numeric("PMIN","Min-Amplitude [V]",value=50,min=10,max=250),ui.input_numeric("PMAX","Max-Amplitude [V]",value=100,min=10,max=250),ui.input_numeric("TP","Runtime [s]",value=10,min=1))
),
ui.column(8,ui.output_plot("preview1")
)
)),
ui.column(4,
ui.h2("test"),
ui.input_slider("n", "N", 20, 60, 20),
ui.output_text_verbatim("txt"),
ui.input_action_button("b1","Plot", class_="btn-success"),
ui.output_plot("pl"),
ui.output_text_verbatim("t2"),
ui.input_action_button("b2","Pumpe Start"),
ui.input_action_button("b3","Pumpe Stopp"),
)
),
ui.row(
ui.input_radio_buttons("PA","",{"Pu1":"Pumpe 1","Pu2":"Pumpe 2","Pu3":"Pumpe 3","Pu4":"Pumpe 4"})
,
ui.column(3,
ui.input_action_button("p1","Pumpe Start"),
ui.input_action_button("p2","Pumpe Stopp"),
ui.input_action_button("t","Test")
)
)
)
def server(input, output, session):
status = reactive.Value("Pumpe aus")
xg = reactive.Value()
yg = reactive.Value()
#yg1 =
sflag = th.Event()
def transmit(e):
msg = "1:1:"+str(e)+":100"
#print(msg)
ser.write(bytes(msg,'utf-8'))
def rtimer(y,sflag):
i = 0
while i<np.size(y) and not sflag.is_set():
transmit(y[i])
i+=1
time.sleep(2)
@reactive.Effect()
@reactive.event(input.p1)
def _():
y = yg.get()
sflag.clear()
timer_thread = th.Thread(target=rtimer,args=[y,sflag])
timer_thread.start()
# for i in range(np.size(y)):
# t1 = th.Timer(2,transmit,args = [i,y])
# t1.start()
#timer = RepeatTimer(2,transmit,[1,y])
#timer.start()
# for e in range(0,20):
# test(i,y)
# x = xg.get()
# y = yg.get()
# ind = 0
#for e in y:
# task = asyncio.create_task(st())
# msg = "1:1:"+str(e)+":100"
# print(msg)
# # ser.write(bytes(msg,'utf-8'))
# await asyncio.sleep(2)
@reactive.Effect
@reactive.event(input.t)
def _():
msg = "1:1:"+str(80)+":100"
print(msg)
#ser.write(bytes(msg,'utf-8'))
#@reactive.event(input.t)
#def _():
@reactive.Effect()
@reactive.event(input.p2)
def stop():
#print("1:0")
sflag.set()
ser.write(bytes("1:0",'utf-8'))
@reactive.Effect
@reactive.event(input.per)
def _():
if(input.per()):
ui.update_switch(
"per",label="Sine"
)
else:
ui.update_switch(
"per",label="Cosine"
)
@reactive.Effect
@reactive.event(input.lin)
def _():
if(input.lin()):
ui.update_switch(
"lin",label="Rising"
)
else:
ui.update_switch(
"lin",label="Falling"
)
@output
@render.text
def txt():
#ser.write(b"input.n()")
#time.sleep(5)
return f"{input.n()} V pro Zeitschritt"
@output
@render.plot
@reactive.event(input.b1,ignore_none=False)
def pl():
px = np.arange(0,100,1)
py = np.arange(100,200,1)
fig, ax = plt.subplots()
ax.plot(px,py)
plt.xlabel("Zeit [s]")
plt.ylabel("Amplitude [V]")
return fig
@reactive.Effect
@reactive.event(input.b2)
def _():
#ser.write(b"1:1:150:100")
#ser.write(b"1:2")
status.set("P1 On")
@reactive.Effect
@reactive.event(input.b3)
def _():
#ser.write(b"0:1")
print("0:1")
status.set("P1 Off")
@output
@render.text
def t2():
return str(status())
@output
@render.plot
@reactive.event(input.AK,input.TK,input.SL,input.EL,input.TL,input.PMIN,input.PMAX,input.TP,input.M1,input.per)
def preview1():
match input.M1():
case "A":
x = np.arange(0,input.TK()+2,2)
y = np.ones(np.size(x))*input.AK()
fig, ax = plt.subplots()
ax.plot(x,y)
plt.title("Constant Profile")
plt.xlabel("Time [s]")
plt.ylabel("Amplitude [V]")
xg.set(np.rint(x).astype(int))
yg.set(np.rint(y).astype(int))
return fig
case "B":
x = np.arange(0,input.TL()+2,2)
#if input.
y = x*((input.EL()-input.SL())/input.TL()) + input.SL()
fig, ax = plt.subplots()
ax.plot(x,y)
plt.title("Linear Profile")
plt.xlabel("Time [s]")
plt.ylabel("Amplitude [V]")
xg.set(np.rint(x).astype(int))
yg.set(np.rint(y).astype(int))
return fig
case "C":
x = np.arange(0,input.TP(),0.1)
if(input.per()):
y = (0.5*(input.PMAX()-input.PMIN())) * np.sin(x) + (np.mean([input.PMAX(),input.PMIN()]))
else:
y = (0.5*(input.PMAX()-input.PMIN())) * np.cos(x) + (np.mean([input.PMAX(),input.PMIN()]))
fig, ax = plt.subplots()
ax.plot(x,y)
plt.title("Periodic Profile")
plt.xlabel("Time [s]")
plt.ylabel("Amplitude [V]")
xg.set(np.rint(x).astype(int))
yg.set(np.rint(y).astype(int))
return fig
app = App(app_ui, server)
| 归档时间： |
|
| 查看次数： |
191 次 |
| 最近记录： |