diff --git a/src/damaris/data/Temperature.py b/src/damaris/data/Temperature.py index 400d018..8ae5f05 100644 --- a/src/damaris/data/Temperature.py +++ b/src/damaris/data/Temperature.py @@ -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})" diff --git a/src/damaris/gui/DamarisGUI.py b/src/damaris/gui/DamarisGUI.py index d97e610..9b41c95 100644 --- a/src/damaris/gui/DamarisGUI.py +++ b/src/damaris/gui/DamarisGUI.py @@ -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 ) diff --git a/src/damaris/gui/TemperatureMonitor.py b/src/damaris/gui/TemperatureMonitor.py new file mode 100644 index 0000000..dfbcf10 --- /dev/null +++ b/src/damaris/gui/TemperatureMonitor.py @@ -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() diff --git a/src/damaris/tools/eurotherm.py b/src/damaris/tools/eurotherm.py index fe34b98..d231779 100644 --- a/src/damaris/tools/eurotherm.py +++ b/src/damaris/tools/eurotherm.py @@ -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) diff --git a/src/damaris/tools/temperature.py b/src/damaris/tools/temperature.py new file mode 100644 index 0000000..5ab04b8 --- /dev/null +++ b/src/damaris/tools/temperature.py @@ -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 diff --git a/src/damaris/tools/temperature_simulator.py b/src/damaris/tools/temperature_simulator.py new file mode 100644 index 0000000..0a57b67 --- /dev/null +++ b/src/damaris/tools/temperature_simulator.py @@ -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)