atse/thermostat
1
0
Fork 0
forked from M-Labs/thermostat
Code Issues Pull Requests Activity

Formating with vscode black formatter

Browse Source 
Invoked with Ctrl+Shift+I
This commit is contained in:
atse 2024-10-28 18:05:25 +08:00
parent 4287cba96c
commit 23cd1a5d25
5 changed files with 83 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
pytec/.flake8 Normal file
View File

@ -0,0 +1,3 @@
[flake8]
max-line-length = 88
extend-ignore = E203,E701

94
pytec/autotune.py
View File

@ -12,15 +12,15 @@ from pytec.client import Client
class PIDAutotuneState(Enum): class PIDAutotuneState(Enum):
STATE_OFF = 'off' STATE_OFF = "off"
STATE_RELAY_STEP_UP = 'relay step up' STATE_RELAY_STEP_UP = "relay step up"
STATE_RELAY_STEP_DOWN = 'relay step down' STATE_RELAY_STEP_DOWN = "relay step down"
STATE_SUCCEEDED = 'succeeded' STATE_SUCCEEDED = "succeeded"
STATE_FAILED = 'failed' STATE_FAILED = "failed"
class PIDAutotune: class PIDAutotune:
PIDParams = namedtuple('PIDParams', ['Kp', 'Ki', 'Kd']) PIDParams = namedtuple("PIDParams", ["Kp", "Ki", "Kd"])
PEAK_AMPLITUDE_TOLERANCE = 0.05 PEAK_AMPLITUDE_TOLERANCE = 0.05
@ -30,13 +30,14 @@ class PIDAutotune:
"ciancone-marlin": [0.303, 0.1364, 0.0481], "ciancone-marlin": [0.303, 0.1364, 0.0481],
"pessen-integral": [0.7, 1.75, 0.105], "pessen-integral": [0.7, 1.75, 0.105],
"some-overshoot": [0.333, 0.667, 0.111], "some-overshoot": [0.333, 0.667, 0.111],
"no-overshoot": [0.2, 0.4, 0.0667] "no-overshoot": [0.2, 0.4, 0.0667],
} }
def __init__(self, setpoint, out_step=10, lookback=60, def __init__(
noiseband=0.5, sampletime=1.2): self, setpoint, out_step=10, lookback=60, noiseband=0.5, sampletime=1.2
):
if setpoint is None: if setpoint is None:
raise ValueError('setpoint must be specified') raise ValueError("setpoint must be specified")
self._inputs = deque(maxlen=round(lookback / sampletime)) self._inputs = deque(maxlen=round(lookback / sampletime))
self._setpoint = setpoint self._setpoint = setpoint
@ -68,7 +69,7 @@ class PIDAutotune:
"""Get a list of all available tuning rules.""" """Get a list of all available tuning rules."""
return self._tuning_rules.keys() return self._tuning_rules.keys()
def get_pid_parameters(self, tuning_rule='ziegler-nichols'): def get_pid_parameters(self, tuning_rule="ziegler-nichols"):
"""Get PID parameters. """Get PID parameters.
Args: Args:
@ -93,27 +94,33 @@ class PIDAutotune:
""" """
now = time_input * 1000 now = time_input * 1000
if (self._state == PIDAutotuneState.STATE_OFF if (
self._state == PIDAutotuneState.STATE_OFF
or self._state == PIDAutotuneState.STATE_SUCCEEDED or self._state == PIDAutotuneState.STATE_SUCCEEDED
or self._state == PIDAutotuneState.STATE_FAILED): or self._state == PIDAutotuneState.STATE_FAILED
):
self._state = PIDAutotuneState.STATE_RELAY_STEP_UP self._state = PIDAutotuneState.STATE_RELAY_STEP_UP
self._last_run_timestamp = now self._last_run_timestamp = now
# check input and change relay state if necessary # check input and change relay state if necessary
if (self._state == PIDAutotuneState.STATE_RELAY_STEP_UP if (
and input_val > self._setpoint + self._noiseband): self._state == PIDAutotuneState.STATE_RELAY_STEP_UP
and input_val > self._setpoint + self._noiseband
):
self._state = PIDAutotuneState.STATE_RELAY_STEP_DOWN self._state = PIDAutotuneState.STATE_RELAY_STEP_DOWN
logging.debug('switched state: {0}'.format(self._state)) logging.debug("switched state: {0}".format(self._state))
logging.debug('input: {0}'.format(input_val)) logging.debug("input: {0}".format(input_val))
elif (self._state == PIDAutotuneState.STATE_RELAY_STEP_DOWN elif (
and input_val < self._setpoint - self._noiseband): self._state == PIDAutotuneState.STATE_RELAY_STEP_DOWN
and input_val < self._setpoint - self._noiseband
):
self._state = PIDAutotuneState.STATE_RELAY_STEP_UP self._state = PIDAutotuneState.STATE_RELAY_STEP_UP
logging.debug('switched state: {0}'.format(self._state)) logging.debug("switched state: {0}".format(self._state))
logging.debug('input: {0}'.format(input_val)) logging.debug("input: {0}".format(input_val))
# set output # set output
if (self._state == PIDAutotuneState.STATE_RELAY_STEP_UP): if self._state == PIDAutotuneState.STATE_RELAY_STEP_UP:
self._output = self._initial_output - self._outputstep self._output = self._initial_output - self._outputstep
elif self._state == PIDAutotuneState.STATE_RELAY_STEP_DOWN: elif self._state == PIDAutotuneState.STATE_RELAY_STEP_DOWN:
self._output = self._initial_output + self._outputstep self._output = self._initial_output + self._outputstep
@ -156,8 +163,8 @@ class PIDAutotune:
self._peak_count += 1 self._peak_count += 1
self._peaks.append(input_val) self._peaks.append(input_val)
self._peak_timestamps.append(now) self._peak_timestamps.append(now)
logging.debug('found peak: {0}'.format(input_val)) logging.debug("found peak: {0}".format(input_val))
logging.debug('peak count: {0}'.format(self._peak_count)) logging.debug("peak count: {0}".format(self._peak_count))
# check for convergence of induced oscillation # check for convergence of induced oscillation
# convergence of amplitude assessed on last 4 peaks (1.5 cycles) # convergence of amplitude assessed on last 4 peaks (1.5 cycles)
@ -167,20 +174,19 @@ class PIDAutotune:
abs_max = self._peaks[-2] abs_max = self._peaks[-2]
abs_min = self._peaks[-2] abs_min = self._peaks[-2]
for i in range(0, len(self._peaks) - 2): for i in range(0, len(self._peaks) - 2):
self._induced_amplitude += abs(self._peaks[i] self._induced_amplitude += abs(self._peaks[i] - self._peaks[i + 1])
- self._peaks[i+1])
abs_max = max(self._peaks[i], abs_max) abs_max = max(self._peaks[i], abs_max)
abs_min = min(self._peaks[i], abs_min) abs_min = min(self._peaks[i], abs_min)
self._induced_amplitude /= 6.0 self._induced_amplitude /= 6.0
# check convergence criterion for amplitude of induced oscillation # check convergence criterion for amplitude of induced oscillation
amplitude_dev = ((0.5 * (abs_max - abs_min) amplitude_dev = (
- self._induced_amplitude) 0.5 * (abs_max - abs_min) - self._induced_amplitude
/ self._induced_amplitude) ) / self._induced_amplitude
logging.debug('amplitude: {0}'.format(self._induced_amplitude)) logging.debug("amplitude: {0}".format(self._induced_amplitude))
logging.debug('amplitude deviation: {0}'.format(amplitude_dev)) logging.debug("amplitude deviation: {0}".format(amplitude_dev))
if amplitude_dev < PIDAutotune.PEAK_AMPLITUDE_TOLERANCE: if amplitude_dev < PIDAutotune.PEAK_AMPLITUDE_TOLERANCE:
self._state = PIDAutotuneState.STATE_SUCCEEDED self._state = PIDAutotuneState.STATE_SUCCEEDED
@ -194,25 +200,24 @@ class PIDAutotune:
if self._state == PIDAutotuneState.STATE_SUCCEEDED: if self._state == PIDAutotuneState.STATE_SUCCEEDED:
self._output = 0 self._output = 0
logging.debug('peak finding successful') logging.debug("peak finding successful")
# calculate ultimate gain # calculate ultimate gain
self._Ku = 4.0 * self._outputstep / \ self._Ku = 4.0 * self._outputstep / (self._induced_amplitude * math.pi)
(self._induced_amplitude * math.pi) print("Ku: {0}".format(self._Ku))
print('Ku: {0}'.format(self._Ku))
# calculate ultimate period in seconds # calculate ultimate period in seconds
period1 = self._peak_timestamps[3] - self._peak_timestamps[1] period1 = self._peak_timestamps[3] - self._peak_timestamps[1]
period2 = self._peak_timestamps[4] - self._peak_timestamps[2] period2 = self._peak_timestamps[4] - self._peak_timestamps[2]
self._Pu = 0.5 * (period1 + period2) / 1000.0 self._Pu = 0.5 * (period1 + period2) / 1000.0
print('Pu: {0}'.format(self._Pu)) print("Pu: {0}".format(self._Pu))
for rule in self._tuning_rules: for rule in self._tuning_rules:
params = self.get_pid_parameters(rule) params = self.get_pid_parameters(rule)
print('rule: {0}'.format(rule)) print("rule: {0}".format(rule))
print('Kp: {0}'.format(params.Kp)) print("Kp: {0}".format(params.Kp))
print('Ki: {0}'.format(params.Ki)) print("Ki: {0}".format(params.Ki))
print('Kd: {0}'.format(params.Kd)) print("Kd: {0}".format(params.Kd))
return True return True
return False return False
@ -239,16 +244,17 @@ def main():
data = next(tec.report_mode()) data = next(tec.report_mode())
ch = data[channel] ch = data[channel]
tuner = PIDAutotune(target_temperature, output_step, tuner = PIDAutotune(
lookback, noiseband, ch['interval']) target_temperature, output_step, lookback, noiseband, ch["interval"]
)
for data in tec.report_mode(): for data in tec.report_mode():
ch = data[channel] ch = data[channel]
temperature = ch['temperature'] temperature = ch["temperature"]
if (tuner.run(temperature, ch['time'])): if tuner.run(temperature, ch["time"]):
break break
tuner_out = tuner.output() tuner_out = tuner.output()

