fix return-value/deprecated API bugs, accumulation lock leaks, code cleanup
Build Debian Packages / build (bookworm, debian12) (push) Successful in 8m56s
Build Debian Packages / build (trixie, debian13) (push) Successful in 9m3s

- raise instead of silent None for unsupported operands and empty negation
- validate before acquiring lock in Accumulation operators (no RLock leak)
- Accumulation._copy_provenance helper, ADC_Result._copy_metadata for all result paths
- shared _sandbox_stdout_redirect contextmanager for sandboxed scripts
- TemperatureMonitor dead code removal (gtk helper, traceback, unused properties)
- new unit tests incl. lock-freedom regression test
- gitignore sphinx generated doc/_autosummary/ and doc/damaris/
This commit is contained in:
markusro committed 2026-10-17 19:17:43 +02:00
1 parent 91ade4376c
commit f9d8de7697
26 files changed
+590 -236

No files matched your search

+24 -14
View File
@@ -1,5 +1,6 @@
import serial
import re
import time
import operator
from functools import reduce
@@ -57,21 +58,22 @@ class Eurotherm(TemperatureController):
def send_read_param(self, param):
self.s.write(EOT + self.device + param + ENQ)
def read_param(self, param):
self.s.flushInput()
self.send_read_param(param)
answer = self.s.read(self._expect_len)
m = reply_pattern.search(answer)
if m is None:
# Reading _expect_len bytes was not enough...
answer += self.s.read(200)
def read_param(self, param, retries=3, retry_delay=0.05):
for attempt in range(retries):
self.s.flushInput()
self.send_read_param(param)
answer = self.s.read(self._expect_len)
m = reply_pattern.search(answer)
if m is not None:
self._expect_len = len(answer)
return m.group(1)
else:
print("received:", repr(answer))
return None
if m is None:
answer += self.s.read(200)
m = reply_pattern.search(answer)
if m is not None:
self._expect_len = len(answer)
return m.group(1)
if attempt < retries - 1:
time.sleep(retry_delay)
print("received:", repr(answer))
return None
def write_param(self, mnemonic, data):
if len(mnemonic) > 2:
@@ -128,6 +130,14 @@ class Eurotherm(TemperatureController):
"""Set target temperature in Celsius."""
return self.write_param('SL', str(temperature))
def check_connection(self) -> bool:
"""Check if the Eurotherm device is connected and responsive."""
try:
result = self.read_param('PV')
return result is not None
except Exception:
return False
def close(self):
"""Close the serial connection."""
if hasattr(self, 's') and self.s is not None: