1
0
forked from M-Labs/thermostat

Compare commits

...

114 Commits

Author SHA1 Message Date
a7ba382dad QVariantList -> list & QVariantMap -> dict 2024-09-16 10:35:35 +08:00
b7cc9db03c No need to create a new task for waiting 2024-09-09 18:29:24 +08:00
92d574a0e3 fixup! Zero limits warning cleanup 2024-09-09 16:29:32 +08:00
2d3bc96da9 super init's first 2024-09-09 15:44:07 +08:00
7efd184496 Format 2024-09-09 15:10:30 +08:00
14621e11a6 ch_tuning -> autotuning_channels 2024-09-09 15:10:16 +08:00
cfe9ee3985 Module docstring?? 2024-09-09 15:09:04 +08:00
a60f55d1c8 Reorder imports 2024-09-09 15:08:17 +08:00
a466e3e09d Sensible names for views 2024-09-09 15:06:22 +08:00
b0a4e1218a fixup! tec_qt: Private everything possible 2024-09-09 15:00:44 +08:00
d199770de1 Unprivate conn_menu
Used for autoconnect
2024-09-09 15:00:14 +08:00
9911f72059 Order? 2024-09-09 14:54:37 +08:00
f62adbbc1a Reorder 2024-09-09 14:54:37 +08:00
c416fd8b2b Not async 2024-09-09 14:54:37 +08:00
b3ba577268 Remove report mode from thermostat data model 2024-09-09 14:54:37 +08:00
3c1228e8a8 Remove report mode box 2024-09-09 14:54:37 +08:00
23653a1ecd More sensible names 2024-09-09 14:54:37 +08:00
29221ce570 tec_qt: Private everything possible 2024-09-09 14:54:35 +08:00
2334a922f7 Remove error handling for connecting task
Just let the exception propagate, even when stopping the connection
2024-09-09 14:54:24 +08:00
1aab3ca1d6 Update thermostat state from controller code 2024-09-09 14:54:24 +08:00
41d4154a28 extra func out 2024-09-09 14:54:24 +08:00
6473905488 No need extra function 2024-09-09 14:54:24 +08:00
b13f481381 State update explicitly 2024-09-09 14:54:24 +08:00
66303efd11 Zero limits warning cleanup 2024-09-02 16:12:55 +08:00
f0e7488682 Add NUM_CHANNELS to Thermostat! 2024-09-02 16:06:21 +08:00
70627342f3 Real concurrently 2024-08-30 17:37:53 +08:00
ffa0783cff Don't create report task if not polling for it 2024-08-30 16:55:15 +08:00
ba6b24fe95 PIDAutotuner: Don't use separate sig for interval 2024-08-30 16:55:15 +08:00
2e0ecd9758 aioexample: Show polling while report mode on
WIP: handle KeyboardInterrupt
2024-08-30 16:55:15 +08:00
199aeb3609 aioclient: More accurate DFU docstring 2024-08-30 16:55:15 +08:00
1e81641169 No need for async as well 2024-08-30 16:55:15 +08:00
2c95bd9d09 No need for async 2024-08-30 16:55:15 +08:00
d83c4ed4f6 Remove extra imports 2024-08-30 16:55:15 +08:00
637d8d029a report group 2024-08-30 16:55:15 +08:00
671004703b end_session in thermostat itself afterall 2024-08-30 16:55:15 +08:00
10873933f4 Reorder MainWindow 2024-08-30 16:55:15 +08:00
ec1fffecd1 Rearrange MainWindow.__init__ 2024-08-30 16:55:15 +08:00
36c063d917 Thermostat: Add disconnect callback
For communicating with the autotuner before the client fully
disconnects

Also then there's no need for explicitly resetting autotune elements
2024-08-30 16:55:15 +08:00
65b9202cea Remove info_box_trigger 2024-08-30 16:55:15 +08:00
a72a0606f1 Remove connection errored 2024-08-30 16:55:15 +08:00
81f09a9f0f Simply test for connectivity for turning PID off 2024-08-30 16:55:15 +08:00
f768ea5d4f Fix pressing enter not working in conn menu 2024-08-30 16:55:15 +08:00
e849610658 Assign connection_errored as well 2024-08-30 16:55:15 +08:00
bbe9a60824 Back out 2024-08-30 16:55:15 +08:00
cd07cdeddd Put UI changes into conn_menu 2024-08-30 16:55:15 +08:00
fe3a22a599 ConnectionBtn + StatusLbl 2024-08-30 16:55:15 +08:00
2b35bf57f9 Clear Graph UI changes 2024-08-30 16:55:15 +08:00
8a83bbc1d1 Concentrate ThermostatCtrlMenu UI changes 2024-08-30 16:55:15 +08:00
82315d65b2 Descriptive name 2024-08-30 16:55:15 +08:00
dfe08546fe Refactor repeated stuff 2024-08-30 16:55:15 +08:00
c4309c5ba0 Split PID handler UI up 2024-08-30 16:55:15 +08:00
3387ad67f7 Fix hwrev getting 2024-08-30 16:55:15 +08:00
903f9f8767 hwrev updates 2024-08-30 15:53:05 +08:00
aadf8a2595 Fix info boxes for load/saving from flash 2024-08-30 15:53:05 +08:00
49561a2f5a Move reset request to thermostat control menu
We don't get auto reconnect anymore
2024-08-30 15:53:05 +08:00
b32b38d830 Move pid autotuning request to CtrlPanel
And update autotune UI only on state change instead of every single
report update
2024-08-30 15:53:05 +08:00
fcda46d1f3 Move channelGraph stuff inside LiveDataPlotter 2024-08-30 15:53:05 +08:00
89e9da4499 Move plot_options_menu stuff into menu 2024-08-30 15:53:05 +08:00
87ca0694cd Move autotune ticking connect 2024-08-30 15:53:05 +08:00
03233386d0 Get thermostat_ctrl_menu to subscribe to hwrev updates 2024-08-30 15:53:05 +08:00
242a2b2e69 Save/load info box content 2024-08-30 15:53:05 +08:00
b84af82722 Fix loading all channel settings would bring up 2 info boxes 2024-08-30 15:53:05 +08:00
7997ef2472 Add back asyncSlot to save_cfg_request 2024-08-30 15:53:05 +08:00
911b4c680f hw_rev 2024-08-30 15:53:05 +08:00
19677c2251 fan_update: tec_qt to thermostat_ctrl_menu 2024-08-30 15:53:05 +08:00
5f30f12875 Fix updating status label for hwrev = None 2024-08-30 15:53:05 +08:00
3283b126b8 Put some menu requests in menu itself 2024-08-30 15:53:05 +08:00
59ba8c692d Broadcast ConnectionState changes from Thermostat 2024-08-30 15:53:05 +08:00
e3ef9115a5 Move sigActivatedHandle to CtrlPanel 2024-08-30 15:53:05 +08:00
c5c311ae14 Get rid of timeout on readline 2024-08-30 15:53:05 +08:00
6699458939 Put send_command in CtrlPanel 2024-08-30 15:53:05 +08:00
8a4a963f9b Extra bail removed 2024-08-30 15:53:05 +08:00
2f10ad8056 return hwrev when start_session 2024-08-30 15:53:05 +08:00
3731282146 state str 2024-08-30 15:53:05 +08:00
50ac1c2078 params update concurrently 2024-08-30 15:53:05 +08:00
0eebc0a6df Stuff non-UI changes in Thermostat model 2024-08-30 15:53:05 +08:00
78ba799d56 Base Thermostat connectivity UI changes to state 2024-08-30 15:53:05 +08:00
f9c0f140fa Remove wait_for
OSError raised anyways
2024-08-30 15:53:05 +08:00
93d96401b9 Actually its OSError 2024-08-30 15:53:05 +08:00
881bf6ff6c {start,end}_session -> [dis]connect 2024-08-30 15:53:05 +08:00
7e0dd61f79 Simplify on_connect_btn_clicked
Raise if OSError
2024-08-30 15:53:05 +08:00
ed34eb9e03 Connecting task moved? 2024-08-30 15:53:05 +08:00
387d1226c5 AsyncIO version Client -> AsyncioClient 2024-08-30 15:53:05 +08:00
058e597ef0 Exclusively use the Thermostat object as a medium
All calls to the Thermostat should be forwarded by the medium.
2024-08-30 15:53:05 +08:00
80975f50b1 Integrate WrappedClient into Thermostat model 2024-08-30 15:53:05 +08:00
86afa19085 Should not stop cancelling read if timeout'd 2024-08-30 15:53:05 +08:00
4b6dfdf478 Fix Autotuner state for forceful disconnect 2024-08-30 15:53:05 +08:00
85cb2027a8 _ 2024-08-30 15:53:05 +08:00
4fedb14727 Make connection loss handling more elegant
Show an info box on connection lost informing the user that the
Thermostat was forcefully disconnected.
2024-08-30 15:53:05 +08:00
863920f922 ================gui_dev-fix_asyncio=============== 2024-08-30 15:53:05 +08:00
96b57f199d Use new style super() 2024-08-30 15:53:05 +08:00
f2da457860 Move command line host:port setting handling
To main
2024-08-30 15:40:31 +08:00
066c75fc21 aioclient: Add missing readline for saving
Saving all channels returns multiple JSON objects, read the extra {}.
2024-08-30 15:38:59 +08:00
60b05ddfeb import order rearrange 2024-08-30 15:38:59 +08:00
6a78a2069d This is bail 2024-08-30 15:38:59 +08:00
6eb5b7298e thermostat: Properly register task
Also Thermostat.task -> Thermostat._update_params_task
2024-08-30 15:38:58 +08:00
4a292c19db ip -> host 2024-08-30 15:38:58 +08:00
0b115b3394 Lazy evaluating for debug string command 2024-08-30 15:38:58 +08:00
2e904cb498 Add pytec runnables 2024-08-30 15:38:58 +08:00
d3f378f058 PYTHON shell 2024-08-30 15:38:58 +08:00
d0273cd41b Exactlier wording 2024-08-30 15:38:58 +08:00
0988ed24d6 unused 2024-08-30 15:38:58 +08:00
ab3b26a2a4 encoding 2024-08-30 15:38:58 +08:00
000b97398b Just catch asyncio.TimeoutError
Will just change to TimeoutError once we switch to Python 3.11 in the
flake.
2024-08-30 15:38:58 +08:00
e162bcf210 Remove exception too general 2024-08-30 15:38:58 +08:00
c3d7804898 Use asserts to check for connectivity 2024-08-30 15:38:58 +08:00
9c0977262c Add back the parent 2024-08-30 15:38:58 +08:00
017206bd34 Fix method call 2024-08-30 15:38:58 +08:00
f46bd2b50e README: Proofread 2024-08-30 15:38:58 +08:00
5b48fb2793 Swap order arounda bit more 2024-08-30 15:38:58 +08:00
358403e651 Formatting 2024-08-30 15:38:58 +08:00
4a48379923 Use qtextras 2024-08-30 15:38:58 +08:00
b766e949e8 flake update 2024-08-30 15:38:58 +08:00
2fc0d265e4 flake.nix: nixfmt-rfc-style 2024-08-26 12:16:57 +08:00
23 changed files with 789 additions and 677 deletions

