diff --git a/docs/modules/thermalSensorControllerModules/virtual_commands.md b/docs/modules/thermalSensorControllerModules/virtual_commands.md new file mode 100644 index 0000000..f8e88e7 --- /dev/null +++ b/docs/modules/thermalSensorControllerModules/virtual_commands.md @@ -0,0 +1,55 @@ +# Thermal Sensor YAML Command Definitions + +**Directory:** `framework/core/thermalSensorControllerModules` + +## Overview + +This document consolidates Thermal Sensor YAML command definitions used by the virtual controller (`virtualThermalSensorController`). The payloads are posted to the vcomponent control-plane endpoint (`/api/postKVP`) with top-level node `IThermalSensor`. + +--- + +## 1. temperature_update + +### Purpose + +This command injects a raw temperature reading. The vcomponent evaluates per-sensor thresholds and automatically emits thermal state transitions. This is the supported control-plane command for thermal event simulation. + +### YAML Content + +```yaml +IThermalSensor: + command: temperature_update + sensorName: SoC Die + temperatureCelsius: 100.0 + timestampMonotonicMs: 1710000000000 +``` + +### Parameters + +| Parameter | Type | Description | +|---|---|---| +| `command` | string | Must be `temperature_update`. | +| `sensorName` | string | Sensor name providing the reading. | +| `temperatureCelsius` | float | Temperature value in Celsius. | +| `timestampMonotonicMs` | int | Reading timestamp in milliseconds. | + +### Auto-transition Behavior + +Based on current state and configured thresholds, `temperature_update` drives transitions such as: + +- `NORMAL` → `CRITICAL_TEMPERATURE_EXCEEDED` +- `CRITICAL_TEMPERATURE_EXCEEDED` → `CRITICAL_TEMPERATURE_RECOVERED` +- `CRITICAL_TEMPERATURE_RECOVERED` → `NORMAL` +- `CRITICAL_TEMPERATURE_EXCEEDED` or `CRITICAL_TEMPERATURE_RECOVERED` → `CRITICAL_SHUTDOWN_IMMINENT` + +> **Note:** Once in `CRITICAL_SHUTDOWN_IMMINENT`, recovery is intentionally suppressed by policy. + +--- + +## Summary + +Thermal Sensor controller uses one YAML command payload: + +- **Temperature Update (`temperature_update`)**: Threshold logic computes transitions automatically. + +These definitions are used for thermal event simulation and validation in vDevice test flows. diff --git a/docs/modules/thermal_sensor_controller_modules.md b/docs/modules/thermal_sensor_controller_modules.md new file mode 100644 index 0000000..4957da1 --- /dev/null +++ b/docs/modules/thermal_sensor_controller_modules.md @@ -0,0 +1,69 @@ +# Thermal Sensor Controller Modules + +**Directory:** `framework/core/thermalSensorControllerModules` + +**Purpose:** + +The thermal sensor controller modules provide a pluggable set of implementations for simulating or interacting with thermal sensor hardware during L3 testing. Each implementation conforms to a common interface, allowing the factory (`ThermalSensorController`) to select the appropriate backend based on rack configuration without any changes to test logic. + +--- + +**Key Classes:** + +* **ThermalSensorControllerInterface** (`ThermalSensorControllerInterface.py`): + * Abstract base class that all thermal sensor controller implementations must inherit. + * Defines the two mandatory methods: + * `triggerThermalStateChange(state, sensorName, temperatureCelsius, timestampMonotonicMs)` — directly commands a state transition. + * `injectTemperatureUpdate(sensorName, temperatureCelsius, timestampMonotonicMs)` — injects a raw temperature reading and lets the vcomponent derive the state. + +* **virtualThermalSensorController** (`virtualThermalSensorController.py`): + * Used in vDevice (software-only) test flows. + * Posts YAML command payloads to the vcomponent control-plane endpoint (`/api/postKVP`). + * See [`thermalSensorControllerModules/virtual_commands.md`](thermalSensorControllerModules/virtual_commands.md) for full YAML payload definitions. + +* **actualThermalSensorController** (`actualThermalSensorController.py`): + * Used for real hardware flows where thermal events are raised by platform firmware. + * Prompts the test operator via `utUserResponse` to manually trigger the required thermal condition and confirm the result. + +* **programmableThermalChamber** (`programmableThermalChamber.py`): + * Drives a programmable thermal chamber through configured shell command templates. + * Supports a dedicated SSH control session to the chamber if an `address` is provided. + +* **smartSwitchThermalController** (`smartSwitchThermalController.py`): + * Controls a heater or blower via a smart power switch (Kasa, Tapo, APC, etc.) using the RAFT `powerControlClass` abstraction. + * Timed heating/cooling pulses are derived from per-sensor configuration keys such as `heating_duration_seconds` and `seconds_per_degree_celsius`. + +--- + +**How to Use:** + +1. **Rack Configuration:** Add a `thermalSensorController` block to the device entry in the rack YAML: + + ```yaml + thermalSensorController: + type: virtual # vdevice | actual | programmable-thermal-chamber | smartswitch_thermalcontroller + control_port: 8081 + ``` + +2. **Initialization:** The helper class instantiates `ThermalSensorController` (factory in `thermalSensorController.py`), which resolves the correct implementation automatically. + +3. **Triggering Events in Tests:** + + ```python + # Inject a temperature reading (vDevice flow) + self.thermalSensorDevice.injectTemperatureUpdate( + sensorName="SoC Die", + temperatureCelsius=100.0) + + # Force a direct state transition + self.thermalSensorDevice.triggerThermalStateChange( + state="CRITICAL_TEMPERATURE_EXCEEDED", + sensorName="SoC Die", + temperatureCelsius=97.0) + ``` + +--- + +**Related Documents:** + +* [`thermalSensorControllerModules/virtual_commands.md`](thermalSensorControllerModules/virtual_commands.md) — YAML payload reference for the virtual controller. diff --git a/framework/core/thermalSensorController.py b/framework/core/thermalSensorController.py new file mode 100644 index 0000000..b4e29a4 --- /dev/null +++ b/framework/core/thermalSensorController.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 +#** ***************************************************************************** +# * +# * Copyright 2026 RDK Management +# * +# * Licensed under the Apache License, Version 2.0 (the "License"); +# * you may not use this file except in compliance with the License. +# * You may obtain a copy of the License at +# * +# * http://www.apache.org/licenses/LICENSE-2.0 +# * +#* ****************************************************************************** + +from framework.core.thermalSensorControllerModules.actualThermalSensorController import actualThermalSensorController +from framework.core.thermalSensorControllerModules.programmableThermalChamber import programmableThermalChamber +from framework.core.thermalSensorControllerModules.smartSwitchThermalController import smartSwitchThermalController +from framework.core.thermalSensorControllerModules.virtualThermalSensorController import virtualThermalSensorController + + +class ThermalSensorController: + """ + Factory class that selects the appropriate thermal sensor implementation. + + The first argument may be either the legacy platform string or a controller + configuration dictionary from the rack config. This keeps the existing + vDevice flow intact while allowing explicit controller types such as a + programmable thermal chamber. + """ + + VIRTUAL_CONTROLLER_TYPES = { + "vdevice", + "virtual", + "virtual-thermalsensor-client", + } + + PROGRAMMABLE_CHAMBER_TYPES = { + "programmable-thermal-chamber", + "thermal-chamber", + "chamber", + } + + SMART_SWITCH_CONTROLLER_TYPES = { + "smartswitch_thermalcontroller", + "smartswitch-thermalcontroller", + "smart-switch-thermalcontroller", + } + + ACTUAL_CONTROLLER_TYPES = { + "actual", + "actual-thermalsensor-client", + "hardware", + "manual", + } + + def __new__(cls, controllerConfig, session, prompt: str = "~#", port: int = 8080): + controllerType, controllerSettings, prompt, port = cls._normaliseControllerConfig( + controllerConfig, prompt, port) + if controllerType in cls.VIRTUAL_CONTROLLER_TYPES: + return virtualThermalSensorController(session, prompt, port) + if controllerType in cls.PROGRAMMABLE_CHAMBER_TYPES: + return programmableThermalChamber(session, prompt, port, controllerSettings) + if controllerType in cls.SMART_SWITCH_CONTROLLER_TYPES: + return smartSwitchThermalController(session, prompt, port, controllerSettings) + return actualThermalSensorController(session, prompt, port) + + @classmethod + def _normaliseControllerConfig(cls, controllerConfig, prompt: str, port: int): + if isinstance(controllerConfig, dict): + controllerSettings = dict(controllerConfig) + controllerType = str( + controllerSettings.get("type") or controllerSettings.get("platform") or "" + ).strip().lower() + prompt = controllerSettings.get("prompt", prompt) + port = controllerSettings.get( + "control_port", + controllerSettings.get("port", port), + ) + return controllerType, controllerSettings, prompt, port + + controllerType = str(controllerConfig or "").strip().lower() + return controllerType, {}, prompt, port diff --git a/framework/core/thermalSensorControllerModules/ThermalSensorControllerInterface.py b/framework/core/thermalSensorControllerModules/ThermalSensorControllerInterface.py new file mode 100644 index 0000000..e1975f5 --- /dev/null +++ b/framework/core/thermalSensorControllerModules/ThermalSensorControllerInterface.py @@ -0,0 +1,37 @@ +#!/usr/bin/env python3 +#** ***************************************************************************** +# * +# * Copyright 2026 RDK Management +# * +# * Licensed under the Apache License, Version 2.0 (the "License"); +# * you may not use this file except in compliance with the License. +# * You may obtain a copy of the License at +# * +# * http://www.apache.org/licenses/LICENSE-2.0 +# * +#* ****************************************************************************** + +from abc import ABCMeta, abstractmethod + + +class ThermalSensorControllerInterface(metaclass=ABCMeta): + """ + Abstract base class for ThermalSensor controller implementations. + Both virtual (vDevice) and actual hardware implementations must inherit this. + """ + + def __init__(self, session, prompt: str = "~#", port: int = 8080): + self.session = session + self.prompt = prompt + self.port = port + + @abstractmethod + def triggerThermalStateChange(self, state: str, sensorName: str = "", + temperatureCelsius: float = 0.0, + timestampMonotonicMs: int = 0): + pass + + @abstractmethod + def injectTemperatureUpdate(self, sensorName: str, temperatureCelsius: float, + timestampMonotonicMs: int = 0): + pass diff --git a/framework/core/thermalSensorControllerModules/actualThermalSensorController.py b/framework/core/thermalSensorControllerModules/actualThermalSensorController.py new file mode 100644 index 0000000..4ac952e --- /dev/null +++ b/framework/core/thermalSensorControllerModules/actualThermalSensorController.py @@ -0,0 +1,45 @@ +#!/usr/bin/env python3 +#** ***************************************************************************** +# * +# * Copyright 2026 RDK Management +# * +# * Licensed under the Apache License, Version 2.0 (the "License"); +# * you may not use this file except in compliance with the License. +# * You may obtain a copy of the License at +# * +# * http://www.apache.org/licenses/LICENSE-2.0 +# * +#* ****************************************************************************** + +from raft.framework.plugins.ut_raft.utUserResponse import utUserResponse + +from framework.core.thermalSensorControllerModules.ThermalSensorControllerInterface import ThermalSensorControllerInterface + + +class actualThermalSensorController(ThermalSensorControllerInterface): + """ + Actual (hardware) Thermal Sensor controller. + On real hardware, thermal events are raised by platform firmware so they + cannot be injected programmatically. Each method prompts the test operator + to manually trigger the required thermal condition and confirm the result. + """ + + def __init__(self, session, prompt: str = "~#", port: int = 8080): + super().__init__(session, prompt, port) + self.testUserResponse = utUserResponse(session, prompt) + + def triggerThermalStateChange(self, state: str, sensorName: str = "", + temperatureCelsius: float = 0.0, + timestampMonotonicMs: int = 0): + msg = f"Manually trigger thermal state '{state}'" + if sensorName: + msg += f" on sensor '{sensorName}' at {temperatureCelsius} degC" + msg += ". Has the device reached the expected state? (Y/N):" + return self.testUserResponse.getUserYN(msg) + + def injectTemperatureUpdate(self, sensorName: str, temperatureCelsius: float, + timestampMonotonicMs: int = 0): + return self.testUserResponse.getUserYN( + f"Set sensor '{sensorName}' to {temperatureCelsius} Celsius on the hardware. " + f"Has the temperature been reached? (Y/N):" + ) diff --git a/framework/core/thermalSensorControllerModules/programmableThermalChamber.py b/framework/core/thermalSensorControllerModules/programmableThermalChamber.py new file mode 100644 index 0000000..339f735 --- /dev/null +++ b/framework/core/thermalSensorControllerModules/programmableThermalChamber.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +#** ***************************************************************************** +# * +# * Copyright 2026 RDK Management +# * +# * Licensed under the Apache License, Version 2.0 (the "License"); +# * you may not use this file except in compliance with the License. +# * You may obtain a copy of the License at +# * +# * http://www.apache.org/licenses/LICENSE-2.0 +# * +#* ****************************************************************************** + +import time + +from raft.framework.plugins.ut_raft.utUserResponse import utUserResponse + +from framework.core.commandModules.sshConsole import sshConsole +from framework.core.logModule import logModule +from framework.core.thermalSensorControllerModules.ThermalSensorControllerInterface import ThermalSensorControllerInterface + + +class programmableThermalChamber(ThermalSensorControllerInterface): + """ + Thermal controller that drives a programmable chamber through configured + shell command templates. + + Supported config keys: + address: optional chamber host/IP for a dedicated SSH control session + username: SSH username for the chamber session + password: SSH password for the chamber session + ssh_port: optional SSH port for the chamber session + port: accepted as an alias for ssh_port + prompt: optional shell prompt for the chamber session + set_temperature_command: optional command template for raw temperature injection + trigger_state_command: optional command template for direct state changes + stabilization_delay_seconds: optional wait after sending a command + prompt_for_confirmation: ask the operator to confirm the chamber action + """ + + DEFAULT_SET_TEMPERATURE_COMMAND = "chamberctl set --temperature {temperatureCelsius}" + DEFAULT_TRIGGER_STATE_COMMAND = ( + "chamberctl set-state --state {state} --sensor {sensorName}" + ) + DEFAULT_STABILIZATION_DELAY_SECONDS = 30 + + def __init__(self, session, prompt: str = "~#", port: int = 8080, config: dict = None): + super().__init__(session, prompt, port) + self.config = config or {} + self.testUserResponse = utUserResponse(session, prompt) + self._log = logModule(self.__class__.__name__) + self._log.setLevel(self._log.INFO) + self.commandSession = self._createCommandSession(session, prompt) + self.setTemperatureCommand = self.config.get( + "set_temperature_command", self.DEFAULT_SET_TEMPERATURE_COMMAND) + self.triggerStateCommand = self.config.get( + "trigger_state_command", self.DEFAULT_TRIGGER_STATE_COMMAND) + self.promptForConfirmation = bool(self.config.get("prompt_for_confirmation", False)) + self.stabilizationDelaySeconds = float( + self.config.get( + "stabilization_delay_seconds", + self.DEFAULT_STABILIZATION_DELAY_SECONDS, + ) + ) + + def _createCommandSession(self, defaultSession, defaultPrompt: str): + address = self.config.get("address", "") + if not address: + return defaultSession + + username = self.config.get("username", "") + password = self.config.get("password", "") + sshPort = int(self.config.get("ssh_port", self.config.get("port", 22))) + sessionPrompt = self.config.get("prompt", defaultPrompt) + + return sshConsole( + self._log, + address, + username, + password, + port=sshPort, + prompt=sessionPrompt, + ) + + def _buildCommand(self, commandTemplate: str, state: str = "", sensorName: str = "", + temperatureCelsius: float = 0.0, timestampMonotonicMs: int = 0) -> str: + if not commandTemplate: + return "" + return commandTemplate.format( + state=state, + sensorName=sensorName, + temperatureCelsius=temperatureCelsius, + timestampMonotonicMs=timestampMonotonicMs, + port=self.port, + ).strip() + + def _sendCommand(self, commandTemplate: str, state: str = "", sensorName: str = "", + temperatureCelsius: float = 0.0, timestampMonotonicMs: int = 0) -> bool: + command = self._buildCommand( + commandTemplate, + state=state, + sensorName=sensorName, + temperatureCelsius=temperatureCelsius, + timestampMonotonicMs=timestampMonotonicMs, + ) + if not command: + return False + + self.commandSession.write(command) + if self.stabilizationDelaySeconds > 0: + time.sleep(self.stabilizationDelaySeconds) + return True + + def _confirm(self, message: str) -> bool: + if not self.promptForConfirmation: + return True + return self.testUserResponse.getUserYN(message) + + def triggerThermalStateChange(self, state: str, sensorName: str = "", + temperatureCelsius: float = 0.0, + timestampMonotonicMs: int = 0): + if timestampMonotonicMs == 0: + timestampMonotonicMs = int(time.time() * 1000) + + if self._sendCommand( + self.triggerStateCommand, + state=state, + sensorName=sensorName, + temperatureCelsius=temperatureCelsius, + timestampMonotonicMs=timestampMonotonicMs, + ): + return self._confirm( + f"Verify the chamber drove state '{state}' for sensor '{sensorName}'. Continue? (Y/N):" + ) + + if self._sendCommand( + self.setTemperatureCommand, + state=state, + sensorName=sensorName, + temperatureCelsius=temperatureCelsius, + timestampMonotonicMs=timestampMonotonicMs, + ): + return self._confirm( + f"Verify the chamber reached {temperatureCelsius} Celsius for sensor '{sensorName}'. Continue? (Y/N):" + ) + + return self.testUserResponse.getUserYN( + f"No programmable chamber command is configured for state '{state}'. " + f"Apply it manually for sensor '{sensorName}' and continue? (Y/N):" + ) + + def injectTemperatureUpdate(self, sensorName: str, temperatureCelsius: float, + timestampMonotonicMs: int = 0): + if timestampMonotonicMs == 0: + timestampMonotonicMs = int(time.time() * 1000) + + if self._sendCommand( + self.setTemperatureCommand, + sensorName=sensorName, + temperatureCelsius=temperatureCelsius, + timestampMonotonicMs=timestampMonotonicMs, + ): + return self._confirm( + f"Verify the chamber reached {temperatureCelsius} Celsius for sensor '{sensorName}'. Continue? (Y/N):" + ) + + return self.testUserResponse.getUserYN( + f"No programmable chamber command is configured. Set sensor '{sensorName}' " + f"to {temperatureCelsius} Celsius manually and continue? (Y/N):" + ) \ No newline at end of file diff --git a/framework/core/thermalSensorControllerModules/smartSwitchThermalController.py b/framework/core/thermalSensorControllerModules/smartSwitchThermalController.py new file mode 100644 index 0000000..d17b2f5 --- /dev/null +++ b/framework/core/thermalSensorControllerModules/smartSwitchThermalController.py @@ -0,0 +1,193 @@ +#!/usr/bin/env python3 +#** ***************************************************************************** +# * +# * Copyright 2026 RDK Management +# * +# * Licensed under the Apache License, Version 2.0 (the "License"); +# * you may not use this file except in compliance with the License. +# * You may obtain a copy of the License at +# * +# * http://www.apache.org/licenses/LICENSE-2.0 +# * +#* ****************************************************************************** + +import time + +from framework.core.logModule import logModule +from framework.core.powerControl import powerControlClass +from framework.core.thermalSensorControllerModules.ThermalSensorControllerInterface import ThermalSensorControllerInterface + + +class smartSwitchThermalController(ThermalSensorControllerInterface): + """ + Thermal controller that drives a heater or blower through the existing RAFT + power-switch abstractions. + + Required config keys: + type: smartswitch_thermalcontroller + power_switch_type: kasa | tapo | hs100 | apc | apcAos | olimex | SLP + ip: IP address of the smart switch + + Optional config keys are passed through to powerControlClass, for example: + port, username, password, outlet, outlet_id, args, options, retryCount, + retryDelay + + Thermal-specific optional keys: + heating_duration_seconds: wait time after turning the blower on + cooling_duration_seconds: wait time after turning the blower off + heat_on_temperature_celsius: threshold used by injectTemperatureUpdate + minimum_heating_duration_seconds: default timed heating pulse for a + temperature update that should raise the temperature + seconds_per_degree_celsius: extra heat-on time added for each degree + above heat_on_temperature_celsius + maximum_heating_duration_seconds: cap for timed heat pulses + temperature_duration_map_seconds: optional mapping of target + temperature to exact heat-on duration, for example + {90: 300, 100: 600, 110: 900} + """ + + HEATING_STATES = { + "CRITICAL_TEMPERATURE_EXCEEDED", + "CRITICAL_SHUTDOWN_IMMINENT", + } + + NORMAL_STATES = { + "NORMAL", + "CRITICAL_TEMPERATURE_RECOVERED", + } + + DEFAULT_HEAT_ON_TEMPERATURE_CELSIUS = 90.0 + DEFAULT_MINIMUM_HEATING_DURATION_SECONDS = 300 + DEFAULT_SECONDS_PER_DEGREE_CELSIUS = 30 + DEFAULT_MAXIMUM_HEATING_DURATION_SECONDS = 1800 + DEFAULT_COOLING_DURATION_SECONDS = 15 + DEFAULT_TEMPERATURE_DURATION_MAP_SECONDS = { + 90.0: 300.0, + 100.0: 600.0, + 110.0: 900.0, + } + + def __init__(self, session, prompt: str = "~#", port: int = 8080, config: dict = None): + super().__init__(session, prompt, port) + self.config = dict(config or {}) + self._log = logModule(self.__class__.__name__) + self._log.setLevel(self._log.INFO) + self.heatingDurationSeconds = float(self.config.get("heating_duration_seconds", 0)) + self.coolingDurationSeconds = float( + self.config.get("cooling_duration_seconds", self.DEFAULT_COOLING_DURATION_SECONDS) + ) + self.heatOnTemperatureCelsius = float( + self.config.get("heat_on_temperature_celsius", self.DEFAULT_HEAT_ON_TEMPERATURE_CELSIUS) + ) + self.minimumHeatingDurationSeconds = float( + self.config.get( + "minimum_heating_duration_seconds", + self.DEFAULT_MINIMUM_HEATING_DURATION_SECONDS, + ) + ) + self.secondsPerDegreeCelsius = float( + self.config.get( + "seconds_per_degree_celsius", + self.DEFAULT_SECONDS_PER_DEGREE_CELSIUS, + ) + ) + self.maximumHeatingDurationSeconds = float( + self.config.get( + "maximum_heating_duration_seconds", + self.DEFAULT_MAXIMUM_HEATING_DURATION_SECONDS, + ) + ) + self.temperatureDurationMapSeconds = self._parseTemperatureDurationMap( + self.config.get( + "temperature_duration_map_seconds", + dict(self.DEFAULT_TEMPERATURE_DURATION_MAP_SECONDS), + ) + ) + self.powerController = powerControlClass(self._log, self._buildPowerSwitchConfig()) + + def _parseTemperatureDurationMap(self, durationMapConfig) -> list: + if not isinstance(durationMapConfig, dict): + return [] + + parsedDurationMap = [] + for temperatureKey, durationValue in durationMapConfig.items(): + parsedDurationMap.append((float(temperatureKey), float(durationValue))) + parsedDurationMap.sort(key=lambda item: item[0]) + return parsedDurationMap + + def _buildPowerSwitchConfig(self) -> dict: + powerConfig = dict(self.config) + powerSwitchType = ( + powerConfig.get("power_switch_type") + or powerConfig.get("switch_type") + or powerConfig.get("vendor") + ) + if powerSwitchType is None: + raise ValueError("smartswitch_thermalcontroller requires power_switch_type in rack config") + + powerConfig["type"] = powerSwitchType + powerConfig.setdefault("name", "thermal-heater") + return powerConfig + + def _waitAfterToggle(self, enabled: bool) -> None: + delay = self.heatingDurationSeconds if enabled else self.coolingDurationSeconds + if delay > 0: + time.sleep(delay) + + def _setHeaterState(self, enabled: bool, reason: str) -> bool: + self._log.info("[%s] turning heater %s", reason, "ON" if enabled else "OFF") + result = self.powerController.powerOn() if enabled else self.powerController.powerOff() + self._waitAfterToggle(enabled) + return result + + def _getHeatingDurationSeconds(self, temperatureCelsius: float) -> float: + requestedTemperature = float(temperatureCelsius) + if self.temperatureDurationMapSeconds: + for mappedTemperature, mappedDuration in self.temperatureDurationMapSeconds: + if requestedTemperature <= mappedTemperature: + return mappedDuration + return self.temperatureDurationMapSeconds[-1][1] + + deltaCelsius = max(0.0, float(temperatureCelsius) - self.heatOnTemperatureCelsius) + durationSeconds = ( + self.minimumHeatingDurationSeconds + + (deltaCelsius * self.secondsPerDegreeCelsius) + ) + return min(durationSeconds, self.maximumHeatingDurationSeconds) + + def _runHeatingPulse(self, durationSeconds: float, reason: str) -> bool: + if durationSeconds <= 0: + return self._setHeaterState(False, reason) + + if not self.powerController.powerOn(): + return False + + self._log.info("[%s] keeping heater ON for %.1f seconds", reason, durationSeconds) + time.sleep(durationSeconds) + return self._setHeaterState(False, f"{reason}:complete") + + def triggerThermalStateChange(self, state: str, sensorName: str = "", + temperatureCelsius: float = 0.0, + timestampMonotonicMs: int = 0): + del sensorName, temperatureCelsius, timestampMonotonicMs + normalizedState = str(state or "").strip().upper() + if normalizedState in self.HEATING_STATES: + return self._setHeaterState(True, normalizedState) + if normalizedState in self.NORMAL_STATES: + return self._setHeaterState(False, normalizedState) + + self._log.warning("Unknown thermal state '%s'. Leaving heater state unchanged.", state) + return True + + def injectTemperatureUpdate(self, sensorName: str, temperatureCelsius: float, + timestampMonotonicMs: int = 0): + del sensorName, timestampMonotonicMs + requestedTemperature = float(temperatureCelsius) + if requestedTemperature < self.heatOnTemperatureCelsius: + return self._setHeaterState(False, f"temperature_update:{requestedTemperature}") + + heatingDurationSeconds = self._getHeatingDurationSeconds(requestedTemperature) + return self._runHeatingPulse( + heatingDurationSeconds, + f"temperature_update:{requestedTemperature}", + ) \ No newline at end of file diff --git a/framework/core/thermalSensorControllerModules/virtualThermalSensorController.py b/framework/core/thermalSensorControllerModules/virtualThermalSensorController.py new file mode 100644 index 0000000..c13bce4 --- /dev/null +++ b/framework/core/thermalSensorControllerModules/virtualThermalSensorController.py @@ -0,0 +1,71 @@ +#!/usr/bin/env python3 +#** ***************************************************************************** +# * +# * Copyright 2026 RDK Management +# * +# * Licensed under the Apache License, Version 2.0 (the "License"); +# * you may not use this file except in compliance with the License. +# * You may obtain a copy of the License at +# * +# * http://www.apache.org/licenses/LICENSE-2.0 +# * +#* ****************************************************************************** + +import time + +import yaml + +from framework.core.thermalSensorControllerModules.ThermalSensorControllerInterface import ThermalSensorControllerInterface + + +class virtualThermalSensorController(ThermalSensorControllerInterface): + """ + Virtual ThermalSensor controller that sends YAML commands to the vcomponent + control-plane endpoint to simulate thermal events. + """ + + def _sendCommand(self, payload: dict): + yaml_str = yaml.dump(payload) + yaml_str = yaml_str.replace('"', '\\"') + cmd = ( + f'curl -X POST -H "Content-Type: application/x-yaml" ' + f'--data-binary "{yaml_str}" ' + f'"http://localhost:{self.port}/api/postKVP"' + ) + self.session.write(cmd) + + def triggerThermalStateChange(self, state: str, sensorName: str = "", + temperatureCelsius: float = 0.0, + timestampMonotonicMs: int = 0): + if timestampMonotonicMs == 0: + timestampMonotonicMs = int(time.time() * 1000) + + payload = { + "IThermalSensor": { + "command": "thermal_state_change", + "state": state, + "timestampMonotonicMs": timestampMonotonicMs, + } + } + + if sensorName: + payload["IThermalSensor"]["sensorName"] = sensorName + payload["IThermalSensor"]["temperatureCelsius"] = temperatureCelsius + + self._sendCommand(payload) + + def injectTemperatureUpdate(self, sensorName: str, temperatureCelsius: float, + timestampMonotonicMs: int = 0): + if timestampMonotonicMs == 0: + timestampMonotonicMs = int(time.time() * 1000) + + payload = { + "IThermalSensor": { + "command": "temperature_update", + "sensorName": sensorName, + "temperatureCelsius": temperatureCelsius, + "timestampMonotonicMs": timestampMonotonicMs, + } + } + + self._sendCommand(payload)