1"""
  2This API communicates with a Watlow F4T controller through TCP, allowing the HITL to control the thermal chamber
  3that's attached to the controller.
  4For register information, see "F4T Modbus Registers (Map 1)" pg 227 in "F4T Setup Operation 16802414 Rev A".
  5
  6# (c) 2020-2024 TurnAround Factor, Inc.
  7#
  8# CUI DISTRIBUTION CONTROL
  9# Controlled by: DLA J68 R&D SBIP
 10# CUI Category: Small Business Research and Technology
 11# Distribution/Dissemination Controls: PROTECTED BY SBIR DATA RIGHTS
 12# POC: GOV SBIP Program Manager Denise Price, 571-767-0111
 13# Distribution authorized to U.S. Government Agencies only, to protect information not owned by the
 14# U.S. Government and protected by a contractor’s ‘limited rights’ statement, or received with the understanding that
 15# it is not routinely transmitted outside the U.S. Government (determination made September 14, 2024). Other requests
 16# for this document shall be referred to ACOR DLA Logistics Operations (J-68), 8725 John J. Kingman Rd., Suite 4317,
 17# Fort Belvoir, VA 22060-6221
 18#
 19# SBIR DATA RIGHTS
 20# Contract No.:SP4701-23-C-0083
 21# Contractor Name: TurnAround Factor, Inc.
 22# Contractor Address: 10365 Wood Park Ct. Suite 313 / Ashland, VA 23005
 23# Expiration of SBIR Data Rights Period: September 24, 2029
 24# The Government's rights to use, modify, reproduce, release, perform, display, or disclose technical data or computer
 25# software marked with this legend are restricted during the period shown as provided in paragraph (b)(4) of the Rights
 26# in Noncommercial Technical Data and Computer Software--Small Business Innovative Research (SBIR) Program clause
 27# contained in the above identified contract. No restrictions apply after the expiration date shown above. Any
 28# reproduction of technical data, computer software, or portions thereof marked with this legend must also reproduce
 29# the markings.
 30"""
 31
 32from __future__ import annotations
 33
 34import atexit
 35import time
 36from datetime import datetime
 37from typing import cast
 38
 39import pytest
 40from chamberconnectlibrary.watlowf4t import WatlowF4T
 41
 42from hitl_tester.modules.bms_types import TimeoutExceededError, BMSFlags
 43from hitl_tester.modules.logger import logger
 44
 45
 46class ThermalChamber:
 47    """Watlow F4T controller command wrapper."""
 48
 49    def __init__(self):
 50        """Initialize the Watlow F4T controller."""
 51        self._controller = None
 52        if (
 53            hasattr(pytest, "flags")
 54            and isinstance(pytest.flags, BMSFlags)
 55            and "thermal_chamber_address" in pytest.flags.config
 56        ):
 57            logger.write_info_to_report("Connecting to thermal chamber...")
 58            self._controller = WatlowF4T(
 59                alarms=8,  # the number of available alarms
 60                profiles=True,  # the controller has programming
 61                loops=1,  # the number of control loops (ie temperature)
 62                cond_event=9,  # the event that enables/disables conditioning
 63                cond_event_toggle=False,  # is the condition momentary(False), or maintained(True)
 64                run_module=None,  # The io module that has the chamber run output
 65                run_io=None,  # The run output on the module that has the chamber run out put
 66                limits=[],  # A list of modules that contain limit type cards.
 67                interface="TCP",
 68                host=pytest.flags.config["thermal_chamber_address"],
 69            )
 70            self._controller.connect()
 71            self.internal_units = "C"
 72            self.display_units = "C"
 73
 74        @atexit.register
 75        def __atexit__():
 76            """Configure a safe shut down for when the class instance is destroyed."""
 77            logger.write_info_to_report("Stopping thermal chamber loop")
 78            # self.stop_loop()  # FIXME(JA): test shutting down the chamber
 79
 80    @property
 81    def controller(self) -> WatlowF4T:
 82        """Check if the controller was set up."""
 83        if self._controller:
 84            return self._controller
 85        raise RuntimeError(
 86            "Attempted to access thermal chamber, but ip was never provided. "
 87            "Did you set the property in the HITL hardware config?"
 88        )
 89
 90    @property
 91    def air_temperature(self) -> float:
 92        """Get process value
 93        :returns: the current chamber temperature
 94        """
 95        return float(self.controller.get_loop_pv(1)["air"])  # Working Process Value, reg 2820 [CHAMBER TEMPERATURE]
 96
 97    @property
 98    def set_point_temperature(self) -> float:
 99        """Get set point
100        :returns: the set point temperature
101        """
102        return float(
103            self.controller.get_loop_sp(1)["constant"]
104        )  # Set point/Closed-Loop Set Point, reg 2782/2810, default 75.0°F
105
106    @set_point_temperature.setter
107    def set_point_temperature(self, new_temp: float):
108        """Apply new set point
109        :param new_temp: the new set point
110        """
111        self.controller.set_loop_sp(1, new_temp)  # Set point, reg 2782, default 75.0°F
112
113    def set_room_temperature(self):
114        """Set the target temperature to room temperature"""
115        old_units = self.internal_units
116        self.internal_units = "C"
117        self.set_point_temperature = 23.0  # Room temperature
118        self.internal_units = old_units
119
120    def block_until_set_point_reached(self, timeout: float | None = None, period: float = 1, buffer: float = 1):
121        """Wait until the thermal chamber has reached its set point
122        :param timeout: raise TimeoutError after a certain amount of seconds has passed
123        :param period: how often the temperature is checked in seconds
124        :param buffer: Since measured temperature may not be precise, we actually wait until set_point +- buffer
125        """
126        logger.write_debug_to_report(f"Blocking until {self.set_point_temperature} {self.internal_units}")
127        start_time = time.perf_counter()
128        while not self.set_point_temperature - buffer <= self.air_temperature <= self.set_point_temperature + buffer:
129            if timeout is not None and time.perf_counter() - start_time >= timeout:
130                raise TimeoutExceededError(f"Temperature was not reached after {timeout} seconds")
131            logger.write_debug_to_report(
132                f"Current temp: {self.air_temperature} - Target temp: {self.set_point_temperature} +- {buffer}"
133            )
134            time.sleep(period)
135
136    @property
137    def status(self) -> str:
138        """Read controller status
139        :returns: the chamber status
140        """
141        return str(self.controller.get_status())
142
143    def start_loop(self):
144        """Start loop 1 on chamber"""
145        current_status = self.controller.get_status()
146        if not any(status_type in current_status for status_type in ("Program Running", "Program Paused", "Constant")):
147            self.controller.const_start()
148
149    def stop_loop(self):
150        """Stop loop 1 on chamber"""
151        self.controller.stop()
152
153    @property
154    def date_and_time(self) -> datetime:
155        """Read current date on controller"""
156        return cast(datetime, self.controller.get_datetime())
157
158    @date_and_time.setter
159    def date_and_time(self, new_date: datetime):
160        """Write new date to controller
161        :param new_date: the date to set the controller to
162        """
163        self.controller.set_datetime(new_date)
164
165    @property
166    def internal_units(self) -> str:
167        """Read internal temperature units
168        :returns: 'C' for Celsius, 'F' for Fahrenheit
169        """
170        eth_rtu_temp_val = self.controller.client.read_holding(
171            14080 if self.controller.interface == "RTU" else 6730, 1
172        )[0]
173        return str(self.controller.watlow_val_dict[eth_rtu_temp_val])
174
175    @internal_units.setter
176    def internal_units(self, new_units: str):
177        """Write internal temperature units
178        :param new_units: 'C' for Celsius, 'F' for Fahrenheit
179        """
180        unit_value = {"C": 15, "F": 30}.get(new_units.upper())
181        self.controller.client.write_holding(14080 if self.controller.interface == "RTU" else 6730, unit_value)
182
183    @property
184    def display_units(self) -> str:
185        """Read display temperature units
186        :returns: 'C' for Celsius, 'F' for Fahrenheit
187        """
188        display_temp_val = self.controller.client.read_holding(1328, 1)[0]
189        return str(self.controller.watlow_val_dict[display_temp_val])
190
191    @display_units.setter
192    def display_units(self, new_units: str):
193        """Write display temperature units
194        :param new_units: 'C' for Celsius, 'F' for Fahrenheit
195        """
196        unit_value = {"C": 15, "F": 30}.get(new_units.upper())
197        self.controller.client.write_holding(1328, unit_value)