View File

@ -71,9 +71,8 @@ openocd -f interface/stlink.cfg -f target/stm32f4x.cfg -c "program target/thumbv
A GUI has been developed for easy configuration and plotting of key parameters.
The Python GUI program is located at pytec/tec_qt.py.
The GUI is developed based on the Python library pyqtgraph. The GUI can be configured and launched automatically by running:
The Python GUI program is located at pytec/tec_qt.py, and is developed based on the Python library pyqtgraph. The GUI can be configured and
launched automatically by running:
```
nix run .#thermostat_gui

12
flake.lock generated
View File

@ -3,11 +3,11 @@
"mozilla-overlay": {
"flake": false,
"locked": {
"lastModified": 1690536331,
"narHash": "sha256-aRIf2FB2GTdfF7gl13WyETmiV/J7EhBGkSWXfZvlxcA=",
"lastModified": 1704373101,
"narHash": "sha256-+gi59LRWRQmwROrmE1E2b3mtocwueCQqZ60CwLG+gbg=",
"owner": "mozilla",
"repo": "nixpkgs-mozilla",
"rev": "db89c8707edcffefcd8e738459d511543a339ff5",
"rev": "9b11a87c0cc54e308fa83aac5b4ee1816d5418a2",
"type": "github"
},
"original": {
@ -18,11 +18,11 @@
},
"nixpkgs": {
"locked": {
"lastModified": 1691421349,
"narHash": "sha256-RRJyX0CUrs4uW4gMhd/X4rcDG8PTgaaCQM5rXEJOx6g=",
"lastModified": 1704290814,
"narHash": "sha256-LWvKHp7kGxk/GEtlrGYV68qIvPHkU9iToomNFGagixU=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "011567f35433879aae5024fc6ec53f2a0568a6c4",
"rev": "70bdadeb94ffc8806c0570eb5c2695ad29f0e421",
"type": "github"
},
"original": {

View File

@ -1,20 +1,25 @@
{
description = "Firmware for the Sinara 8451 Thermostat";
inputs.nixpkgs.url = github:NixOS/nixpkgs/nixos-23.05;
inputs.mozilla-overlay = { url = github:mozilla/nixpkgs-mozilla; flake = false; };
inputs.nixpkgs.url = "github:NixOS/nixpkgs/nixos-23.05";
inputs.mozilla-overlay = {
url = "github:mozilla/nixpkgs-mozilla";
flake = false;
};
outputs = { self, nixpkgs, mozilla-overlay }:
outputs = { self, nixpkgs, mozilla-overlay, }:
let
pkgs = import nixpkgs { system = "x86_64-linux"; overlays = [ (import mozilla-overlay) ]; };
pkgs = import nixpkgs {
system = "x86_64-linux";
overlays = [ (import mozilla-overlay) ];
};
rustManifest = pkgs.fetchurl {
url = "https://static.rust-lang.org/dist/2022-12-15/channel-rust-stable.toml";
url =
"https://static.rust-lang.org/dist/2022-12-15/channel-rust-stable.toml";
hash = "sha256-S7epLlflwt0d1GZP44u5Xosgf6dRrmr8xxC+Ml2Pq7c=";
};
targets = [
"thumbv7em-none-eabihf"
];
targets = [ "thumbv7em-none-eabihf" ];
rustChannelOfTargets = _channel: _date: targets:
(pkgs.lib.rustLib.fromManifestFile rustManifest {
inherit (pkgs) stdenv lib fetchurl patchelf;
@ -35,7 +40,8 @@
cargoLock = {
lockFile = ./Cargo.lock;
outputHashes = {
"stm32-eth-0.2.0" = "sha256-48RpZgagUqgVeKm7GXdk3Oo0v19ScF9Uby0nTFlve2o=";
"stm32-eth-0.2.0" =
"sha256-48RpZgagUqgVeKm7GXdk3Oo0v19ScF9Uby0nTFlve2o=";
};
};
@ -79,6 +85,22 @@
propagatedBuildInputs = with pkgs.python3Packages; [ numpy pyqt6 ];
};
qtextras = pkgs.python3Packages.buildPythonPackage rec {
pname = "qtextras";
version = "0.6.8";
format = "pyproject";
src = pkgs.fetchPypi {
inherit pname version;
hash = "sha256-d1ZotSlOI4surUy0H0N4xHoq94IRQvMHunwRH1uubFg=";
};
buildInputs = [ pkgs.python3Packages.hatchling ];
propagatedBuildInputs = with pkgs.python3Packages; [
numpy
pyqtgraph
ruamel-yaml
];
};
pglive = pkgs.python3Packages.buildPythonPackage rec {
pname = "pglive";
version = "0.7.2";
@ -91,6 +113,19 @@
propagatedBuildInputs = [ pyqtgraph pkgs.python3Packages.numpy ];
};
pytec-dev-wrappers = pkgs.runCommandNoCC "pytec-dev-wrappers" { } ''
mkdir -p $out/bin
for program in ${self}/pytec/*.py; do
if [ -x $program ]; then
progname=`basename -s .py $program`
outname=$out/bin/$progname
echo "#!${pkgs.bash}/bin/bash" >> $outname
echo "exec python3 -m pytec.$progname \"\$@\"" >> $outname
chmod 755 $outname
fi
done
'';
thermostat_gui = pkgs.python3Packages.buildPythonPackage {
pname = "thermostat_gui";
version = "0.0.0";
@ -98,7 +133,14 @@
src = "${self}/pytec";
nativeBuildInputs = [ pkgs.qt6.wrapQtAppsHook ];
propagatedBuildInputs = [ pkgs.qt6.qtbase ] ++ (with pkgs.python3Packages; [ pyqtgraph pyqt6 qasync pglive ]);
propagatedBuildInputs = [ pkgs.qt6.qtbase ]
++ (with pkgs.python3Packages; [
pyqtgraph
pyqt6
qasync
pglive
qtextras
]);
dontWrapQtApps = true;
postFixup = ''
@ -106,26 +148,34 @@
'';
};
in {
packages.x86_64-linux = {
inherit thermostat thermostat_gui;
};
packages.x86_64-linux = { inherit thermostat thermostat_gui; };
formatter.x86_64-linux = nixpkgs.legacyPackages.x86_64-linux.nixfmt;
apps.x86_64-linux.thermostat_gui = {
type = "app";
program = "${self.packages.x86_64-linux.thermostat_gui}/bin/tec_qt";
};
hydraJobs = {
inherit thermostat;
};
hydraJobs = { inherit thermostat; };
devShell.x86_64-linux = pkgs.mkShell {
name = "thermostat-dev-shell";
buildInputs = with pkgs; [
rust openocd dfu-util
] ++ (with python3Packages; [
numpy matplotlib pyqtgraph setuptools pyqt6 qasync pglive
buildInputs = with pkgs;
[ rust openocd dfu-util pytec-dev-wrappers ]
++ (with python3Packages; [
numpy
matplotlib
pyqtgraph
setuptools
pyqt6
qasync
pglive
qtextras
]);
shellHook = ''
export PYTHONPATH=`pwd`/pytec:$PYTHONPATH
'';
};
defaultPackage.x86_64-linux = thermostat;
};

3
pytec/.flake8 Normal file
View File

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

96
pytec/autotune.py Normal file → Executable file
View File

@ -12,16 +12,16 @@ from pytec.client import Client
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'
STATE_READY = 'ready'
STATE_OFF = "off"
STATE_RELAY_STEP_UP = "relay step up"
STATE_RELAY_STEP_DOWN = "relay step down"
STATE_SUCCEEDED = "succeeded"
STATE_FAILED = "failed"
STATE_READY = "ready"
class PIDAutotune:
PIDParams = namedtuple('PIDParams', ['Kp', 'Ki', 'Kd'])
PIDParams = namedtuple("PIDParams", ["Kp", "Ki", "Kd"])
PEAK_AMPLITUDE_TOLERANCE = 0.05
@ -31,13 +31,14 @@ class PIDAutotune:
"ciancone-marlin": [0.303, 0.1364, 0.0481],
"pessen-integral": [0.7, 1.75, 0.105],
"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,
noiseband=0.5, sampletime=1.2):
def __init__(
self, setpoint, out_step=10, lookback=60, noiseband=0.5, sampletime=1.2
):
if setpoint is None:
raise ValueError('setpoint must be specified')
raise ValueError("setpoint must be specified")
self._inputs = deque(maxlen=round(lookback / sampletime))
self._setpoint = setpoint
@ -84,7 +85,7 @@ class PIDAutotune:
"""Get a list of all available tuning rules."""
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.
Args:
@ -116,28 +117,34 @@ class PIDAutotune:
"""
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_FAILED
or self._state == PIDAutotuneState.STATE_READY):
or self._state == PIDAutotuneState.STATE_READY
):
self._state = PIDAutotuneState.STATE_RELAY_STEP_UP
self._last_run_timestamp = now
# check input and change relay state if necessary
if (self._state == PIDAutotuneState.STATE_RELAY_STEP_UP
and input_val > self._setpoint + self._noiseband):
if (
self._state == PIDAutotuneState.STATE_RELAY_STEP_UP
and input_val > self._setpoint + self._noiseband
):
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 == PIDAutotuneState.STATE_RELAY_STEP_DOWN
and input_val < self._setpoint - self._noiseband):
logging.debug("switched state: {0}".format(self._state))
logging.debug("input: {0}".format(input_val))
elif (
self._state == PIDAutotuneState.STATE_RELAY_STEP_DOWN
and input_val < self._setpoint - self._noiseband
):
self._state = PIDAutotuneState.STATE_RELAY_STEP_UP
logging.debug('switched state: {0}'.format(self._state))
logging.debug('input: {0}'.format(input_val))
logging.debug("switched state: {0}".format(self._state))
logging.debug("input: {0}".format(input_val))
# 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
elif self._state == PIDAutotuneState.STATE_RELAY_STEP_DOWN:
self._output = self._initial_output + self._outputstep
@ -180,8 +187,8 @@ class PIDAutotune:
self._peak_count += 1
self._peaks.append(input_val)
self._peak_timestamps.append(now)
logging.debug('found peak: {0}'.format(input_val))
logging.debug('peak count: {0}'.format(self._peak_count))
logging.debug("found peak: {0}".format(input_val))
logging.debug("peak count: {0}".format(self._peak_count))
# check for convergence of induced oscillation
# convergence of amplitude assessed on last 4 peaks (1.5 cycles)
@ -191,20 +198,19 @@ class PIDAutotune:
abs_max = self._peaks[-2]
abs_min = self._peaks[-2]
for i in range(0, len(self._peaks) - 2):
self._induced_amplitude += abs(self._peaks[i]
- self._peaks[i+1])
self._induced_amplitude += abs(self._peaks[i] - self._peaks[i + 1])
abs_max = max(self._peaks[i], abs_max)
abs_min = min(self._peaks[i], abs_min)
self._induced_amplitude /= 6.0
# check convergence criterion for amplitude of induced oscillation
amplitude_dev = ((0.5 * (abs_max - abs_min)
- self._induced_amplitude)
/ self._induced_amplitude)
amplitude_dev = (
0.5 * (abs_max - abs_min) - self._induced_amplitude
) / self._induced_amplitude
logging.debug('amplitude: {0}'.format(self._induced_amplitude))
logging.debug('amplitude deviation: {0}'.format(amplitude_dev))
logging.debug("amplitude: {0}".format(self._induced_amplitude))
logging.debug("amplitude deviation: {0}".format(amplitude_dev))
if amplitude_dev < PIDAutotune.PEAK_AMPLITUDE_TOLERANCE:
self._state = PIDAutotuneState.STATE_SUCCEEDED
@ -218,25 +224,24 @@ class PIDAutotune:
if self._state == PIDAutotuneState.STATE_SUCCEEDED:
self._output = 0
logging.debug('peak finding successful')
logging.debug("peak finding successful")
# calculate ultimate gain
self._Ku = 4.0 * self._outputstep / \
(self._induced_amplitude * math.pi)
logging.debug('Ku: {0}'.format(self._Ku))
self._Ku = 4.0 * self._outputstep / (self._induced_amplitude * math.pi)
logging.debug("Ku: {0}".format(self._Ku))
# calculate ultimate period in seconds
period1 = self._peak_timestamps[3] - self._peak_timestamps[1]
period2 = self._peak_timestamps[4] - self._peak_timestamps[2]
self._Pu = 0.5 * (period1 + period2) / 1000.0
logging.debug('Pu: {0}'.format(self._Pu))
logging.debug("Pu: {0}".format(self._Pu))
for rule in self._tuning_rules:
params = self.get_pid_parameters(rule)
logging.debug('rule: {0}'.format(rule))
logging.debug('Kp: {0}'.format(params.Kp))
logging.debug('Ki: {0}'.format(params.Ki))
logging.debug('Kd: {0}'.format(params.Kd))
logging.debug("rule: {0}".format(rule))
logging.debug("Kp: {0}".format(params.Kp))
logging.debug("Ki: {0}".format(params.Ki))
logging.debug("Kd: {0}".format(params.Kd))
return True
return False
@ -263,16 +268,17 @@ def main():
data = next(tec.report_mode())
ch = data[channel]
tuner = PIDAutotune(target_temperature, output_step,
lookback, noiseband, ch['interval'])
tuner = PIDAutotune(
target_temperature, output_step, lookback, noiseband, ch["interval"]
)
for data in tec.report_mode():
ch = data[channel]
temperature = ch['temperature']
temperature = ch["temperature"]
if (tuner.run(temperature, ch['time'])):
if tuner.run(temperature, ch["time"]):
break
tuner_out = tuner.output()

View File

@ -1,16 +1,36 @@
import asyncio
from pytec.aioclient import Client
from contextlib import suppress
from pytec.aioclient import AsyncioClient
async def poll_for_info(tec):
while True:
print(tec.get_pwm())
print(tec.get_steinhart_hart())
print(tec.get_pid())
print(tec.get_postfilter())
print(tec.get_fan())
await asyncio.sleep(1)
async def main():
tec = Client()
await tec.start_session() #(host="192.168.1.26", port=23)
tec = AsyncioClient()
await tec.connect() # (host="192.168.1.26", port=23)
await tec.set_param("s-h", 1, "t0", 20)
print(await tec.get_pwm())
print(await tec.get_pid())
print(await tec.get_pwm())
print(await tec.get_postfilter())
print(await tec.get_steinhart_hart())
polling_task = asyncio.create_task(poll_for_info(tec))
async for data in tec.report_mode():
print(data)
polling_task.cancel()
with suppress(asyncio.CancelledError):
await polling_task
asyncio.run(main())

23
pytec/plot.py Normal file → Executable file
View File

@ -7,9 +7,10 @@ from pytec.client import Client
TIME_WINDOW = 300.0
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))
class Series:
def __init__(self, conv=lambda x: x):
self.conv = conv
@ -27,24 +28,26 @@ class Series:
self.x_data = self.x_data[drop:]
self.y_data = self.y_data[drop:]
series = {
# 'adc': Series(),
# 'sens': Series(lambda x: x * 0.0001),
'temperature': Series(),
"temperature": Series(),
# 'i_set': Series(),
'pid_output': Series(),
"pid_output": Series(),
# 'vref': Series(),
# 'dac_value': Series(),
# 'dac_feedback': Series(),
# 'i_tec': Series(),
'tec_i': Series(),
'tec_u_meas': Series(),
"tec_i": Series(),
"tec_u_meas": Series(),
# 'interval': Series(),
}
series_lock = Lock()
quit = False
def recv_data(tec):
global last_packet_time
for data in tec.report_mode():
@ -55,22 +58,24 @@ def recv_data(tec):
if k in ch0:
v = ch0[k]
if type(v) is float:
s.append(ch0['time'], v)
s.append(ch0["time"], v)
finally:
series_lock.release()
if quit:
break
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)
(s.plot,) = ax.plot([], [], label=k)
legend = ax.legend()
def animate(i):
min_x, max_x, min_y, max_y = None, None, None, None
@ -120,8 +125,8 @@ def animate(i):
legend.remove()
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()
quit = True

View File

@ -16,3 +16,6 @@ tec_qt = "tec_qt:main"
[tool.setuptools]
packages.find = {}
py-modules = ["autotune", "plot", "tec_qt"]
[tool.pylint.format]
max-line-length = "88"

View File

@ -2,58 +2,34 @@ import asyncio
import json
import logging
class CommandError(Exception):
pass
class StoppedConnecting(Exception):
pass
class Client:
class AsyncioClient:
def __init__(self):
self._reader = None
self._writer = None
self._connecting_task = None
self._command_lock = asyncio.Lock()
self._report_mode_on = False
self.timeout = None
async def start_session(self, host='192.168.1.26', port=23, timeout=None):
"""Start session to Thermostat at specified host and port.
Throws StoppedConnecting if disconnect was called while connecting.
Throws asyncio.TimeoutError if timeout was exceeded.
async def connect(self, host="192.168.1.26", port=23):
"""Connect to Thermostat at specified host and port.
Example::
client = Client()
try:
await client.start_session()
except StoppedConnecting:
print("Stopped connecting")
client = AsyncioClient()
await client.connect()
"""
self._connecting_task = asyncio.create_task(
asyncio.wait_for(asyncio.open_connection(host, port), timeout)
)
self.timeout = timeout
try:
self._reader, self._writer = await self._connecting_task
except asyncio.CancelledError:
raise StoppedConnecting
finally:
self._connecting_task = None
self._reader, self._writer = await asyncio.open_connection(host, port)
await self._check_zero_limits()
def connecting(self):
"""Returns True if client is connecting"""
return self._connecting_task is not None
def connected(self):
"""Returns True if client is connected"""
return self._writer is not None
async def end_session(self):
"""End session to Thermostat if connected, cancel connection if connecting"""
if self._connecting_task is not None:
self._connecting_task.cancel()
async def disconnect(self):
"""Disconnect from the Thermostat"""
if self._writer is None:
return
@ -69,26 +45,29 @@ class Client:
for pwm_channel in pwm_report:
for limit in ["max_i_neg", "max_i_pos", "max_v"]:
if pwm_channel[limit]["value"] == 0.0:
logging.warning("`{}` limit is set to zero on channel {}".format(limit, pwm_channel["channel"]))
logging.warning(
"`{}` limit is set to zero on channel {}".format(
limit, pwm_channel["channel"]
)
)
async def _read_line(self):
# read 1 line
chunk = await asyncio.wait_for(self._reader.readline(), self.timeout) # Only wait for response until timeout
return chunk.decode('utf-8', errors='ignore')
chunk = await self._reader.readline()
return chunk.decode("utf-8", errors="ignore")
async def _read_write(self, command):
self._writer.write(((" ".join(command)).strip() + "\n").encode('utf-8'))
self._writer.write(((" ".join(command)).strip() + "\n").encode("utf-8"))
await self._writer.drain()
return await self._read_line()
async def _command(self, *command):
async with self._command_lock:
# protect the read-write process from being cancelled midway
line = await asyncio.shield(self._read_write(command))
line = await self._read_write(command)
response = json.loads(line)
logging.debug(f"{command}: {response}")
logging.debug("%s: %s", command, response)
if "error" in response:
raise CommandError(response["error"])
return response
@ -239,6 +218,8 @@ class Client:
async def save_config(self, channel=""):
"""Save current configuration to EEPROM"""
await self._command("save", str(channel))
if channel == "":
await self._read_line() # Read the extra {}
async def load_config(self, channel=""):
"""Load current configuration from EEPROM"""
@ -256,24 +237,24 @@ class Client:
The client is disconnected as the TCP session is terminated.
"""
async with self._command_lock:
self._writer.write("reset\n".encode('utf-8'))
self._writer.write("reset\n".encode("utf-8"))
await self._writer.drain()
await self.end_session()
await self.disconnect()
async def dfu(self):
"""Put the Thermostat in DFU update mode
"""Put the Thermostat in DFU mode
The client is disconnected as the Thermostat stops responding to
TCP commands in DFU update mode. The only way to exit it is by
power-cycling.
TCP commands in DFU mode. To exit it, submit a DFU leave request
or power-cycle the Thermostat.
"""
async with self._command_lock:
self._writer.write("dfu\n".encode('utf-8'))
self._writer.write("dfu\n".encode("utf-8"))
await self._writer.drain()
await self.end_session()
await self.disconnect()
async def ipv4(self):
"""Get the IPv4 settings of the Thermostat"""
return await self._command('ipv4')
return await self._command("ipv4")

View File

@ -2,9 +2,11 @@ import socket
import json
import logging
class CommandError(Exception):
pass
class Client:
def __init__(self, host="192.168.1.26", port=23, timeout=None):
self._socket = socket.create_connection((host, port), timeout)
@ -20,7 +22,11 @@ class Client:
for pwm_channel in pwm_report:
for limit in ["max_i_neg", "max_i_pos", "max_v"]:
if pwm_channel[limit]["value"] == 0.0:
logging.warning("`{}` limit is set to zero on channel {}".format(limit, pwm_channel["channel"]))
logging.warning(
"`{}` limit is set to zero on channel {}".format(
limit, pwm_channel["channel"]
)
)
def _read_line(self):
# read more lines
@ -28,7 +34,7 @@ class Client:
chunk = self._socket.recv(4096)
if not chunk:
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")
line = self._lines[0]
@ -36,7 +42,7 @@ class Client:
return line
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()
response = json.loads(line)

View File

@ -1,13 +1,17 @@
from PyQt6.QtCore import QObject, pyqtSlot
from PyQt6.QtCore import QObject, pyqtSlot, pyqtSignal
from qasync import asyncSlot
from autotune import PIDAutotuneState, PIDAutotune
class PIDAutoTuner(QObject):
def __init__(self, parent, client, num_of_channel):
super().__init__()
autotune_state_changed = pyqtSignal(int, PIDAutotuneState)
def __init__(self, parent, thermostat, num_of_channel):
super().__init__(parent)
self._thermostat = thermostat
self._thermostat.report_update.connect(self.tick)
self._client = client
self.autotuners = [PIDAutotune(25) for _ in range(num_of_channel)]
self.target_temp = [20.0 for _ in range(num_of_channel)]
self.test_current = [1.0 for _ in range(num_of_channel)]
@ -15,10 +19,6 @@ class PIDAutoTuner(QObject):
self.lookback = [3.0 for _ in range(num_of_channel)]
self.sampling_interval = [1 / 16.67 for _ in range(num_of_channel)]
@pyqtSlot(list)
def update_sampling_interval(self, interval):
self.sampling_interval = interval
def set_params(self, params_name, ch, val):
getattr(self, params_name)[ch] = val
@ -34,39 +34,51 @@ class PIDAutoTuner(QObject):
self.lookback[ch],
)
self.autotuners[ch].setReady()
self.autotune_state_changed.emit(ch, self.autotuners[ch].state())
async def stop_pid_from_running(self, ch):
self.autotuners[ch].setOff()
await self._client.set_param("pwm", ch, "i_set", 0)
self.autotune_state_changed.emit(ch, self.autotuners[ch].state())
if self._thermostat.connected():
await self._thermostat.set_param("pwm", ch, "i_set", 0)
@asyncSlot(list)
async def tick(self, report):
for channel_report in report:
ch = channel_report["channel"]
self.sampling_interval[ch] = channel_report["interval"]
# TODO: Skip when PID Autotune or emit error message if NTC is not connected
if channel_report["temperature"] is None:
continue
ch = channel_report["channel"]
match self.autotuners[ch].state():
case PIDAutotuneState.STATE_READY | PIDAutotuneState.STATE_RELAY_STEP_UP | PIDAutotuneState.STATE_RELAY_STEP_DOWN:
case (
PIDAutotuneState.STATE_READY
| PIDAutotuneState.STATE_RELAY_STEP_UP
| PIDAutotuneState.STATE_RELAY_STEP_DOWN
):
self.autotuners[ch].run(
channel_report["temperature"], channel_report["time"]
)
await self._client.set_param(
await self._thermostat.set_param(
"pwm", ch, "i_set", self.autotuners[ch].output()
)
case PIDAutotuneState.STATE_SUCCEEDED:
kp, ki, kd = self.autotuners[ch].get_tec_pid()
self.autotuners[ch].setOff()
self.autotune_state_changed.emit(ch, self.autotuners[ch].state())
await self._client.set_param("pid", ch, "kp", kp)
await self._client.set_param("pid", ch, "ki", ki)
await self._client.set_param("pid", ch, "kd", kd)
await self._client.set_param("pwm", ch, "pid")
await self._thermostat.set_param("pid", ch, "kp", kp)
await self._thermostat.set_param("pid", ch, "ki", ki)
await self._thermostat.set_param("pid", ch, "kd", kd)
await self._thermostat.set_param("pwm", ch, "pid")
await self._client.set_param(
await self._thermostat.set_param(
"pid", ch, "target", self.target_temp[ch]
)
case PIDAutotuneState.STATE_FAILED:
self.autotuners[ch].setOff()
await self._client.set_param("pwm", ch, "i_set", 0)
self.autotune_state_changed.emit(ch, self.autotuners[ch].state())
await self._thermostat.set_param("pwm", ch, "i_set", 0)

View File

@ -1,111 +1,104 @@
from pytec.aioclient import Client
from PyQt6.QtCore import pyqtSignal, QObject, pyqtSlot
from qasync import asyncSlot
from pytec.gui.model.property import Property, PropertyMeta
import asyncio
import logging
from enum import Enum
from pytec.aioclient import AsyncioClient
class WrappedClient(QObject, Client):
connection_error = pyqtSignal()
async def _read_line(self):
try:
return await super()._read_line()
except (Exception, TimeoutError, asyncio.exceptions.TimeoutError):
logging.error("Client connection error, disconnecting", exc_info=True)
self.connection_error.emit()
class ThermostatConnectionState(Enum):
DISCONNECTED = "disconnected"
CONNECTING = "connecting"
CONNECTED = "connected"
class Thermostat(QObject, metaclass=PropertyMeta):
connection_state = Property(ThermostatConnectionState)
hw_rev = Property(dict)
fan = Property(dict)
thermistor = Property(list)
pid = Property(list)
pwm = Property(list)
postfilter = Property(list)
interval = Property(list)
report = Property(list)
info_box_trigger = pyqtSignal(str, str)
def __init__(self, parent, client, update_s):
self._update_s = update_s
self._client = client
self._watch_task = None
self._report_mode_task = None
self._poll_for_report = True
connection_error = pyqtSignal()
NUM_CHANNELS = 2
def __init__(self, parent, update_s, disconnect_cb=None):
super().__init__(parent)
async def run(self):
self.task = asyncio.create_task(self.update_params())
while True:
if self.task.done():
if self.task.exception() is not None:
try:
raise self.task.exception()
except (
Exception,
TimeoutError,
asyncio.exceptions.TimeoutError,
):
logging.error(
"Encountered an error while updating parameter tree.",
exc_info=True,
)
_ = self.task.result()
self.task = asyncio.create_task(self.update_params())
await asyncio.sleep(self._update_s)
self._update_s = update_s
self._client = AsyncioClient()
self._watch_task = None
self._update_params_task = None
self.disconnect_cb = disconnect_cb
self.connection_state = ThermostatConnectionState.DISCONNECTED
async def get_hw_rev(self):
async def start_session(self, host, port):
await self._client.connect(host, port)
self.hw_rev = await self._client.hw_rev()
return self.hw_rev
async def update_params(self):
self.fan = await self._client.get_fan()
self.pwm = await self._client.get_pwm()
if self._poll_for_report:
self.report = await self._client.report()
self.interval = [
self.report[i]["interval"] for i in range(len(self.report))
]
self.pid = await self._client.get_pid()
self.thermistor = await self._client.get_steinhart_hart()
self.postfilter = await self._client.get_postfilter()
@asyncSlot()
async def end_session(self):
self.stop_watching()
def connected(self):
return self._client.connected
if self.disconnect_cb is not None:
if asyncio.iscoroutinefunction(self.disconnect_cb):
await self.disconnect_cb()
else:
self.disconnect_cb()
def connecting(self):
return self._client.connecting
await self._client.disconnect()
def start_watching(self):
self._watch_task = asyncio.create_task(self.run())
@asyncSlot()
async def stop_watching(self):
def stop_watching(self):
if self._watch_task is not None:
await self.set_report_mode(False)
self._watch_task.cancel()
self._watch_task = None
self.task.cancel()
self.task = None
self._update_params_task.cancel()
self._update_params_task = None
async def set_report_mode(self, enabled: bool):
self._poll_for_report = not enabled
if enabled:
self._report_mode_task = asyncio.create_task(self.report_mode())
else:
self._client.stop_report_mode()
async def run(self):
self._update_params_task = asyncio.create_task(self.update_params())
while True:
if self._update_params_task.done():
try:
self._update_params_task.result()
except OSError:
logging.error(
"Encountered an error while polling for information from Thermostat.",
exc_info=True,
)
await self.end_session()
self.connection_state = ThermostatConnectionState.DISCONNECTED
self.connection_error.emit()
return
self._update_params_task = asyncio.create_task(self.update_params())
await asyncio.sleep(self._update_s)
async def report_mode(self):
async for report in self._client.report_mode():
self.report_update.emit(report)
self.interval = [
self.report[i]["interval"] for i in range(len(self.report))
]
async def update_params(self):
self.fan, self.pwm, self.report, self.pid, self.thermistor, self.postfilter = (
await asyncio.gather(
self._client.get_fan(),
self._client.get_pwm(),
self._client.report(),
self._client.get_pid(),
self._client.get_steinhart_hart(),
self._client.get_postfilter(),
)
)
async def end_session(self):
await self._client.end_session()
def connected(self):
return self._client.connected()
@pyqtSlot(float)
def set_update_s(self, update_s):
self._update_s = update_s
async def set_ipv4(self, ipv4):
await self._client.set_param("ipv4", ipv4)
@ -114,18 +107,12 @@ class Thermostat(QObject, metaclass=PropertyMeta):
return await self._client.ipv4()
@asyncSlot()
async def save_cfg(self, ch):
async def save_cfg(self, ch=""):
await self._client.save_config(ch)
self.info_box_trigger.emit(
"Config saved", f"Channel {ch} Config has been saved from flash."
)
@asyncSlot()
async def load_cfg(self, ch):
async def load_cfg(self, ch=""):
await self._client.load_config(ch)
self.info_box_trigger.emit(
"Config loaded", f"Channel {ch} Config has been loaded from flash."
)
async def dfu(self):
await self._client.dfu()
@ -133,6 +120,11 @@ class Thermostat(QObject, metaclass=PropertyMeta):
async def reset(self):
await self._client.reset()
@pyqtSlot(float)
def set_update_s(self, update_s):
self._update_s = update_s
async def set_fan(self, power="auto"):
await self._client.set_fan(power)
async def get_fan(self):
return await self._client.get_fan()
async def set_param(self, topic, channel, field="", value=""):
await self._client.set_param(topic, channel, field, value)

View File

@ -1,9 +1,17 @@
from PyQt6 import QtWidgets, QtCore
from PyQt6.QtCore import pyqtSlot
from pytec.gui.model.thermostat import ThermostatConnectionState
class ConnMenu(QtWidgets.QMenu):
def __init__(self):
class ConnectionDetailsMenu(QtWidgets.QMenu):
def __init__(self, thermostat, connect_btn):
super().__init__()
self._thermostat = thermostat
self._connect_btn = connect_btn
self._thermostat.connection_state_update.connect(
self.thermostat_state_change_handler
)
self.setTitle("Connection Settings")
self.host_set_line = QtWidgets.QLineEdit()
@ -13,7 +21,7 @@ class ConnMenu(QtWidgets.QMenu):
self.host_set_line.setClearButtonEnabled(True)
def connect_on_enter_press():
self.connect_btn.click()
self._connect_btn.click()
self.hide()
self.host_set_line.returnPressed.connect(connect_on_enter_press)
@ -54,3 +62,12 @@ class ConnMenu(QtWidgets.QMenu):
exit_action.setDefaultWidget(self.exit_button)
self.addAction(exit_action)
self.exit_action = exit_action
@pyqtSlot(ThermostatConnectionState)
def thermostat_state_change_handler(self, state):
self.host_set_line.setEnabled(
state == ThermostatConnectionState.DISCONNECTED
)
self.port_set_spin.setEnabled(
state == ThermostatConnectionState.DISCONNECTED
)

View File

@ -1,9 +1,12 @@
from functools import partial
from PyQt6.QtCore import pyqtSignal, QObject, QSignalBlocker, pyqtSlot
import pyqtgraph.parametertree.parameterTypes as pTypes
from pyqtgraph.parametertree import (
Parameter,
registerParameterType,
)
from qasync import asyncSlot
from autotune import PIDAutotuneState
class MutexParameter(pTypes.ListParameter):
@ -43,18 +46,20 @@ registerParameterType("mutex", MutexParameter)
class CtrlPanel(QObject):
set_zero_limits_warning_sig = pyqtSignal(list)
def __init__(
self,
thermostat,
autotuners,
info_box,
trees_ui,
param_tree,
sigTreeStateChanged_handle,
sigActivated_handles,
parent=None,
):
super().__init__(parent)
self.thermostat = thermostat
self.autotuners = autotuners
self.info_box = info_box
self.trees_ui = trees_ui
self.NUM_CHANNELS = len(trees_ui)
@ -77,10 +82,24 @@ class CtrlPanel(QObject):
tree.setHeaderHidden(True)
tree.setParameters(self.params[i], showTop=False)
self.params[i].setValue = self._setValue
self.params[i].sigTreeStateChanged.connect(sigTreeStateChanged_handle)
self.params[i].sigTreeStateChanged.connect(self.send_command)
for handle in sigActivated_handles[i]:
self.params[i].child(*handle[0]).sigActivated.connect(handle[1])
self.params[i].child("Save to flash").sigActivated.connect(
partial(self.save_settings, i)
)
self.params[i].child("Load from flash").sigActivated.connect(
partial(self.load_settings, i)
)
self.params[i].child(
"PID Config", "PID Auto Tune", "Run"
).sigActivated.connect(partial(self.pid_auto_tune_request, i))
self.thermostat.pid_update.connect(self.update_pid)
self.thermostat.report_update.connect(self.update_report)
self.thermostat.thermistor_update.connect(self.update_thermistor)
self.thermostat.pwm_update.connect(self.update_pwm)
self.thermostat.postfilter_update.connect(self.update_postfilter)
self.autotuners.autotune_state_changed.connect(self.update_pid_autotune)
def _setValue(self, value, blockSignal=None):
"""
@ -111,7 +130,45 @@ class CtrlPanel(QObject):
def change_params_title(self, channel, path, title):
self.params[channel].child(*path).setOpts(title=title)
@pyqtSlot("QVariantList")
@asyncSlot(object, object)
async def send_command(self, param, changes):
"""Translates parameter tree changes into thermostat set_param calls"""
ch = param.channel
for inner_param, change, data in changes:
if change == "value":
if inner_param.opts.get("param", None) is not None:
if inner_param.opts.get("suffix", None) == "mA":
data /= 1000 # Given in mA
thermostat_param = inner_param.opts["param"]
if thermostat_param[1] == "ch":
thermostat_param[1] = ch
if inner_param.name() == "Postfilter Rate" and data is None:
set_param_args = (*thermostat_param[:2], "off")
else:
set_param_args = (*thermostat_param, data)
param.child(*param.childPath(inner_param)).setOpts(lock=True)
await self.thermostat.set_param(*set_param_args)
param.child(*param.childPath(inner_param)).setOpts(lock=False)
if inner_param.opts.get("pid_autotune", None) is not None:
auto_tuner_param = inner_param.opts["pid_autotune"][0]
if inner_param.opts["pid_autotune"][1] != "ch":
ch = inner_param.opts["pid_autotune"][1]
self.autotuners.set_params(auto_tuner_param, ch, data)
if inner_param.opts.get("activaters", None) is not None:
activater = inner_param.opts["activaters"][
inner_param.opts["limits"].index(data)
]
if activater is not None:
if activater[1] == "ch":
activater[1] = ch
await self.thermostat.set_param(*activater)
@pyqtSlot(list)
def update_pid(self, pid_settings):
for settings in pid_settings:
channel = settings["channel"]
@ -135,7 +192,7 @@ class CtrlPanel(QObject):
"Output Config", "Control Method", "Set Temperature"
).setValue(settings["target"])
@pyqtSlot("QVariantList")
@pyqtSlot(list)
def update_report(self, report_data):
for settings in report_data:
channel = settings["channel"]
@ -155,7 +212,7 @@ class CtrlPanel(QObject):
settings["tec_i"] * 1000
)
@pyqtSlot("QVariantList")
@pyqtSlot(list)
def update_thermistor(self, sh_data):
for sh_param in sh_data:
channel = sh_param["channel"]
@ -170,10 +227,8 @@ class CtrlPanel(QObject):
sh_param["params"]["b"]
)
@pyqtSlot("QVariantList")
@pyqtSlot(list)
def update_pwm(self, pwm_data):
channels_zeroed_limits = [set() for i in range(self.NUM_CHANNELS)]
for pwm_params in pwm_data:
channel = pwm_params["channel"]
with QSignalBlocker(self.params[channel]):
@ -187,12 +242,7 @@ class CtrlPanel(QObject):
"Output Config", "Limits", "Max Heating Current"
).setValue(pwm_params["max_i_neg"]["value"] * 1000)
for limit in "max_i_pos", "max_i_neg", "max_v":
if pwm_params[limit]["value"] == 0.0:
channels_zeroed_limits[channel].add(limit)
self.set_zero_limits_warning_sig.emit(channels_zeroed_limits)
@pyqtSlot("QVariantList")
@pyqtSlot(list)
def update_postfilter(self, postfilter_data):
for postfilter_params in postfilter_data:
channel = postfilter_params["channel"]
@ -200,3 +250,61 @@ class CtrlPanel(QObject):
self.params[channel].child(
"Thermistor Config", "Postfilter Rate"
).setValue(postfilter_params["rate"])
def update_pid_autotune(self, ch, state):
match state:
case PIDAutotuneState.STATE_OFF:
self.change_params_title(
ch, ("PID Config", "PID Auto Tune", "Run"), "Run"
)
case (
PIDAutotuneState.STATE_READY
| PIDAutotuneState.STATE_RELAY_STEP_UP
| PIDAutotuneState.STATE_RELAY_STEP_DOWN
):
self.change_params_title(
ch, ("PID Config", "PID Auto Tune", "Run"), "Stop"
)
case PIDAutotuneState.STATE_SUCCEEDED:
self.info_box.display_info_box(
"PID Autotune Success",
f"Channel {ch} PID Config has been loaded to Thermostat. Regulating temperature.",
)
case PIDAutotuneState.STATE_FAILED:
self.info_box.display_info_box(
"PID Autotune Failed",
f"Channel {ch} PID Autotune has failed.",
)
@asyncSlot(int)
async def load_settings(self, ch):
await self.thermostat.load_cfg(ch)
self.info_box.display_info_box(
f"Channel {ch} settings loaded",
f"Channel {ch} settings has been loaded from flash.",
)
@asyncSlot(int)
async def save_settings(self, ch):
await self.thermostat.save_cfg(ch)
self.info_box.display_info_box(
f"Channel {ch} settings saved",
f"Channel {ch} settings has been saved to flash.\n"
"It will be loaded on Thermostat reset, or when settings are explicitly loaded.",
)
@asyncSlot()
async def pid_auto_tune_request(self, ch=0):
match self.autotuners.get_state(ch):
case PIDAutotuneState.STATE_OFF | PIDAutotuneState.STATE_FAILED:
self.autotuners.load_params_and_set_ready(ch)
case (
PIDAutotuneState.STATE_READY
| PIDAutotuneState.STATE_RELAY_STEP_UP
| PIDAutotuneState.STATE_RELAY_STEP_DOWN
):
await self.autotuners.stop_pid_from_running(ch)

View File

@ -5,13 +5,21 @@ from pglive.sources.live_plot import LiveLinePlot
from pglive.sources.live_axis import LiveAxis
from collections import deque
import pyqtgraph as pg
from pytec.gui.model.thermostat import ThermostatConnectionState
pg.setConfigOptions(antialias=True)
class LiveDataPlotter(QObject):
def __init__(self, live_plots):
def __init__(self, thermostat, live_plots):
super().__init__()
self._thermostat = thermostat
self._thermostat.report_update.connect(self.update_report)
self._thermostat.pid_update.connect(self.update_pid)
self._thermostat.connection_state_update.connect(
self.thermostat_state_change_handler
)
self.NUM_CHANNELS = len(live_plots)
self.graphs = []
@ -21,6 +29,11 @@ class LiveDataPlotter(QObject):
live_plot[1].setTitle(f"Channel {i} Current")
self.graphs.append(_TecGraphs(live_plot[0], live_plot[1]))
@pyqtSlot(ThermostatConnectionState)
def thermostat_state_change_handler(self, state):
if state == ThermostatConnectionState.DISCONNECTED:
self.clear_graphs()
def _config_connector_max_pts(self, connector, samples):
connector.max_points = samples
connector.x = deque(maxlen=int(connector.max_points))
@ -67,9 +80,7 @@ class _TecGraphs:
self._t_line = self._t_widget.getPlotItem().addLine(label="{value} °C")
self._t_line.setVisible(False)
# Hack for keeping setpoint line in plot range
self._t_setpoint_plot = (
LiveLinePlot()
)
self._t_setpoint_plot = LiveLinePlot()
for graph in t_widget, i_widget:
time_axis = LiveAxis(

View File

@ -2,18 +2,22 @@ from PyQt6 import QtWidgets, QtGui
class PlotOptionsMenu(QtWidgets.QMenu):
def __init__(self, max_samples=1000):
def __init__(self, channel_graphs, max_samples=1000):
super().__init__()
self.channel_graphs = channel_graphs
self.setTitle("Plot Settings")
clear = QtGui.QAction("Clear graphs", self)
self.addAction(clear)
self.clear = clear
self.clear.triggered.connect(self.channel_graphs.clear_graphs)
self.samples_spinbox = QtWidgets.QSpinBox()
self.samples_spinbox.setRange(2, 100000)
self.samples_spinbox.setSuffix(" samples")
self.samples_spinbox.setValue(max_samples)
self.samples_spinbox.valueChanged.connect(self.channel_graphs.set_max_samples)
limit_samples = QtWidgets.QWidgetAction(self)
limit_samples.setDefaultWidget(self.samples_spinbox)

View File

@ -369,7 +369,7 @@
<number>0</number>
</property>
<item>
<layout class="QHBoxLayout" name="report_layout" stretch="0,1,1,1">
<layout class="QHBoxLayout" name="report_layout" stretch="0,1,1">
<property name="spacing">
<number>6</number>
</property>
@ -435,31 +435,6 @@
</property>
</widget>
</item>
<item>
<widget class="QCheckBox" name="report_box">
<property name="sizePolicy">
<sizepolicy hsizetype="Fixed" vsizetype="Expanding">
<horstretch>0</horstretch>
<verstretch>0</verstretch>
</sizepolicy>
</property>
<property name="maximumSize">
<size>
<width>80</width>
<height>16777215</height>
</size>
</property>
<property name="baseSize">
<size>
<width>80</width>
<height>0</height>
</size>
</property>
<property name="text">
<string>Report</string>
</property>
</widget>
</item>
<item>
<widget class="QPushButton" name="report_apply_btn">
<property name="sizePolicy">

View File

@ -1,24 +1,24 @@
import logging
from PyQt6 import QtWidgets, QtGui, QtCore
from PyQt6.QtCore import pyqtSignal, pyqtSlot
from PyQt6.QtCore import pyqtSignal, pyqtSlot, QSignalBlocker
from qasync import asyncSlot
from pytec.gui.view.net_settings_input_diag import NetSettingsInputDiag
from pytec.gui.model.thermostat import ThermostatConnectionState
class ThermostatCtrlMenu(QtWidgets.QMenu):
fan_set_act = pyqtSignal(int)
fan_auto_set_act = pyqtSignal(int)
connect_act = pyqtSignal()
reset_act = pyqtSignal(bool)
dfu_act = pyqtSignal(bool)
load_cfg_act = pyqtSignal(int)
save_cfg_act = pyqtSignal(int)
net_cfg_act = pyqtSignal(bool)
def __init__(self, style):
class ThermostatSettingsMenu(QtWidgets.QMenu):
def __init__(self, thermostat, info_box, style):
super().__init__()
self._thermostat = thermostat
self._info_box = info_box
self._style = style
self.setTitle("Thermostat settings")
self.hw_rev_data = dict()
self._thermostat.hw_rev_update.connect(self.hw_rev)
self._thermostat.connection_state_update.connect(
self.thermostat_state_change_handler
)
self.fan_group = QtWidgets.QWidget()
self.fan_group.setEnabled(False)
@ -45,8 +45,9 @@ class ThermostatCtrlMenu(QtWidgets.QMenu):
self.fan_pwm_warning.setMinimumSize(QtCore.QSize(16, 0))
self.fan_layout.addWidget(self.fan_pwm_warning)
self.fan_power_slider.valueChanged.connect(self.fan_set_act)
self.fan_auto_box.stateChanged.connect(self.fan_auto_set_act)
self.fan_power_slider.valueChanged.connect(self.fan_set_request)
self.fan_auto_box.stateChanged.connect(self.fan_auto_set_request)
self._thermostat.fan_update.connect(self.fan_update)
self.fan_lbl.setToolTip("Adjust the fan")
self.fan_lbl.setText("Fan:")
@ -58,40 +59,36 @@ class ThermostatCtrlMenu(QtWidgets.QMenu):
self.fan = fan
self.actionReset = QtGui.QAction("Reset Thermostat", self)
self.actionReset.triggered.connect(self.reset_act)
self.actionReset.triggered.connect(self.reset_request)
self.addAction(self.actionReset)
self.actionEnter_DFU_Mode = QtGui.QAction("Enter DFU Mode", self)
self.actionEnter_DFU_Mode.triggered.connect(self.dfu_act)
self.actionEnter_DFU_Mode.triggered.connect(self.dfu_request)
self.addAction(self.actionEnter_DFU_Mode)
self.actionnet_settings_input_diag = QtGui.QAction("Set IPV4 Settings", self)
self.actionnet_settings_input_diag.triggered.connect(self.net_cfg_act)
self.actionnet_settings_input_diag.triggered.connect(self.net_settings_request)
self.addAction(self.actionnet_settings_input_diag)
@pyqtSlot(bool)
def load(_):
self.load_cfg_act.emit(0)
self.load_cfg_act.emit(1)
loaded = QtWidgets.QMessageBox(self)
loaded.setWindowTitle("Config loaded")
loaded.setText("All channel configs have been loaded from flash.")
loaded.setIcon(QtWidgets.QMessageBox.Icon.Information)
loaded.show()
@asyncSlot(bool)
async def load(_):
await self._thermostat.load_cfg()
self._info_box.display_info_box(
"Config loaded", "All channel configs have been loaded from flash."
)
self.actionLoad_all_configs = QtGui.QAction("Load Config", self)
self.actionLoad_all_configs.triggered.connect(load)
self.addAction(self.actionLoad_all_configs)
@pyqtSlot(bool)
def save(_):
self.save_cfg_act.emit(0)
self.save_cfg_act.emit(1)
saved = QtWidgets.QMessageBox(self)
saved.setWindowTitle("Config saved")
saved.setText("All channel configs have been saved to flash.")
saved.setIcon(QtWidgets.QMessageBox.Icon.Information)
saved.show()
@asyncSlot(bool)
async def save(_):
await self._thermostat.save_cfg()
self._info_box.display_info_box(
"Config saved", "All channel configs have been saved to flash."
)
self.actionSave_all_configs = QtGui.QAction("Save Config", self)
self.actionSave_all_configs.triggered.connect(save)
@ -127,6 +124,18 @@ class ThermostatCtrlMenu(QtWidgets.QMenu):
self.actionAbout_Thermostat.triggered.connect(about_thermostat)
self.addAction(self.actionAbout_Thermostat)
@pyqtSlot(dict)
def fan_update(self, fan_settings):
logging.debug(fan_settings)
if fan_settings is None:
return
with QSignalBlocker(self.fan_power_slider):
self.fan_power_slider.setValue(
fan_settings["fan_pwm"] or 100 # 0 = PWM off = full strength
)
with QSignalBlocker(self.fan_auto_box):
self.fan_auto_box.setChecked(fan_settings["auto_mode"])
def set_fan_pwm_warning(self):
if self.fan_power_slider.value() != 100:
pixmapi = getattr(QtWidgets.QStyle.StandardPixmap, "SP_MessageBoxWarning")
@ -139,7 +148,68 @@ class ThermostatCtrlMenu(QtWidgets.QMenu):
self.fan_pwm_warning.setPixmap(QtGui.QPixmap())
self.fan_pwm_warning.setToolTip("")
@pyqtSlot("QVariantMap")
@pyqtSlot(ThermostatConnectionState)
def thermostat_state_change_handler(self, state):
if state == ThermostatConnectionState.DISCONNECTED:
self.fan_pwm_warning.setPixmap(QtGui.QPixmap())
self.fan_pwm_warning.setToolTip("")
@pyqtSlot(dict)
def hw_rev(self, hw_rev):
self.hw_rev_data = hw_rev
self.fan_group.setEnabled(self.hw_rev_data["settings"]["fan_available"])
@asyncSlot(int)
async def fan_set_request(self, value):
assert self._thermostat.connected()
if self.fan_auto_box.isChecked():
with QSignalBlocker(self.fan_auto_box):
self.fan_auto_box.setChecked(False)
await self._thermostat.set_fan(value)
if not self.hw_rev_data["settings"]["fan_pwm_recommended"]:
self.set_fan_pwm_warning()
@asyncSlot(int)
async def fan_auto_set_request(self, enabled):
assert self._thermostat.connected()
if enabled:
await self._thermostat.set_fan("auto")
self.fan_update(await self._thermostat.get_fan())
else:
await self.thermostat.set_fan(
self.fan_power_slider.value()
)
@asyncSlot(bool)
async def reset_request(self, _):
assert self._thermostat.connected()
await self._thermostat.reset()
await self._thermostat.end_session()
self._thermostat.connection_state = ThermostatConnectionState.DISCONNECTED
@asyncSlot(bool)
async def dfu_request(self, _):
assert self._thermostat.connected()
await self._thermostat.dfu()
await self._thermostat.end_session()
self._thermostat.connection_state = ThermostatConnectionState.DISCONNECTED
@asyncSlot(bool)
async def net_settings_request(self, _):
assert self._thermostat.connected()
ipv4 = await self._thermostat.get_ipv4()
self.net_settings_input_diag = NetSettingsInputDiag(ipv4["addr"])
self.net_settings_input_diag.set_ipv4_act.connect(self.set_net_settings_request)
@asyncSlot(str)
async def set_net_settings_request(self, ipv4_settings):
assert self._thermostat.connected()
await self._thermostat.set_ipv4(ipv4_settings)
await self._thermostat.end_session()
self._thermostat.connection_state = ThermostatConnectionState.DISCONNECTED

View File

@ -67,16 +67,30 @@ class QtWaitingSpinner(QWidget):
painter.setPen(Qt.PenStyle.NoPen)
for i in range(0, self._numberOfLines):
painter.save()
painter.translate(self._innerRadius + self._lineLength, self._innerRadius + self._lineLength)
painter.translate(
self._innerRadius + self._lineLength,
self._innerRadius + self._lineLength,
)
rotateAngle = float(360 * i) / float(self._numberOfLines)
painter.rotate(rotateAngle)
painter.translate(self._innerRadius, 0)
distance = self.lineCountDistanceFromPrimary(i, self._currentCounter, self._numberOfLines)
color = self.currentLineColor(distance, self._numberOfLines, self._trailFadePercentage,
self._minimumTrailOpacity, self._color)
distance = self.lineCountDistanceFromPrimary(
i, self._currentCounter, self._numberOfLines
)
color = self.currentLineColor(
distance,
self._numberOfLines,
self._trailFadePercentage,
self._minimumTrailOpacity,
self._color,
)
painter.setBrush(color)
painter.drawRoundedRect(QRect(0, int(-self._lineWidth / 2), self._lineLength, self._lineWidth), self._roundness,
self._roundness, Qt.SizeMode.RelativeSize)
painter.drawRoundedRect(
QRect(0, int(-self._lineWidth / 2), self._lineLength, self._lineWidth),
self._roundness,
self._roundness,
Qt.SizeMode.RelativeSize,
)
painter.restore()
def start(self):
@ -160,7 +174,9 @@ class QtWaitingSpinner(QWidget):
self.setFixedSize(self.size, self.size)
def updateTimer(self):
self._timer.setInterval(int(1000 / (self._numberOfLines * self._revolutionsPerSecond)))
self._timer.setInterval(
int(1000 / (self._numberOfLines * self._revolutionsPerSecond))
)
def lineCountDistanceFromPrimary(self, current, primary, totalNrOfLines):
distance = primary - current
@ -168,7 +184,9 @@ class QtWaitingSpinner(QWidget):
distance += totalNrOfLines
return distance
def currentLineColor(self, countDistance, totalNrOfLines, trailFadePerc, minOpacity, colorinput):
def currentLineColor(
self, countDistance, totalNrOfLines, trailFadePerc, minOpacity, colorinput
):
color = QColor(colorinput)
if countDistance == 0:
return color
@ -186,7 +204,7 @@ class QtWaitingSpinner(QWidget):
return color
if __name__ == '__main__':
if __name__ == "__main__":
app = QApplication([])
waiting_spinner = QtWaitingSpinner()
waiting_spinner.show()

View File

@ -3,15 +3,24 @@ from PyQt6 import QtWidgets, QtGui
class ZeroLimitsWarningView(QObject):
def __init__(self, style, limit_warning):
def __init__(self, thermostat, style, limit_warning):
super().__init__()
self._thermostat = thermostat
self._thermostat.pwm_update.connect(self.set_limits_warning)
self._lbl = limit_warning
self._style = style
@pyqtSlot("QVariantList")
def set_limits_warning(self, channels_zeroed_limits: list):
channel_disabled = [False, False]
@pyqtSlot(list)
def set_limits_warning(self, pwm_data: list):
channels_zeroed_limits = [set() for i in range(self._thermostat.NUM_CHANNELS)]
for pwm_params in pwm_data:
channel = pwm_params["channel"]
for limit in "max_i_pos", "max_i_neg", "max_v":
if pwm_params[limit]["value"] == 0.0:
channels_zeroed_limits[channel].add(limit)
channel_disabled = [False, False]
report_str = "The following output limit(s) are set to zero:\n"
for ch, zeroed_limits in enumerate(channels_zeroed_limits):
if {"max_i_pos", "max_i_neg"}.issubset(zeroed_limits):

459
pytec/tec_qt.py Normal file → Executable file
View File

@ -1,26 +1,24 @@
from pytec.gui.view.zero_limits_warning import ZeroLimitsWarningView
from pytec.gui.view.net_settings_input_diag import NetSettingsInputDiag
from pytec.gui.view.thermostat_ctrl_menu import ThermostatCtrlMenu
from pytec.gui.view.conn_menu import ConnMenu
"""GUI for the Sinara 8451 Thermostat"""
import json
import asyncio
import logging
import argparse
import importlib.resources
import qasync
from qasync import asyncSlot, asyncClose
from autotune import PIDAutotuneState
from PyQt6 import QtWidgets, QtGui, uic
from PyQt6.QtCore import pyqtSlot
from pytec.gui.model.thermostat import Thermostat, ThermostatConnectionState
from pytec.gui.model.pid_autotuner import PIDAutoTuner
from pytec.gui.view.zero_limits_warning_view import ZeroLimitsWarningView
from pytec.gui.view.thermostat_settings_menu import ThermostatSettingsMenu
from pytec.gui.view.connection_details_menu import ConnectionDetailsMenu
from pytec.gui.view.plot_options_menu import PlotOptionsMenu
from pytec.gui.view.live_plot_view import LiveDataPlotter
from pytec.gui.view.ctrl_panel import CtrlPanel
from pytec.gui.view.info_box import InfoBox
from pytec.gui.model.pid_autotuner import PIDAutoTuner
from pytec.gui.model.thermostat import WrappedClient, Thermostat
import json
from autotune import PIDAutotuneState
from qasync import asyncSlot, asyncClose
import qasync
from pytec.aioclient import StoppedConnecting
import asyncio
import logging
import argparse
from PyQt6 import QtWidgets, QtGui, uic
from PyQt6.QtCore import QSignalBlocker, pyqtSlot
import pyqtgraph as pg
from functools import partial
import importlib.resources
def get_argparser():
@ -30,9 +28,9 @@ def get_argparser():
"--connect",
default=None,
action="store_true",
help="Automatically connect to the specified Thermostat in IP:port format",
help="Automatically connect to the specified Thermostat in host:port format",
)
parser.add_argument("IP", metavar="ip", default=None, nargs="?")
parser.add_argument("HOST", metavar="host", default=None, nargs="?")
parser.add_argument("PORT", metavar="port", default=None, nargs="?")
parser.add_argument(
"-l",
@ -55,353 +53,169 @@ class MainWindow(QtWidgets.QMainWindow):
NUM_CHANNELS = 2
def __init__(self, args):
super(MainWindow, self).__init__()
super().__init__()
ui_file_path = importlib.resources.files("pytec.gui.view").joinpath("tec_qt.ui")
uic.loadUi(ui_file_path, self)
self.hw_rev_data = None
self.info_box = InfoBox()
self._info_box = InfoBox()
self.client = WrappedClient(self)
self.client.connection_error.connect(self.bail)
self.thermostat = Thermostat(
self, self.client, self.report_refresh_spin.value()
# Models
self._thermostat = Thermostat(self, self.report_refresh_spin.value())
self._connecting_task = None
self._thermostat.connection_state_update.connect(
self._on_connection_state_changed
)
self.autotuners = PIDAutoTuner(self, self.client, 2)
self._autotuners = PIDAutoTuner(self, self._thermostat, 2)
self._autotuners.autotune_state_changed.connect(
self._on_pid_autotune_state_changed
)
# Handlers for disconnections
async def autotune_disconnect():
for ch in range(self.NUM_CHANNELS):
if self._autotuners.get_state(ch) != PIDAutotuneState.STATE_OFF:
await self._autotuners.stop_pid_from_running(ch)
self._thermostat.disconnect_cb = autotune_disconnect
@pyqtSlot()
def handle_connection_error():
self._info_box.display_info_box(
"Connection Error", "Thermostat connection lost. Is it unplugged?"
)
self._thermostat.connection_error.connect(handle_connection_error)
# Control Panel
def get_ctrl_panel_config(args):
with open(args.param_tree, "r") as f:
with open(args.param_tree, "r", encoding="utf-8") as f:
return json.load(f)["ctrl_panel"]
param_tree_sigActivated_handles = [
[
[["Save to flash"], partial(self.thermostat.save_cfg, ch)],
[["Load from flash"], partial(self.thermostat.load_cfg, ch)],
[
["PID Config", "PID Auto Tune", "Run"],
partial(self.pid_auto_tune_request, ch),
],
]
for ch in range(self.NUM_CHANNELS)
]
self.thermostat.info_box_trigger.connect(self.info_box.display_info_box)
self.zero_limits_warning = ZeroLimitsWarningView(
self.style(), self.limits_warning
)
self.ctrl_panel_view = CtrlPanel(
self._ctrl_panel_view = CtrlPanel(
self._thermostat,
self._autotuners,
self._info_box,
[self.ch0_tree, self.ch1_tree],
get_ctrl_panel_config(args),
self.send_command,
param_tree_sigActivated_handles,
)
self.ctrl_panel_view.set_zero_limits_warning_sig.connect(
self.zero_limits_warning.set_limits_warning
)
self.thermostat.fan_update.connect(self.fan_update)
self.thermostat.report_update.connect(self.ctrl_panel_view.update_report)
self.thermostat.report_update.connect(self.autotuners.tick)
self.thermostat.report_update.connect(self.pid_autotune_handler)
self.thermostat.pid_update.connect(self.ctrl_panel_view.update_pid)
self.thermostat.pwm_update.connect(self.ctrl_panel_view.update_pwm)
self.thermostat.thermistor_update.connect(
self.ctrl_panel_view.update_thermistor
)
self.thermostat.postfilter_update.connect(
self.ctrl_panel_view.update_postfilter
)
self.thermostat.interval_update.connect(
self.autotuners.update_sampling_interval
)
self.report_apply_btn.clicked.connect(
lambda: self.thermostat.set_update_s(self.report_refresh_spin.value())
)
self.channel_graphs = LiveDataPlotter(
# Graphs
self._channel_graphs = LiveDataPlotter(
self._thermostat,
[
[getattr(self, f"ch{ch}_t_graph"), getattr(self, f"ch{ch}_i_graph")]
for ch in range(self.NUM_CHANNELS)
]
],
)
self.thermostat.report_update.connect(self.channel_graphs.update_report)
self.thermostat.pid_update.connect(self.channel_graphs.update_pid)
self.plot_options_menu = PlotOptionsMenu()
self.plot_options_menu.clear.triggered.connect(self.clear_graphs)
self.plot_options_menu.samples_spinbox.valueChanged.connect(
self.channel_graphs.set_max_samples
# Bottom bar menus
self.connection_details_menu = ConnectionDetailsMenu(
self._thermostat, self.connect_btn
)
self.plot_settings.setMenu(self.plot_options_menu)
self.connect_btn.setMenu(self.connection_details_menu)
self.conn_menu = ConnMenu()
self.connect_btn.setMenu(self.conn_menu)
self._thermostat_settings_menu = ThermostatSettingsMenu(
self._thermostat, self._info_box, self.style()
)
self.thermostat_settings.setMenu(self._thermostat_settings_menu)
self.thermostat_ctrl_menu = ThermostatCtrlMenu(self.style())
self.thermostat_ctrl_menu.fan_set_act.connect(self.fan_set_request)
self.thermostat_ctrl_menu.fan_auto_set_act.connect(self.fan_auto_set_request)
self.thermostat_ctrl_menu.reset_act.connect(self.reset_request)
self.thermostat_ctrl_menu.dfu_act.connect(self.dfu_request)
self.thermostat_ctrl_menu.save_cfg_act.connect(self.save_cfg_request)
self.thermostat_ctrl_menu.load_cfg_act.connect(self.load_cfg_request)
self.thermostat_ctrl_menu.net_cfg_act.connect(self.net_settings_request)
self.thermostat.hw_rev_update.connect(self.thermostat_ctrl_menu.hw_rev)
self.thermostat_settings.setMenu(self.thermostat_ctrl_menu)
self._plot_options_menu = PlotOptionsMenu(self._channel_graphs)
self.plot_settings.setMenu(self._plot_options_menu)
# Status line
self._zero_limits_warning_view = ZeroLimitsWarningView(
self._thermostat, self.style(), self.limits_warning
)
self.loading_spinner.hide()
if args.connect:
if args.IP:
self.host_set_line.setText(args.IP)
if args.PORT:
self.port_set_spin.setValue(int(args.PORT))
self.connect_btn.click()
def clear_graphs(self):
self.channel_graphs.clear_graphs()
async def _on_connection_changed(self, result):
self.graph_group.setEnabled(result)
self.report_group.setEnabled(result)
self.thermostat_settings.setEnabled(result)
self.conn_menu.host_set_line.setEnabled(not result)
self.conn_menu.port_set_spin.setEnabled(not result)
self.connect_btn.setText("Disconnect" if result else "Connect")
if result:
self.hw_rev_data = await self.thermostat.get_hw_rev()
logging.debug(self.hw_rev_data)
self._status(self.hw_rev_data)
self.thermostat.start_watching()
else:
self.status_lbl.setText("Disconnected")
self.background_task_lbl.setText("Ready.")
self.loading_spinner.hide()
self.loading_spinner.stop()
self.thermostat_ctrl_menu.fan_pwm_warning.setPixmap(QtGui.QPixmap())
self.thermostat_ctrl_menu.fan_pwm_warning.setToolTip("")
self.clear_graphs()
self.report_box.setChecked(False)
if not Thermostat.connecting or Thermostat.connected:
for ch in range(self.NUM_CHANNELS):
if self.autotuners.get_state(ch) != PIDAutotuneState.STATE_OFF:
await self.autotuners.stop_pid_from_running(ch)
await self.thermostat.set_report_mode(False)
self.thermostat.stop_watching()
def _status(self, hw_rev_d: dict):
logging.debug(hw_rev_d)
self.status_lbl.setText(
f"Connected to Thermostat v{hw_rev_d['rev']['major']}.{hw_rev_d['rev']['minor']}"
self.report_apply_btn.clicked.connect(
lambda: self._thermostat.set_update_s(self.report_refresh_spin.value())
)
@pyqtSlot("QVariantMap")
def fan_update(self, fan_settings):
logging.debug(fan_settings)
if fan_settings is None:
return
with QSignalBlocker(self.thermostat_ctrl_menu.fan_power_slider):
self.thermostat_ctrl_menu.fan_power_slider.setValue(
fan_settings["fan_pwm"] or 100 # 0 = PWM off = full strength
)
with QSignalBlocker(self.thermostat_ctrl_menu.fan_auto_box):
self.thermostat_ctrl_menu.fan_auto_box.setChecked(fan_settings["auto_mode"])
if not self.hw_rev_data["settings"]["fan_pwm_recommended"]:
self.thermostat_ctrl_menu.set_fan_pwm_warning()
@asyncSlot(int)
async def on_report_box_stateChanged(self, enabled):
await self.thermostat.set_report_mode(enabled)
@asyncClose
async def closeEvent(self, event):
async def closeEvent(self, _event):
try:
await self.bail()
await self._thermostat.end_session()
self._thermostat.connection_state = ThermostatConnectionState.DISCONNECTED
except:
pass
@asyncSlot()
async def on_connect_btn_clicked(self):
host, port = (
self.conn_menu.host_set_line.text(),
self.conn_menu.port_set_spin.value(),
@pyqtSlot(ThermostatConnectionState)
def _on_connection_state_changed(self, state):
self.graph_group.setEnabled(state == ThermostatConnectionState.CONNECTED)
self.thermostat_settings.setEnabled(
state == ThermostatConnectionState.CONNECTED
)
try:
if not (self.client.connecting() or self.client.connected()):
self.status_lbl.setText("Connecting...")
self.report_group.setEnabled(state == ThermostatConnectionState.CONNECTED)
match state:
case ThermostatConnectionState.CONNECTED:
self.connect_btn.setText("Disconnect")
self.status_lbl.setText(
"Connected to Thermostat v"
f"{self._thermostat.hw_rev['rev']['major']}."
f"{self._thermostat.hw_rev['rev']['minor']}"
)
case ThermostatConnectionState.CONNECTING:
self.connect_btn.setText("Stop")
self.conn_menu.host_set_line.setEnabled(False)
self.conn_menu.port_set_spin.setEnabled(False)
self.status_lbl.setText("Connecting...")
try:
await self.client.start_session(host=host, port=port, timeout=5)
except StoppedConnecting:
return
await self._on_connection_changed(True)
else:
await self.bail()
case ThermostatConnectionState.DISCONNECTED:
self.connect_btn.setText("Connect")
self.status_lbl.setText("Disconnected")
# TODO: Remove asyncio.TimeoutError in Python 3.11
except (OSError, TimeoutError, asyncio.TimeoutError):
try:
await self.bail()
except ConnectionResetError:
pass
@asyncSlot()
async def bail(self):
await self._on_connection_changed(False)
await self.client.end_session()
@asyncSlot(object, object)
async def send_command(self, param, changes):
"""Translates parameter tree changes into thermostat set_param calls"""
ch = param.channel
for inner_param, change, data in changes:
if change == "value":
if inner_param.opts.get("param", None) is not None:
if inner_param.opts.get("suffix", None) == "mA":
data /= 1000 # Given in mA
thermostat_param = inner_param.opts["param"]
if thermostat_param[1] == "ch":
thermostat_param[1] = ch
if inner_param.name() == "Postfilter Rate" and data is None:
set_param_args = (*thermostat_param[:2], "off")
else:
set_param_args = (*thermostat_param, data)
param.child(*param.childPath(inner_param)).setOpts(lock=True)
await self.client.set_param(*set_param_args)
param.child(*param.childPath(inner_param)).setOpts(lock=False)
if inner_param.opts.get("pid_autotune", None) is not None:
auto_tuner_param = inner_param.opts["pid_autotune"][0]
if inner_param.opts["pid_autotune"][1] != "ch":
ch = inner_param.opts["pid_autotune"][1]
self.autotuners.set_params(auto_tuner_param, ch, data)
if inner_param.opts.get("activaters", None) is not None:
activater = inner_param.opts["activaters"][
inner_param.opts["limits"].index(data)
]
if activater is not None:
if activater[1] == "ch":
activater[1] = ch
await self.client.set_param(*activater)
@asyncSlot()
async def pid_auto_tune_request(self, ch=0):
match self.autotuners.get_state(ch):
case PIDAutotuneState.STATE_OFF | PIDAutotuneState.STATE_FAILED:
self.autotuners.load_params_and_set_ready(ch)
case PIDAutotuneState.STATE_READY | PIDAutotuneState.STATE_RELAY_STEP_UP | PIDAutotuneState.STATE_RELAY_STEP_DOWN:
await self.autotuners.stop_pid_from_running(ch)
# To Update the UI elements
self.pid_autotune_handler([])
@asyncSlot(list)
async def pid_autotune_handler(self, _):
ch_tuning = []
@pyqtSlot(int, PIDAutotuneState)
def _on_pid_autotune_state_changed(self, _ch, _state):
autotuning_channels = []
for ch in range(self.NUM_CHANNELS):
match self.autotuners.get_state(ch):
case PIDAutotuneState.STATE_OFF:
self.ctrl_panel_view.change_params_title(
ch, ("PID Config", "PID Auto Tune", "Run"), "Run"
)
case PIDAutotuneState.STATE_READY | PIDAutotuneState.STATE_RELAY_STEP_UP | PIDAutotuneState.STATE_RELAY_STEP_DOWN:
self.ctrl_panel_view.change_params_title(
ch, ("PID Config", "PID Auto Tune", "Run"), "Stop"
)
ch_tuning.append(ch)
if self._autotuners.get_state(ch) in {
PIDAutotuneState.STATE_READY,
PIDAutotuneState.STATE_RELAY_STEP_UP,
PIDAutotuneState.STATE_RELAY_STEP_DOWN,
}:
autotuning_channels.append(ch)
case PIDAutotuneState.STATE_SUCCEEDED:
self.info_box.display_info_box(
"PID Autotune Success",
f"Channel {ch} PID Config has been loaded to Thermostat. Regulating temperature.",
)
self.info_box.show()
case PIDAutotuneState.STATE_FAILED:
self.info_box.display_info_box(
"PID Autotune Failed", f"Channel {ch} PID Autotune has failed."
)
self.info_box.show()
if len(ch_tuning) == 0:
if len(autotuning_channels) == 0:
self.background_task_lbl.setText("Ready.")
self.loading_spinner.hide()
self.loading_spinner.stop()
else:
self.background_task_lbl.setText(
"Autotuning channel {ch}...".format(ch=ch_tuning)
f"Autotuning channel {autotuning_channels}..."
)
self.loading_spinner.start()
self.loading_spinner.show()
@asyncSlot(int)
async def fan_set_request(self, value):
if not self.client.connected():
return
if self.thermostat_ctrl_menu.fan_auto_box.isChecked():
with QSignalBlocker(self.thermostat_ctrl_menu.fan_auto_box):
self.thermostat_ctrl_menu.fan_auto_box.setChecked(False)
await self.client.set_fan(value)
if not self.hw_rev_data["settings"]["fan_pwm_recommended"]:
self.thermostat_ctrl_menu.set_fan_pwm_warning()
@asyncSlot()
async def on_connect_btn_clicked(self):
match self._thermostat.connection_state:
case ThermostatConnectionState.DISCONNECTED:
self._connecting_task = asyncio.current_task()
self._thermostat.connection_state = ThermostatConnectionState.CONNECTING
await self._thermostat.start_session(
host=self.connection_details_menu.host_set_line.text(),
port=self.connection_details_menu.port_set_spin.value(),
)
self._connecting_task = None
self._thermostat.connection_state = ThermostatConnectionState.CONNECTED
self._thermostat.start_watching()
@asyncSlot(int)
async def fan_auto_set_request(self, enabled):
if not self.client.connected():
return
if enabled:
await self.client.set_fan("auto")
self.fan_update(await self.client.get_fan())
else:
await self.client.set_fan(
self.thermostat_ctrl_menu.fan_power_slider.value()
case ThermostatConnectionState.CONNECTING:
self._connecting_task.cancel()
self._connecting_task = None
await self._thermostat.end_session()
self._thermostat.connection_state = (
ThermostatConnectionState.DISCONNECTED
)
@asyncSlot(int)
async def save_cfg_request(self, ch):
await self.thermostat.save_cfg(str(ch))
@asyncSlot(int)
async def load_cfg_request(self, ch):
await self.thermostat.load_cfg(str(ch))
@asyncSlot(bool)
async def dfu_request(self, _):
await self._on_connection_changed(False)
await self.thermostat.dfu()
@asyncSlot(bool)
async def reset_request(self, _):
await self._on_connection_changed(False)
await self.thermostat.reset()
await asyncio.sleep(0.1) # Wait for the reset to start
self.connect_btn.click() # Reconnect
@asyncSlot(bool)
async def net_settings_request(self, _):
ipv4 = await self.thermostat.get_ipv4()
self.net_settings_input_diag = NetSettingsInputDiag(ipv4["addr"])
self.net_settings_input_diag.set_ipv4_act.connect(self.set_net_settings_request)
@asyncSlot(str)
async def set_net_settings_request(self, ipv4_settings):
await self.thermostat.set_ipv4(ipv4_settings)
await self.thermostat._client.end_session()
await self._on_connection_changed(False)
case ThermostatConnectionState.CONNECTED:
await self._thermostat.end_session()
self._thermostat.connection_state = (
ThermostatConnectionState.DISCONNECTED
)
async def coro_main():
@ -414,12 +228,21 @@ async def coro_main():
app = QtWidgets.QApplication.instance()
app.aboutToQuit.connect(app_quit_event.set)
app.setWindowIcon(
QtGui.QIcon(str(importlib.resources.files("pytec.gui.resources").joinpath("artiq.ico")))
QtGui.QIcon(
str(importlib.resources.files("pytec.gui.resources").joinpath("artiq.ico"))
)
)
main_window = MainWindow(args)
main_window.show()
if args.connect:
if args.HOST:
main_window.connection_details_menu.host_set_line.setText(args.HOST)
if args.PORT:
main_window.connection_details_menu.port_set_spin.setValue(int(args.PORT))
main_window.connect_btn.click()
await app_quit_event.wait()