Public Access
added POC for temperature logging
This commit is contained in:
@@ -15,17 +15,60 @@ from .Drawable import Drawable
|
||||
class TemperatureResult(Resultable, Drawable):
|
||||
"""
|
||||
Specialised class of Resultable and Drawable
|
||||
Contains recorded temperature data
|
||||
Contains recorded temperature data.
|
||||
|
||||
Attributes:
|
||||
x: List of timestamps in seconds.
|
||||
y: List of temperature readings in Celsius.
|
||||
setpoint: Optional list of setpoint temperatures in Celsius.
|
||||
xlabel: Label for x-axis (default: "Time (s)").
|
||||
ylabel: Label for y-axis (default: "Temperature (C)").
|
||||
"""
|
||||
def __init__(self, x = None, y = None, desc = None, job_id = None, job_date = None):
|
||||
def __init__(self, x = None, y = None, setpoint = None, desc = None, job_id = None, job_date = None):
|
||||
Resultable.__init__(self)
|
||||
Drawable.__init__(self)
|
||||
|
||||
# Set default labels
|
||||
self.xlabel = "Time (s)"
|
||||
self.ylabel = "Temperature (C)"
|
||||
|
||||
# Initialize data lists
|
||||
self.setpoint = [] if setpoint is None else setpoint
|
||||
|
||||
if (x is None) and (y is None) and (desc is None) and (job_id is None) and (job_date is None):
|
||||
pass
|
||||
# Empty initialization - will be filled later
|
||||
self.x = []
|
||||
self.y = []
|
||||
|
||||
elif (x is not None) and (y is not None) and (desc is not None) and (job_id is not None) and (job_date is not None):
|
||||
pass
|
||||
|
||||
# Full initialization
|
||||
self.x = x
|
||||
self.y = y
|
||||
self.description = desc
|
||||
self.job_id = job_id
|
||||
self.job_date = job_date
|
||||
|
||||
else:
|
||||
raise ValueError("Wrong usage of __init__!")
|
||||
|
||||
def get_number_of_channels(self):
|
||||
"""Returns 1 for temperature data."""
|
||||
return 1
|
||||
|
||||
def get_xdata(self):
|
||||
"""Returns the x data (timestamps)."""
|
||||
return self.x
|
||||
|
||||
def get_ydata(self, channel=0):
|
||||
"""Returns the y data (temperature readings)."""
|
||||
if channel == 0:
|
||||
return self.y
|
||||
return []
|
||||
|
||||
def __len__(self):
|
||||
"""Returns the number of data points."""
|
||||
return len(self.y) if self.y else 0
|
||||
|
||||
def __repr__(self):
|
||||
"""String representation of the temperature result."""
|
||||
return f"TemperatureResult(points={len(self)}, job_id={self.job_id})"
|
||||
|
||||
@@ -59,6 +59,7 @@ import matplotlib.figure
|
||||
from damaris.gui import ExperimentWriter, ExperimentHandling
|
||||
from damaris.gui import ResultReader, ResultHandling
|
||||
from damaris.gui import BackendDriver
|
||||
from damaris.gui.TemperatureMonitor import TemperatureMonitor
|
||||
#from damaris.gui.gtkcodebuffer import CodeBuffer, SyntaxLoader
|
||||
#from damaris.data import Drawable # this is a base class, it should be used...
|
||||
from damaris.data import DataPool, Accumulation, ADC_Result, MeasurementResult
|
||||
@@ -180,6 +181,10 @@ class DamarisGUI:
|
||||
|
||||
#to stop queued experiments
|
||||
self.stop_experiment_flag = threading.Event()
|
||||
|
||||
# Temperature monitoring
|
||||
self.temp_monitor = None
|
||||
self.temp_controller = None
|
||||
|
||||
exp_script = ""
|
||||
if exp_script_filename is not None and exp_script_filename != "":
|
||||
@@ -572,6 +577,10 @@ class DamarisGUI:
|
||||
self.dump_filename = ""
|
||||
if actual_config[ "data_pool_name" ] != "":
|
||||
self.dump_states( init=True )
|
||||
|
||||
# Start temperature monitoring if configured
|
||||
self.start_temperature_monitoring(actual_config)
|
||||
|
||||
gobject.timeout_add( 200, self.observe_running_experiment )
|
||||
|
||||
def observe_running_experiment( self ):
|
||||
@@ -974,8 +983,91 @@ class DamarisGUI:
|
||||
still_running = [_f for _f in [ self.si.exp_handling, self.si.res_handling, self.si.back_driver ] if _f]
|
||||
for r in still_running:
|
||||
r.quit_flag.set( )
|
||||
|
||||
# Stop temperature monitoring
|
||||
self.stop_temperature_monitoring()
|
||||
|
||||
self.state = DamarisGUI.Stop_State
|
||||
|
||||
def start_temperature_monitoring(self, config):
|
||||
"""
|
||||
Start temperature monitoring thread if configured.
|
||||
|
||||
Args:
|
||||
config: Configuration dictionary from self.config.get()
|
||||
"""
|
||||
# Check if temperature monitoring is enabled
|
||||
if not config.get("enable_temperature_monitoring", False):
|
||||
return
|
||||
|
||||
controller_type = config.get("temperature_controller_type", "eurotherm")
|
||||
temp_interval = config.get("temperature_interval", 2.0)
|
||||
|
||||
try:
|
||||
# Import the factory
|
||||
from damaris.tools.temperature import create_controller
|
||||
|
||||
# Create controller based on type
|
||||
if controller_type == "eurotherm":
|
||||
temp_device = config.get("temperature_device", "")
|
||||
temp_baudrate = config.get("temperature_baudrate", 19200)
|
||||
|
||||
if not temp_device:
|
||||
print("Temperature monitoring: No device configured for Eurotherm")
|
||||
return
|
||||
|
||||
self.temp_controller = create_controller(
|
||||
"eurotherm", temp_device, baudrate=temp_baudrate
|
||||
)
|
||||
print(f"Temperature monitoring started with Eurotherm on {temp_device} (interval: {temp_interval}s)")
|
||||
|
||||
elif controller_type == "simulated":
|
||||
# Simulated controller doesn't need device configuration
|
||||
# Use default parameters: 290K base, 10K amplitude, 60s period, 5% noise, 10% missing
|
||||
self.temp_controller = create_controller("simulated")
|
||||
print(f"Temperature monitoring started with simulated controller (interval: {temp_interval}s)")
|
||||
|
||||
else:
|
||||
print(f"Temperature monitoring: Unknown controller type: {controller_type}")
|
||||
return
|
||||
|
||||
# Create and start monitor thread
|
||||
self.temp_monitor = TemperatureMonitor(
|
||||
controller=self.temp_controller,
|
||||
interval=temp_interval,
|
||||
data_pool=self.data,
|
||||
data_key="temperature"
|
||||
)
|
||||
self.temp_monitor.start()
|
||||
|
||||
except Exception as e:
|
||||
print(f"Failed to start temperature monitoring: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
self.temp_controller = None
|
||||
self.temp_monitor = None
|
||||
|
||||
def stop_temperature_monitoring(self):
|
||||
"""
|
||||
Stop the temperature monitoring thread if running.
|
||||
"""
|
||||
if self.temp_monitor is not None:
|
||||
try:
|
||||
self.temp_monitor.stop()
|
||||
print("Temperature monitoring stopped")
|
||||
except Exception as e:
|
||||
print(f"Error stopping temperature monitoring: {e}")
|
||||
finally:
|
||||
self.temp_monitor = None
|
||||
|
||||
if self.temp_controller is not None:
|
||||
try:
|
||||
self.temp_controller.close()
|
||||
except Exception as e:
|
||||
print(f"Error closing temperature controller: {e}")
|
||||
finally:
|
||||
self.temp_controller = None
|
||||
|
||||
def documentation_init( self ):
|
||||
|
||||
self.doc_urls = {
|
||||
@@ -2059,6 +2151,73 @@ class ConfigTab:
|
||||
backend_vbox.pack_start(self.config_adc_bit_depth_hbox, False, False, 0)
|
||||
self.config_adc_bit_depth_hbox.show_all()
|
||||
|
||||
# Temperature monitoring settings - single line layout
|
||||
self.config_temp_hbox = gtk.HBox(homogeneous=False, spacing=6)
|
||||
|
||||
# Controller type selector
|
||||
self.config_temp_controller_type_label = gtk.Label(label="Controller:")
|
||||
self.config_temp_controller_type_combobox = gtk.ComboBoxText()
|
||||
self.config_temp_controller_type_combobox.append_text("eurotherm")
|
||||
self.config_temp_controller_type_combobox.append_text("simulated")
|
||||
self.config_temp_controller_type_combobox.set_active(0) # Default to eurotherm
|
||||
self.config_temp_controller_type_combobox.set_tooltip_text("Select temperature controller type")
|
||||
self.config_temp_controller_type_combobox.connect("changed", self.on_temp_controller_type_changed)
|
||||
|
||||
self.config_temp_controller_type_hbox = gtk.HBox(homogeneous=False, spacing=0)
|
||||
self.config_temp_controller_type_hbox.pack_start(self.config_temp_controller_type_label, False, False, 0)
|
||||
self.config_temp_controller_type_hbox.pack_start(self.config_temp_controller_type_combobox, False, False, 0)
|
||||
self.config_temp_hbox.pack_start(self.config_temp_controller_type_hbox, False, False, 0)
|
||||
|
||||
# Enable checkbox
|
||||
self.config_temp_monitoring_checkbutton = gtk.CheckButton(label="Temp:")
|
||||
self.config_temp_monitoring_checkbutton.set_tooltip_text("Enable periodic temperature reading from controller")
|
||||
self.config_temp_hbox.pack_start(self.config_temp_monitoring_checkbutton, False, False, 0)
|
||||
|
||||
# Device file chooser with entry (shown for eurotherm, hidden for simulated)
|
||||
self.config_temp_device_entry = gtk.Entry()
|
||||
self.config_temp_device_entry.set_tooltip_text("Serial port device (e.g., /dev/serial/by-id/...)")
|
||||
self.config_temp_device_entry.set_width_chars(30)
|
||||
self.config_temp_device_button = gtk.Button(label="...")
|
||||
self.config_temp_device_button.set_tooltip_text("Browse for temperature controller device")
|
||||
self.config_temp_device_button.connect("clicked", self.on_temp_device_browse_clicked)
|
||||
|
||||
self.config_temp_device_hbox = gtk.HBox(homogeneous=False, spacing=0)
|
||||
self.config_temp_device_hbox.pack_start(self.config_temp_device_entry, True, True, 0)
|
||||
self.config_temp_device_hbox.pack_start(self.config_temp_device_button, False, False, 0)
|
||||
self.config_temp_hbox.pack_start(self.config_temp_device_hbox, True, True, 0)
|
||||
|
||||
# Interval spinbutton
|
||||
self.config_temp_interval_label = gtk.Label(label="Interval (s):")
|
||||
self.config_temp_interval_spinbutton = gtk.SpinButton.new_with_range(0.1, 60, 0.1)
|
||||
self.config_temp_interval_spinbutton.set_value(2.0) # Default to 2 seconds
|
||||
self.config_temp_interval_spinbutton.set_tooltip_text("Time between temperature readings in seconds")
|
||||
|
||||
self.config_temp_interval_hbox = gtk.HBox(homogeneous=False, spacing=6)
|
||||
self.config_temp_interval_hbox.pack_start(self.config_temp_interval_label, False, False, 0)
|
||||
self.config_temp_interval_hbox.pack_start(self.config_temp_interval_spinbutton, False, False, 0)
|
||||
self.config_temp_hbox.pack_start(self.config_temp_interval_hbox, False, False, 0)
|
||||
|
||||
# Baud rate (for eurotherm)
|
||||
self.config_temp_baudrate_combobox = gtk.ComboBoxText()
|
||||
self.config_temp_baudrate_combobox.append_text("19200")
|
||||
self.config_temp_baudrate_combobox.append_text("9600")
|
||||
self.config_temp_baudrate_combobox.append_text("38400")
|
||||
self.config_temp_baudrate_combobox.append_text("57600")
|
||||
self.config_temp_baudrate_combobox.append_text("115200")
|
||||
self.config_temp_baudrate_combobox.set_active(0)
|
||||
self.config_temp_baudrate_combobox.set_tooltip_text("Serial baud rate for temperature controller")
|
||||
self.config_temp_baudrate_hbox = gtk.HBox(homogeneous=False, spacing=6)
|
||||
self.config_temp_baudrate_hbox.pack_start(self.config_temp_baudrate_combobox, False, False, 0)
|
||||
self.config_temp_hbox.pack_start(self.config_temp_baudrate_hbox, False, False, 0)
|
||||
# Note: baud rate is shown for eurotherm, hidden for simulated
|
||||
|
||||
# Set initial visibility based on controller type
|
||||
self._update_temp_widgets_visibility()
|
||||
|
||||
if backend_vbox:
|
||||
backend_vbox.pack_start(self.config_temp_hbox, False, False, 0)
|
||||
self.config_temp_hbox.show_all()
|
||||
|
||||
# insert version informations
|
||||
components_text = """
|
||||
operating system %(os)s
|
||||
@@ -2165,10 +2324,26 @@ pygobject version %(pygobject)s
|
||||
"data_pool_complib": complib,
|
||||
"data_pool_comprate": self.config_data_pool_comprate.get_value_as_int( ),
|
||||
"script_font": self.config_script_font_button.get_font_name( ),
|
||||
"adc_bit_depth": self.config_adc_bit_depth_spinbutton.get_value_as_int()
|
||||
"adc_bit_depth": self.config_adc_bit_depth_spinbutton.get_value_as_int(),
|
||||
# Temperature monitoring configuration
|
||||
"enable_temperature_monitoring": self.config_temp_monitoring_checkbutton.get_active(),
|
||||
"temperature_controller_type": self.config_temp_controller_type_combobox.get_active_text(),
|
||||
"temperature_device": self.config_temp_device_entry.get_text(),
|
||||
"temperature_baudrate": self._get_temp_baudrate(),
|
||||
"temperature_interval": self.config_temp_interval_spinbutton.get_value()
|
||||
}
|
||||
ADC_Result.default_bit_depth = actual_config["adc_bit_depth"]
|
||||
return actual_config
|
||||
|
||||
def _get_temp_baudrate(self):
|
||||
"""Get selected baud rate from combobox."""
|
||||
active_text = self.config_temp_baudrate_combobox.get_active_text()
|
||||
if active_text:
|
||||
try:
|
||||
return int(active_text)
|
||||
except ValueError:
|
||||
pass
|
||||
return 19200
|
||||
|
||||
def set( self, config ):
|
||||
if "start_backend" in config:
|
||||
@@ -2200,6 +2375,29 @@ pygobject version %(pygobject)s
|
||||
ADC_Result.default_bit_depth = int(config["adc_bit_depth"])
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
# Temperature monitoring configuration
|
||||
if "enable_temperature_monitoring" in config:
|
||||
self.config_temp_monitoring_checkbutton.set_active(config["enable_temperature_monitoring"])
|
||||
if "temperature_controller_type" in config:
|
||||
controller_type = config["temperature_controller_type"]
|
||||
if controller_type in ["eurotherm", "simulated"]:
|
||||
self.config_temp_controller_type_combobox.set_active_text(controller_type)
|
||||
# Update widget visibility based on type
|
||||
self._update_temp_widgets_visibility()
|
||||
if "temperature_device" in config:
|
||||
self.config_temp_device_entry.set_text(config["temperature_device"])
|
||||
if "temperature_baudrate" in config:
|
||||
baudrate = str(config["temperature_baudrate"])
|
||||
if baudrate in ["9600", "19200", "38400", "57600", "115200"]:
|
||||
self.config_temp_baudrate_combobox.set_active_text(baudrate)
|
||||
else:
|
||||
self.config_temp_baudrate_combobox.set_active(0) # Default to 19200
|
||||
if "temperature_interval" in config:
|
||||
try:
|
||||
self.config_temp_interval_spinbutton.set_value(float(config["temperature_interval"]))
|
||||
except (ValueError, TypeError):
|
||||
self.config_temp_interval_spinbutton.set_value(2.0)
|
||||
if "data_pool_complib" in config:
|
||||
# find combo-box entry and make it active...
|
||||
model = self.config_data_pool_complib.get_model( )
|
||||
@@ -2216,6 +2414,60 @@ pygobject version %(pygobject)s
|
||||
def on_adc_bit_depth_changed(self, widget):
|
||||
ADC_Result.default_bit_depth = widget.get_value_as_int()
|
||||
|
||||
def on_temp_device_browse_clicked(self, widget):
|
||||
"""
|
||||
Open file chooser dialog to select temperature controller device.
|
||||
Defaults to /dev/serial/by-id/ which is typically used for USB serial converters.
|
||||
"""
|
||||
dialog = gtk.FileChooserDialog(
|
||||
title="Select Temperature Controller Device",
|
||||
parent=None,
|
||||
action=gtk.FileChooserAction.OPEN,
|
||||
buttons=(
|
||||
gtk.STOCK_CANCEL, gtk.ResponseType.CANCEL,
|
||||
gtk.STOCK_OPEN, gtk.ResponseType.OK
|
||||
)
|
||||
)
|
||||
|
||||
# Set default folder to /dev/serial/by-id/ if it exists
|
||||
default_folder = "/dev/serial/by-id/"
|
||||
if os.path.isdir(default_folder):
|
||||
dialog.set_current_folder(default_folder)
|
||||
else:
|
||||
# Fallback to /dev
|
||||
dialog.set_current_folder("/dev/")
|
||||
|
||||
# Only show regular files (device files)
|
||||
dialog.set_select_multiple(False)
|
||||
|
||||
response = dialog.run()
|
||||
if response == gtk.ResponseType.OK:
|
||||
selected_file = dialog.get_filename()
|
||||
self.config_temp_device_entry.set_text(selected_file)
|
||||
|
||||
dialog.destroy()
|
||||
|
||||
def on_temp_controller_type_changed(self, widget):
|
||||
"""Update widget visibility when controller type changes."""
|
||||
self._update_temp_widgets_visibility()
|
||||
|
||||
def _update_temp_widgets_visibility(self):
|
||||
"""Show/hide widgets based on selected controller type."""
|
||||
active_type = self.config_temp_controller_type_combobox.get_active_text()
|
||||
|
||||
if active_type == "eurotherm":
|
||||
# Show device selection widgets
|
||||
self.config_temp_device_hbox.show()
|
||||
self.config_temp_baudrate_hbox.show()
|
||||
elif active_type == "simulated":
|
||||
# Hide device-specific widgets for simulated controller
|
||||
self.config_temp_device_hbox.hide()
|
||||
self.config_temp_baudrate_hbox.hide()
|
||||
else:
|
||||
# Show everything for unknown types
|
||||
self.config_temp_device_hbox.show()
|
||||
self.config_temp_baudrate_hbox.show()
|
||||
|
||||
def load_config_handler( self, widget ):
|
||||
if self.system_default_filename:
|
||||
self.load_config( self.system_default_filename )
|
||||
|
||||
@@ -0,0 +1,159 @@
|
||||
# -*- coding: iso-8859-1 -*-
|
||||
"""
|
||||
Temperature monitoring thread.
|
||||
|
||||
This module provides a thread that periodically reads temperature from a controller
|
||||
and stores it in a TemperatureResult object.
|
||||
"""
|
||||
|
||||
import threading
|
||||
import time
|
||||
import datetime
|
||||
|
||||
from damaris.data import TemperatureResult
|
||||
from damaris.tools.temperature import TemperatureController
|
||||
|
||||
|
||||
class TemperatureMonitor(threading.Thread):
|
||||
"""
|
||||
Thread that periodically reads temperature from a controller
|
||||
and stores it in a TemperatureResult object.
|
||||
|
||||
This thread runs independently and can be configured with any
|
||||
TemperatureController implementation (Eurotherm, Lakeshore, etc.).
|
||||
|
||||
Attributes:
|
||||
controller: The TemperatureController instance to read from.
|
||||
interval: Polling interval in seconds.
|
||||
result: The TemperatureResult object storing collected data.
|
||||
quit_flag: Event to signal the thread to stop.
|
||||
data_pool: Optional DataPool to store results.
|
||||
data_key: Key to use when storing in data_pool.
|
||||
"""
|
||||
|
||||
def __init__(self, controller: TemperatureController,
|
||||
interval: float = 1.0,
|
||||
data_pool=None,
|
||||
data_key: str = "temperature"):
|
||||
"""
|
||||
Initialize the temperature monitor.
|
||||
|
||||
Args:
|
||||
controller: TemperatureController implementation (Eurotherm, etc.)
|
||||
interval: Polling interval in seconds (default: 1.0)
|
||||
data_pool: Optional DataPool to store results (default: None)
|
||||
data_key: Key to use when storing in data_pool (default: "temperature")
|
||||
"""
|
||||
threading.Thread.__init__(self, name="TemperatureMonitor")
|
||||
self.controller = controller
|
||||
self.interval = float(interval)
|
||||
self.data_pool = data_pool
|
||||
self.data_key = data_key
|
||||
self.quit_flag = threading.Event()
|
||||
self.result = TemperatureResult()
|
||||
self._running = False
|
||||
self._lock = threading.RLock()
|
||||
|
||||
# Initialize the result with metadata
|
||||
self.result.xlabel = "Time (s)"
|
||||
self.result.ylabel = "Temperature (C)"
|
||||
self.result.description = {"type": "temperature_monitoring"}
|
||||
|
||||
def run(self):
|
||||
"""
|
||||
Main thread loop: periodically reads temperature and stores it.
|
||||
"""
|
||||
self._running = True
|
||||
start_time = time.time()
|
||||
|
||||
while not self.quit_flag.is_set() and self._running:
|
||||
try:
|
||||
# Read temperature and setpoint
|
||||
timestamp = time.time() - start_time
|
||||
current_temp = self.controller.get_temperature()
|
||||
setpoint = self.controller.get_setpoint()
|
||||
|
||||
# Store in result
|
||||
with self._lock:
|
||||
if not hasattr(self.result, 'x'):
|
||||
self.result.x = []
|
||||
self.result.y = []
|
||||
self.result.setpoint = []
|
||||
|
||||
self.result.x.append(timestamp)
|
||||
self.result.y.append(current_temp)
|
||||
self.result.setpoint.append(setpoint)
|
||||
|
||||
# Update job metadata on first sample
|
||||
if len(self.result.x) == 1:
|
||||
self.result.job_date = datetime.datetime.now()
|
||||
self.result.job_id = id(self)
|
||||
|
||||
# Optional: store in data_pool
|
||||
if self.data_pool is not None:
|
||||
with self._lock:
|
||||
self.data_pool[self.data_key] = self.result
|
||||
|
||||
except Exception as e:
|
||||
print(f"TemperatureMonitor error: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
# Sleep in small increments for responsive shutdown
|
||||
sleep_remaining = self.interval
|
||||
while sleep_remaining > 0 and not self.quit_flag.is_set() and self._running:
|
||||
sleep_time = min(0.1, sleep_remaining)
|
||||
self.quit_flag.wait(sleep_time)
|
||||
sleep_remaining -= sleep_time
|
||||
|
||||
def stop(self):
|
||||
"""
|
||||
Graceful shutdown: signal the thread to stop and wait for it.
|
||||
"""
|
||||
self._running = False
|
||||
self.quit_flag.set()
|
||||
# Wait for the thread to finish its current iteration
|
||||
self.join(timeout=self.interval + 0.5)
|
||||
|
||||
def reset(self):
|
||||
"""
|
||||
Clear the collected data and start fresh.
|
||||
"""
|
||||
with self._lock:
|
||||
self.result = TemperatureResult()
|
||||
self.result.xlabel = "Time (s)"
|
||||
self.result.ylabel = "Temperature (C)"
|
||||
self.result.description = {"type": "temperature_monitoring"}
|
||||
|
||||
def get_current_temperature(self) -> float:
|
||||
"""
|
||||
Get the latest temperature reading.
|
||||
|
||||
Returns:
|
||||
float: Latest temperature in Celsius, or 0.0 if no data.
|
||||
"""
|
||||
with self._lock:
|
||||
if hasattr(self.result, 'y') and len(self.result.y) > 0:
|
||||
return self.result.y[-1]
|
||||
return 0.0
|
||||
|
||||
def get_current_setpoint(self) -> float:
|
||||
"""
|
||||
Get the latest setpoint reading.
|
||||
|
||||
Returns:
|
||||
float: Latest setpoint in Celsius, or 0.0 if no data.
|
||||
"""
|
||||
with self._lock:
|
||||
if hasattr(self.result, 'setpoint') and len(self.result.setpoint) > 0:
|
||||
return self.result.setpoint[-1]
|
||||
return 0.0
|
||||
|
||||
def is_running(self) -> bool:
|
||||
"""
|
||||
Check if the monitor is currently running.
|
||||
|
||||
Returns:
|
||||
bool: True if the monitor thread is active.
|
||||
"""
|
||||
return self._running and not self.quit_flag.is_set()
|
||||
@@ -3,6 +3,8 @@ import re
|
||||
import operator
|
||||
from functools import reduce
|
||||
|
||||
from .temperature import TemperatureController, register_controller
|
||||
|
||||
|
||||
DEBUG=False
|
||||
|
||||
@@ -40,7 +42,7 @@ def checksum(message):
|
||||
bcc = (reduce(operator.xor, list(map(ord,message))))
|
||||
return chr(bcc)
|
||||
|
||||
class Eurotherm(object):
|
||||
class Eurotherm(TemperatureController):
|
||||
def __init__(self, serial_device, baudrate = 19200):
|
||||
self.device = standard_device
|
||||
# timeout: 110 ms to get all answers.
|
||||
@@ -100,6 +102,39 @@ class Eurotherm(object):
|
||||
def get_setpoint_temperature(self):
|
||||
return self.read_param('SL')
|
||||
|
||||
# TemperatureController interface implementation
|
||||
|
||||
def get_temperature(self) -> float:
|
||||
"""Read current temperature in Celsius."""
|
||||
temp = self.read_param('PV')
|
||||
if temp is None:
|
||||
return 0.0
|
||||
try:
|
||||
return float(temp)
|
||||
except (ValueError, TypeError):
|
||||
return 0.0
|
||||
|
||||
def get_setpoint(self) -> float:
|
||||
"""Read target setpoint temperature in Celsius."""
|
||||
temp = self.read_param('SL')
|
||||
if temp is None:
|
||||
return 0.0
|
||||
try:
|
||||
return float(temp)
|
||||
except (ValueError, TypeError):
|
||||
return 0.0
|
||||
|
||||
def set_setpoint(self, temperature: float) -> bool:
|
||||
"""Set target temperature in Celsius."""
|
||||
return self.write_param('SL', str(temperature))
|
||||
|
||||
def close(self):
|
||||
"""Close the serial connection."""
|
||||
if hasattr(self, 's') and self.s is not None:
|
||||
if self.s.is_open:
|
||||
self.s.close()
|
||||
self.s = None
|
||||
|
||||
if __name__ == '__main__':
|
||||
import time
|
||||
delta=5
|
||||
@@ -118,3 +153,6 @@ if __name__ == '__main__':
|
||||
f.write(l)
|
||||
f.flush()
|
||||
f.write('# MARK -- %s --\n'%(time.asctime()))
|
||||
|
||||
# Register Eurotherm with the factory
|
||||
register_controller('eurotherm', Eurotherm)
|
||||
|
||||
@@ -0,0 +1,143 @@
|
||||
# -*- coding: iso-8859-1 -*-
|
||||
"""
|
||||
Temperature controller interface and base classes.
|
||||
|
||||
This module provides a consistent interface for temperature controllers
|
||||
to enable pluggable temperature monitoring systems.
|
||||
"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
# Plugin registry for temperature controllers
|
||||
_controller_registry = {}
|
||||
|
||||
|
||||
def register_controller(name: str, controller_class):
|
||||
"""
|
||||
Register a temperature controller implementation.
|
||||
|
||||
Args:
|
||||
name: Unique identifier for this controller type
|
||||
controller_class: Class that implements TemperatureController
|
||||
"""
|
||||
_controller_registry[name] = controller_class
|
||||
|
||||
|
||||
def create_controller(name: str, *args, **kwargs):
|
||||
"""
|
||||
Create a temperature controller instance by name.
|
||||
|
||||
Args:
|
||||
name: Registered controller type name
|
||||
*args: Positional arguments to pass to controller constructor
|
||||
**kwargs: Keyword arguments to pass to controller constructor
|
||||
|
||||
Returns:
|
||||
TemperatureController: Instance of the requested controller
|
||||
|
||||
Raises:
|
||||
ValueError: If controller name is not registered and cannot be imported
|
||||
"""
|
||||
# If not found, try to auto-import the module
|
||||
if name not in _controller_registry:
|
||||
# Map known controller names to their modules
|
||||
module_map = {
|
||||
'simulated': 'damaris.tools.temperature_simulator',
|
||||
'eurotherm': 'damaris.tools.eurotherm',
|
||||
}
|
||||
if name in module_map:
|
||||
try:
|
||||
# Import the module which will register the controller
|
||||
import importlib
|
||||
importlib.import_module(module_map[name])
|
||||
except ImportError as e:
|
||||
raise ValueError(f"Failed to import controller '{name}': {e}. Available: {list(_controller_registry.keys())}")
|
||||
|
||||
# Check again after import
|
||||
if name not in _controller_registry:
|
||||
raise ValueError(f"Unknown temperature controller: {name}. Available: {list(_controller_registry.keys())}")
|
||||
|
||||
return _controller_registry[name](*args, **kwargs)
|
||||
|
||||
|
||||
def get_controller_names() -> list:
|
||||
"""
|
||||
Get list of available controller names.
|
||||
|
||||
Returns:
|
||||
list: List of registered controller type names
|
||||
"""
|
||||
return list(_controller_registry.keys())
|
||||
|
||||
|
||||
# Pre-register built-in controllers by importing their modules
|
||||
# This ensures they're available without explicit imports
|
||||
try:
|
||||
from . import eurotherm
|
||||
except ImportError as e:
|
||||
# Eurotherm may require serial module which might not be installed
|
||||
# That's OK, it will be imported on-demand if needed
|
||||
pass
|
||||
|
||||
try:
|
||||
from . import temperature_simulator
|
||||
except ImportError as e:
|
||||
# Simulator should always work (no external dependencies)
|
||||
# But if it fails, it will be imported on-demand
|
||||
pass
|
||||
|
||||
|
||||
class TemperatureController(ABC):
|
||||
"""
|
||||
Abstract base class for temperature controllers.
|
||||
|
||||
All temperature controller implementations (Eurotherm, Lakeshore, etc.)
|
||||
should inherit from this class and implement the abstract methods.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def get_temperature(self) -> float:
|
||||
"""
|
||||
Read current temperature in Celsius.
|
||||
|
||||
Returns:
|
||||
float: Current temperature in degrees Celsius.
|
||||
|
||||
Raises:
|
||||
IOError: If communication with the device fails.
|
||||
RuntimeError: If the device is not ready or returns invalid data.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_setpoint(self) -> float:
|
||||
"""
|
||||
Read the target setpoint temperature in Celsius.
|
||||
|
||||
Returns:
|
||||
float: Target temperature in degrees Celsius.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def set_setpoint(self, temperature: float) -> bool:
|
||||
"""
|
||||
Set the target temperature setpoint.
|
||||
|
||||
Args:
|
||||
temperature: Target temperature in degrees Celsius.
|
||||
|
||||
Returns:
|
||||
bool: True if the setpoint was successfully set, False otherwise.
|
||||
"""
|
||||
pass
|
||||
|
||||
def close(self):
|
||||
"""
|
||||
Clean up resources (e.g., close serial connections).
|
||||
|
||||
This method should be called when the controller is no longer needed.
|
||||
The default implementation does nothing; subclasses should override
|
||||
if they have resources to clean up.
|
||||
"""
|
||||
pass
|
||||
@@ -0,0 +1,126 @@
|
||||
# -*- coding: iso-8859-1 -*-
|
||||
"""
|
||||
Simulated temperature controller for testing.
|
||||
|
||||
Provides a temperature controller that simulates real temperature readings
|
||||
with configurable parameters including sinusoidal variation, noise, and missing values.
|
||||
"""
|
||||
|
||||
import time
|
||||
import math
|
||||
import random
|
||||
import threading
|
||||
|
||||
from .temperature import TemperatureController, register_controller
|
||||
|
||||
|
||||
class SimulatedTemperatureController(TemperatureController):
|
||||
"""
|
||||
A simulated temperature controller that generates realistic temperature readings.
|
||||
|
||||
This controller is useful for testing and development without requiring
|
||||
physical hardware. It simulates temperature readings with:
|
||||
- Base temperature around 290K (17C)
|
||||
- Sinusoidal variation with amplitude of 10K
|
||||
- 5% random noise
|
||||
- 10% probability of missing values (returns previous value)
|
||||
- 60 second period for the sine wave
|
||||
|
||||
Attributes:
|
||||
base_temp: Base temperature in Kelvin (default: 290.0)
|
||||
amplitude: Amplitude of sinusoidal variation in Kelvin (default: 10.0)
|
||||
period: Period of sine wave in seconds (default: 60.0)
|
||||
noise_level: Fraction of random noise (default: 0.05 = 5%)
|
||||
missing_probability: Probability of missing value (default: 0.10 = 10%)
|
||||
"""
|
||||
|
||||
def __init__(self, base_temp: float = 290.0, amplitude: float = 10.0,
|
||||
period: float = 60.0, noise_level: float = 0.05,
|
||||
missing_probability: float = 0.10):
|
||||
"""
|
||||
Initialize the simulated temperature controller.
|
||||
|
||||
Args:
|
||||
base_temp: Base temperature in Kelvin (default: 290.0)
|
||||
amplitude: Amplitude of sinusoidal variation in Kelvin (default: 10.0)
|
||||
period: Period of sine wave in seconds (default: 60.0)
|
||||
noise_level: Fraction of random noise (default: 0.05 = 5%)
|
||||
missing_probability: Probability of missing value (default: 0.10 = 10%)
|
||||
"""
|
||||
self.base_temp = float(base_temp)
|
||||
self.amplitude = float(amplitude)
|
||||
self.period = float(period)
|
||||
self.noise_level = float(noise_level)
|
||||
self.missing_probability = float(missing_probability)
|
||||
|
||||
self._start_time = time.time()
|
||||
self._lock = threading.RLock()
|
||||
self._setpoint = self.base_temp
|
||||
self._last_temp = self.base_temp
|
||||
self._last_reading_time = self._start_time
|
||||
|
||||
def get_temperature(self) -> float:
|
||||
"""
|
||||
Get simulated current temperature in Kelvin.
|
||||
|
||||
Returns:
|
||||
float: Simulated temperature in Kelvin
|
||||
"""
|
||||
with self._lock:
|
||||
elapsed = time.time() - self._start_time
|
||||
current_time = time.time()
|
||||
|
||||
# Calculate ideal sinusoidal temperature
|
||||
temp = self.base_temp + self.amplitude * math.sin(2 * math.pi * elapsed / self.period)
|
||||
|
||||
# Add 5% random noise
|
||||
noise = temp * self.noise_level * (random.random() * 2 - 1) # -5% to +5%
|
||||
temp += noise
|
||||
|
||||
# 10% chance of missing value (return last good value)
|
||||
if random.random() < self.missing_probability:
|
||||
temp = self._last_temp
|
||||
else:
|
||||
self._last_temp = temp
|
||||
self._last_reading_time = current_time
|
||||
|
||||
return temp
|
||||
|
||||
def get_setpoint(self) -> float:
|
||||
"""
|
||||
Get the target setpoint temperature in Kelvin.
|
||||
|
||||
Returns:
|
||||
float: Target temperature in Kelvin
|
||||
"""
|
||||
with self._lock:
|
||||
return self._setpoint
|
||||
|
||||
def set_setpoint(self, temperature: float) -> bool:
|
||||
"""
|
||||
Set the target temperature setpoint.
|
||||
|
||||
Args:
|
||||
temperature: Target temperature in Kelvin
|
||||
|
||||
Returns:
|
||||
bool: True if successful, False otherwise
|
||||
"""
|
||||
with self._lock:
|
||||
self._setpoint = float(temperature)
|
||||
return True
|
||||
|
||||
def close(self):
|
||||
"""Clean up resources."""
|
||||
pass
|
||||
|
||||
def reset(self):
|
||||
"""Reset the simulation to initial state."""
|
||||
with self._lock:
|
||||
self._start_time = time.time()
|
||||
self._last_temp = self.base_temp
|
||||
self._last_reading_time = self._start_time
|
||||
|
||||
|
||||
# Register the simulated controller with the factory
|
||||
register_controller('simulated', SimulatedTemperatureController)
|
||||
Reference in New Issue
Block a user