2
pytec/example.py
View File

@ -1,6 +1,6 @@
from pytec.client import Client from pytec.client import Client
tec = Client() #(host="localhost", port=6667) tec = Client() # (host="localhost", port=6667)
tec.set_param("b-p", 1, "t0", 20) tec.set_param("b-p", 1, "t0", 20)
print(tec.get_output()) print(tec.get_output())
print(tec.get_pid()) print(tec.get_pid())

23
pytec/plot.py
View File

@ -7,9 +7,10 @@ from pytec.client import Client
TIME_WINDOW = 300.0 TIME_WINDOW = 300.0
tec = Client() tec = Client()
target_temperature = tec.get_pid()[0]['target'] target_temperature = tec.get_pid()[0]["target"]
print("Channel 0 target temperature: {:.3f}".format(target_temperature)) print("Channel 0 target temperature: {:.3f}".format(target_temperature))
class Series: class Series:
def __init__(self, conv=lambda x: x): def __init__(self, conv=lambda x: x):
self.conv = conv self.conv = conv
@ -27,24 +28,26 @@ class Series:
self.x_data = self.x_data[drop:] self.x_data = self.x_data[drop:]
self.y_data = self.y_data[drop:] self.y_data = self.y_data[drop:]
series = { series = {
# 'adc': Series(), # 'adc': Series(),
# 'sens': Series(lambda x: x * 0.0001), # 'sens': Series(lambda x: x * 0.0001),
'temperature': Series(), "temperature": Series(),
# 'i_set': Series(), # 'i_set': Series(),
'pid_output': Series(), "pid_output": Series(),
# 'vref': Series(), # 'vref': Series(),
# 'dac_value': Series(), # 'dac_value': Series(),
# 'dac_feedback': Series(), # 'dac_feedback': Series(),
# 'i_tec': Series(), # 'i_tec': Series(),
'tec_i': Series(), "tec_i": Series(),
'tec_u_meas': Series(), "tec_u_meas": Series(),
# 'interval': Series(), # 'interval': Series(),
} }
series_lock = Lock() series_lock = Lock()
quit = False quit = False
def recv_data(tec): def recv_data(tec):
global last_packet_time global last_packet_time
for data in tec.report_mode(): for data in tec.report_mode():
@ -55,22 +58,24 @@ def recv_data(tec):
if k in ch0: if k in ch0:
v = ch0[k] v = ch0[k]
if type(v) is float: if type(v) is float:
s.append(ch0['time'], v) s.append(ch0["time"], v)
finally: finally:
series_lock.release() series_lock.release()
if quit: if quit:
break break
thread = Thread(target=recv_data, args=(tec,)) thread = Thread(target=recv_data, args=(tec,))
thread.start() thread.start()
fig, ax = plt.subplots() fig, ax = plt.subplots()
for k, s in series.items(): for k, s in series.items():
s.plot, = ax.plot([], [], label=k) (s.plot,) = ax.plot([], [], label=k)
legend = ax.legend() legend = ax.legend()
def animate(i): def animate(i):
min_x, max_x, min_y, max_y = None, None, None, None min_x, max_x, min_y, max_y = None, None, None, None
@ -120,8 +125,8 @@ def animate(i):
legend.remove() legend.remove()
legend = ax.legend() legend = ax.legend()
ani = animation.FuncAnimation(
fig, animate, interval=1, blit=False, save_count=50) ani = animation.FuncAnimation(fig, animate, interval=1, blit=False, save_count=50)
plt.show() plt.show()
quit = True quit = True

