pytec: __name__ check, examples does not run when pytec module is imported, change autotune state management to use enum

This commit is contained in:
topquark12 2021-01-04 10:54:04 +08:00
parent d1f8a4761b
commit ab305d5fa5
3 changed files with 175 additions and 164 deletions

View File

@ -3,35 +3,41 @@ import logging
from time import time
from collections import deque, namedtuple
from pytec.client import Client
from enum import Enum
# Based on hirshmann pid-autotune libiary
# See https://github.com/hirschmann/pid-autotune
# Which is in turn based on a fork of Arduino PID AutoTune Library
# See https://github.com/t0mpr1c3/Arduino-PID-AutoTune-Library
# Auto tune parameters
# Thermostat channel
channel = 0
# Target temperature of the autotune routine, celcius
target_temperature = 30
# Value by which output will be increased/decreased from zero, amps
output_step = 1
# Reference period for local minima/maxima, seconds
lookback = 3
# Determines by how much the input value must overshoot/undershoot the setpoint, celcius
noiseband = 1.5
if __name__ == "__main__":
class PIDAutotune(object):
# Auto tune parameters
# Thermostat channel
channel = 0
# Target temperature of the autotune routine, celcius
target_temperature = 30
# Value by which output will be increased/decreased from zero, amps
output_step = 1
# Reference period for local minima/maxima, seconds
lookback = 3
# Determines by how much the input value must overshoot/undershoot the setpoint, celcius
noiseband = 1.5
PIDParams = namedtuple('PIDParams', ['Kp', 'Ki', 'Kd'])
PEAK_AMPLITUDE_TOLERANCE = 0.05
class PIDAutotuneState(Enum):
STATE_OFF = 'off'
STATE_RELAY_STEP_UP = 'relay step up'
STATE_RELAY_STEP_DOWN = 'relay step down'
STATE_SUCCEEDED = 'succeeded'
STATE_FAILED = 'failed'
class PIDAutotune():
PIDParams = namedtuple('PIDParams', ['Kp', 'Ki', 'Kd'])
PEAK_AMPLITUDE_TOLERANCE = 0.05
_tuning_rules = {
"ziegler-nichols": [0.6, 1.2, 0.075],
"tyreus-luyben": [0.4545, 0.2066, 0.07214],
@ -62,7 +68,7 @@ class PIDAutotune(object):
self._noiseband = noiseband
self._out_min = -out_step
self._out_max = out_step
self._state = PIDAutotune.STATE_OFF
self._state = PIDAutotuneState.STATE_OFF
self._peak_timestamps = deque(maxlen=5)
self._peaks = deque(maxlen=5)
self._output = 0
@ -115,30 +121,30 @@ class PIDAutotune(object):
"""
now = time_input * 1000
if (self._state == PIDAutotune.STATE_OFF
or self._state == PIDAutotune.STATE_SUCCEEDED
or self._state == PIDAutotune.STATE_FAILED):
if (self._state == PIDAutotuneState.STATE_OFF
or self._state == PIDAutotuneState.STATE_SUCCEEDED
or self._state == PIDAutotuneState.STATE_FAILED):
self._initTuner(input_val, now)
self._last_run_timestamp = now
# print("temp : ", input_val)
# check input and change relay state if necessary
if (self._state == PIDAutotune.STATE_RELAY_STEP_UP
if (self._state == PIDAutotuneState.STATE_RELAY_STEP_UP
and input_val > self._setpoint + self._noiseband):
self._state = PIDAutotune.STATE_RELAY_STEP_DOWN
self._state = PIDAutotuneState.STATE_RELAY_STEP_DOWN
logging.debug('switched state: {0}'.format(self._state))
logging.debug('input: {0}'.format(input_val))
elif (self._state == PIDAutotune.STATE_RELAY_STEP_DOWN
elif (self._state == PIDAutotuneState.STATE_RELAY_STEP_DOWN
and input_val < self._setpoint - self._noiseband):
self._state = PIDAutotune.STATE_RELAY_STEP_UP
self._state = PIDAutotuneState.STATE_RELAY_STEP_UP
logging.debug('switched state: {0}'.format(self._state))
logging.debug('input: {0}'.format(input_val))
# set output
if (self._state == PIDAutotune.STATE_RELAY_STEP_UP):
if (self._state == PIDAutotuneState.STATE_RELAY_STEP_UP):
self._output = self._initial_output + self._outputstep
elif self._state == PIDAutotune.STATE_RELAY_STEP_DOWN:
elif self._state == PIDAutotuneState.STATE_RELAY_STEP_DOWN:
self._output = self._initial_output - self._outputstep
# respect output limits
@ -204,17 +210,17 @@ class PIDAutotune(object):
logging.debug('amplitude deviation: {0}'.format(amplitude_dev))
if amplitude_dev < PIDAutotune.PEAK_AMPLITUDE_TOLERANCE:
self._state = PIDAutotune.STATE_SUCCEEDED
self._state = PIDAutotuneState.STATE_SUCCEEDED
# logging.debug('peak finding succeeded')
# if the autotune has not already converged
# terminate after 10 cycles
if self._peak_count >= 20:
self._output = 0
self._state = PIDAutotune.STATE_FAILED
self._state = PIDAutotuneState.STATE_FAILED
return True
if self._state == PIDAutotune.STATE_SUCCEEDED:
if self._state == PIDAutotuneState.STATE_SUCCEEDED:
self._output = 0
logging.debug('peak finding successful')
@ -250,18 +256,20 @@ class PIDAutotune(object):
self._peaks.clear()
self._peak_timestamps.clear()
self._peak_timestamps.append(timestamp)
self._state = PIDAutotune.STATE_RELAY_STEP_UP
self._state = PIDAutotuneState.STATE_RELAY_STEP_UP
# logging.basicConfig(level=logging.DEBUG)
if __name__ == "__main__":
tec = Client() #(host="localhost", port=6667)
# logging.basicConfig(level=logging.DEBUG)
data = next(tec.report_mode())
ch = data[channel]
tec = Client() #(host="localhost", port=6667)
tuner = PIDAutotune(target_temperature, output_step, lookback, noiseband, ch['interval'])
data = next(tec.report_mode())
ch = data[channel]
for data in tec.report_mode():
tuner = PIDAutotune(target_temperature, output_step, lookback, noiseband, ch['interval'])
for data in tec.report_mode():
try:
ch = data[channel]
@ -269,7 +277,6 @@ for data in tec.report_mode():
temperature = ch['temperature']
if (tuner.run(temperature, ch['time'])):
# logging.debug('true')
break
tunerOut = tuner.output()
@ -279,4 +286,4 @@ for data in tec.report_mode():
except:
pass
tec.set_param("pwm", channel, "i_set" , channel)
tec.set_param("pwm", channel, "i_set" , channel)

View File

@ -1,11 +1,13 @@
from pytec.client import Client
if __name__ == "__main__":
tec = Client() #(host="localhost", port=6667)
tec.set_param("s-h", 1, "t0", 20)
print(tec.get_pwm())
print(tec.get_pid())
print(tec.get_pwm())
print(tec.get_postfilter())
print(tec.get_steinhart_hart())
for data in tec.report_mode():
from pytec.client import Client
tec = Client() #(host="localhost", port=6667)
tec.set_param("s-h", 1, "t0", 20)
print(tec.get_pwm())
print(tec.get_pid())
print(tec.get_pwm())
print(tec.get_postfilter())
print(tec.get_steinhart_hart())
for data in tec.report_mode():
print(data)

View File

@ -4,13 +4,15 @@ import matplotlib.animation as animation
from threading import Thread, Lock
from pytec.client import Client
TIME_WINDOW = 300.0
if __name__ == "__main__":
tec = Client()
target_temperature = tec.get_pid()[0]['target']
print("Channel 0 target temperature: {:.3f}".format(target_temperature))
TIME_WINDOW = 300.0
class Series:
tec = Client()
target_temperature = tec.get_pid()[0]['target']
print("Channel 0 target temperature: {:.3f}".format(target_temperature))
class Series:
def __init__(self, conv=lambda x: x):
self.conv = conv
self.x_data = []
@ -27,10 +29,10 @@ class Series:
self.x_data = self.x_data[drop:]
self.y_data = self.y_data[drop:]
series = {
series = {
'adc': Series(),
'sens': Series(lambda x: x * 0.0001),
'temperature': Series(lambda t: t - target_temperature),
'temperature': Series(),
'i_set': Series(),
'pid_output': Series(),
'vref': Series(),
@ -40,12 +42,12 @@ series = {
'tec_i': Series(),
'tec_u_meas': 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
for data in tec.report_mode():
ch0 = data[0]
@ -62,16 +64,16 @@ def recv_data(tec):
if quit:
break
thread = Thread(target=recv_data, args=(tec,))
thread.start()
thread = Thread(target=recv_data, args=(tec,))
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)
legend = ax.legend()
legend = ax.legend()
def animate(i):
def animate(i):
min_x, max_x, min_y, max_y = None, None, None, None
series_lock.acquire()
@ -120,9 +122,9 @@ def animate(i):
legend.remove()
legend = ax.legend()
ani = animation.FuncAnimation(
ani = animation.FuncAnimation(
fig, animate, interval=1, blit=False, save_count=50)
plt.show()
quit = True
thread.join()
plt.show()
quit = True
thread.join()