M-Labs/thermostat
9
0
Fork 5
Code Issues 9 Pull Requests 5 Releases Wiki Activity

PyThermostat: Add entry points to runnable scripts #149

Merged
sb10q merged 2 commits from atse/thermostat:pytec-entry-points into master 2024-11-25 10:56:45 +08:00
Conversation 1 Commits 2 Files Changed 2 +123 -108
2 changed files with 123 additions and 108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

218
pythermostat/pythermostat/plot.py
View File

@ -5,127 +5,133 @@ import matplotlib.animation as animation
from threading import Thread, Lock from threading import Thread, Lock
from pythermostat.client import Client from pythermostat.client import Client
TIME_WINDOW = 300.0
tec = Client() def main():
target_temperature = tec.get_pid()[0]['target'] TIME_WINDOW = 300.0
print("Channel 0 target temperature: {:.3f}".format(target_temperature))
class Series: tec = Client()
def __init__(self, conv=lambda x: x): target_temperature = tec.get_pid()[0]['target']
self.conv = conv print("Channel 0 target temperature: {:.3f}".format(target_temperature))
self.x_data = []
self.y_data = []
def append(self, x, y): class Series:
self.x_data.append(x) def __init__(self, conv=lambda x: x):
self.y_data.append(self.conv(y)) self.conv = conv
self.x_data = []
self.y_data = []
def clip(self, min_x): def append(self, x, y):
drop = 0 self.x_data.append(x)
while drop < len(self.x_data) and self.x_data[drop] < min_x: self.y_data.append(self.conv(y))
drop += 1
self.x_data = self.x_data[drop:]
self.y_data = self.y_data[drop:]
series = { def clip(self, min_x):
# 'adc': Series(), drop = 0
# 'sens': Series(lambda x: x * 0.0001), while drop < len(self.x_data) and self.x_data[drop] < min_x:
'temperature': Series(), drop += 1
# 'i_set': Series(), self.x_data = self.x_data[drop:]
'pid_output': Series(), self.y_data = self.y_data[drop:]
# 'vref': Series(),
# 'dac_value': Series(),
# 'dac_feedback': Series(),
# 'i_tec': Series(),
'tec_i': Series(),
'tec_u_meas': Series(),
# 'interval': Series(),
}
series_lock = Lock()
quit = False series = {
# 'adc': Series(),
# 'sens': Series(lambda x: x * 0.0001),
'temperature': Series(),
# 'i_set': Series(),
'pid_output': Series(),
# 'vref': Series(),
# 'dac_value': Series(),
# 'dac_feedback': Series(),
# 'i_tec': Series(),
'tec_i': Series(),
'tec_u_meas': Series(),
# 'interval': Series(),
}
series_lock = Lock()
quit = False
def recv_data(tec):
global last_packet_time
while True:
data = tec.get_report()
ch0 = data[0]
series_lock.acquire()
try:
for k, s in series.items():
if k in ch0:
v = ch0[k]
if type(v) is float:
s.append(ch0['time'], v)
finally:
series_lock.release()
if quit:
break
time.sleep(0.05)
thread = Thread(target=recv_data, args=(tec,))
thread.start()
fig, ax = plt.subplots()
for k, s in series.items():
s.plot, = ax.plot([], [], label=k)
legend = ax.legend()
def animate(i):
min_x, max_x, min_y, max_y = None, None, None, None
def recv_data(tec):
global last_packet_time
while True:
data = tec.get_report()
ch0 = data[0]
series_lock.acquire() series_lock.acquire()
try: try:
for k, s in series.items(): for k, s in series.items():
if k in ch0: s.plot.set_data(s.x_data, s.y_data)
v = ch0[k] if len(s.y_data) > 0:
if type(v) is float: s.plot.set_label("{}: {:.3f}".format(k, s.y_data[-1]))
s.append(ch0['time'], v)
if len(s.x_data) > 0:
min_x_ = min(s.x_data)
if min_x is None:
min_x = min_x_
else:
min_x = min(min_x, min_x_)
max_x_ = max(s.x_data)
if max_x is None:
max_x = max_x_
else:
max_x = max(max_x, max_x_)
if len(s.y_data) > 0:
min_y_ = min(s.y_data)
if min_y is None:
min_y = min_y_
else:
min_y = min(min_y, min_y_)
max_y_ = max(s.y_data)
if max_y is None:
max_y = max_y_
else:
max_y = max(max_y, max_y_)
if min_x and max_x - TIME_WINDOW > min_x:
for s in series.values():
s.clip(max_x - TIME_WINDOW)
finally: finally:
series_lock.release() series_lock.release()
if quit: if min_x != max_x:
break ax.set_xlim(min_x, max_x)
time.sleep(0.05) if min_y != max_y:
margin_y = 0.01 * (max_y - min_y)
ax.set_ylim(min_y - margin_y, max_y + margin_y)
thread = Thread(target=recv_data, args=(tec,)) nonlocal legend
thread.start() legend.remove()
legend = ax.legend()
fig, ax = plt.subplots() ani = animation.FuncAnimation(
fig, animate, interval=1, blit=False, save_count=50)
for k, s in series.items(): plt.show()
s.plot, = ax.plot([], [], label=k) quit = True
legend = ax.legend() thread.join()
def animate(i):
min_x, max_x, min_y, max_y = None, None, None, None
series_lock.acquire() if __name__ == "__main__":
try: main()
for k, s in series.items():
s.plot.set_data(s.x_data, s.y_data)
if len(s.y_data) > 0:
s.plot.set_label("{}: {:.3f}".format(k, s.y_data[-1]))
if len(s.x_data) > 0:
min_x_ = min(s.x_data)
if min_x is None:
min_x = min_x_
else:
min_x = min(min_x, min_x_)
max_x_ = max(s.x_data)
if max_x is None:
max_x = max_x_
else:
max_x = max(max_x, max_x_)
if len(s.y_data) > 0:
min_y_ = min(s.y_data)
if min_y is None:
min_y = min_y_
else:
min_y = min(min_y, min_y_)
max_y_ = max(s.y_data)
if max_y is None:
max_y = max_y_
else:
max_y = max(max_y, max_y_)
if min_x and max_x - TIME_WINDOW > min_x:
for s in series.values():
s.clip(max_x - TIME_WINDOW)
finally:
series_lock.release()
if min_x != max_x:
ax.set_xlim(min_x, max_x)
if min_y != max_y:
margin_y = 0.01 * (max_y - min_y)
ax.set_ylim(min_y - margin_y, max_y + margin_y)
global legend
legend.remove()
legend = ax.legend()
ani = animation.FuncAnimation(
fig, animate, interval=1, blit=False, save_count=50)
plt.show()
quit = True
thread.join()

9
pythermostat/setup.py
View File

@ -9,4 +9,13 @@ setup(
license="GPLv3", license="GPLv3",
install_requires=["setuptools"], install_requires=["setuptools"],
packages=find_packages(), packages=find_packages(),
entry_points={
"gui_scripts": [
"thermostat_plot = pythermostat.plot:main",
],
"console_scripts": [
"thermostat_autotune = pythermostat.autotune:main",
"thermostat_test = pythermostat.test:main",
]
},
) )