14
pytec/pytec/client.py
View File

@ -3,9 +3,11 @@ import json
import logging import logging
import time import time
class CommandError(Exception): class CommandError(Exception):
pass pass
class Client: class Client:
def __init__(self, host="192.168.1.26", port=23, timeout=None): def __init__(self, host="192.168.1.26", port=23, timeout=None):
self._socket = socket.create_connection((host, port), timeout) self._socket = socket.create_connection((host, port), timeout)
@ -17,7 +19,11 @@ class Client:
for output_channel in output_report: for output_channel in output_report:
for limit in ["max_i_neg", "max_i_pos", "max_v"]: for limit in ["max_i_neg", "max_i_pos", "max_v"]:
if output_channel[limit] == 0.0: if output_channel[limit] == 0.0:
logging.warning("`{}` limit is set to zero on channel {}".format(limit, output_channel["channel"])) logging.warning(
"`{}` limit is set to zero on channel {}".format(
limit, output_channel["channel"]
)
)
def _read_line(self): def _read_line(self):
# read more lines # read more lines
@ -25,7 +31,7 @@ class Client:
chunk = self._socket.recv(4096) chunk = self._socket.recv(4096)
if not chunk: if not chunk:
return None return None
buf = self._lines[-1] + chunk.decode('utf-8', errors='ignore') buf = self._lines[-1] + chunk.decode("utf-8", errors="ignore")
self._lines = buf.split("\n") self._lines = buf.split("\n")
line = self._lines[0] line = self._lines[0]
@ -33,7 +39,7 @@ class Client:
return line return line
def _command(self, *command): def _command(self, *command):
self._socket.sendall((" ".join(command) + "\n").encode('utf-8')) self._socket.sendall((" ".join(command) + "\n").encode("utf-8"))
line = self._read_line() line = self._read_line()
response = json.loads(line) response = json.loads(line)
@ -130,7 +136,7 @@ class Client:
'pid_output': 2.067581958092247} 'pid_output': 2.067581958092247}
""" """
while True: while True:
self._socket.sendall("report\n".encode('utf-8')) self._socket.sendall("report\n".encode("utf-8"))
line = self._read_line() line = self._read_line()
if not line: if not line:
break break