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, timedelta
 37from typing import cast
 38
 39import pytest
 40
 41from hitl_tester.modules.bms_types import TimeoutExceededError, BMSFlags, SafeResource
 42from hitl_tester.modules.logger import logger
 43
 44
 45class ThermalChamber:
 46    """Watlow F4T controller command wrapper."""
 47
 48    def __init__(self):
 49        """Initialize the Watlow F4T controller."""
 50        if (
 51            hasattr(pytest, "flags")
 52            and isinstance(pytest.flags, BMSFlags)
 53            and "thermal_chamber_address" in pytest.flags.config
 54        ):
 55            logger.write_info_to_report("Connecting to thermal chamber...")
 56            ip = pytest.flags.config["thermal_chamber_address"]
 57            self.resource = SafeResource((f"TCPIP::{ip}::5025::SOCKET", "\n", "\n"))
 58            self.internal_units = "C"
 59            self.display_units = "C"
 60            self._timedelta = timedelta()
 61
 62        @atexit.register
 63        def __atexit__():
 64            """Configure a safe shut down for when the class instance is destroyed."""
 65            logger.write_info_to_report("Stopping thermal chamber loop")
 66
 67    @property
 68    def air_temperature(self) -> float:
 69        """Get the current chamber temperature."""
 70        return float(self.resource.query("SOUR:CLO1:PVAL?"))
 71
 72    @property
 73    def set_point_temperature(self) -> float:
 74        """Get the set point temperature."""
 75        return float(self.resource.query("SOUR:CLO1:SPO?"))  # Set point/Closed-Loop Set Point, default 75.0°F
 76
 77    @set_point_temperature.setter
 78    def set_point_temperature(self, new_temp: float):
 79        """Apply new set point."""
 80        self.resource.write(f"SOUR:CLO1:SPO {new_temp}")  # Set point, default 75.0°F
 81
 82    def set_room_temperature(self):
 83        """Set the target temperature to room temperature."""
 84        old_units = self.internal_units
 85        self.internal_units = "C"
 86        self.set_point_temperature = 23.0  # Room temperature
 87        self.internal_units = old_units
 88
 89    def block_until_set_point_reached(self, timeout: float | None = None, period: float = 1, buffer: float = 1):
 90        """Wait until the thermal chamber has reached its set point
 91        :param timeout: raise TimeoutError after a certain amount of seconds has passed
 92        :param period: how often the temperature is checked in seconds
 93        :param buffer: Since measured temperature may not be precise, we actually wait until set_point +- buffer
 94        """
 95        logger.write_debug_to_report(f"Blocking until {self.set_point_temperature} {self.internal_units}")
 96        start_time = time.perf_counter()
 97        while not self.set_point_temperature - buffer <= self.air_temperature <= self.set_point_temperature + buffer:
 98            if timeout is not None and time.perf_counter() - start_time >= timeout:
 99                raise TimeoutExceededError(f"Temperature was not reached after {timeout} seconds")
100            logger.write_debug_to_report(
101                f"Current temp: {self.air_temperature} - Target temp: {self.set_point_temperature} +- {buffer}"
102            )
103            time.sleep(period)
104
105    @property
106    def status(self) -> str:
107        """Read controller status."""
108        return "Constant"
109
110    def start_loop(self):
111        """Start loop 1 on chamber."""
112
113    def stop_loop(self):
114        """Stop loop 1 on chamber"""
115
116    @property
117    def date_and_time(self) -> datetime:
118        """Read current date on controller."""
119        return cast(datetime, datetime.now() - self._timedelta)
120
121    @date_and_time.setter
122    def date_and_time(self, new_date: datetime):
123        """Write new date to controller."""
124        self._timedelta = datetime.now() - new_date
125
126    @property
127    def internal_units(self) -> str:
128        """Read internal temperature units ('C' for Celsius, 'F' for Fahrenheit)."""
129        return str(self.resource.query(":UNIT:TEMP?"))
130
131    @internal_units.setter
132    def internal_units(self, new_units: str):
133        """Write internal temperature units ('C' for Celsius, 'F' for Fahrenheit)."""
134        self.resource.write(f":UNIT:TEMP {new_units.upper()}")
135
136    @property
137    def display_units(self) -> str:
138        """Read display temperature units ('C' for Celsius, 'F' for Fahrenheit)."""
139        return str(self.resource.query(":UNIT:TEMP:DISP?"))
140
141    @display_units.setter
142    def display_units(self, new_units: str):
143        """Write display temperature units ('C' for Celsius, 'F' for Fahrenheit)."""
144        self.resource.write(f":UNIT:TEMP:DISP {new_units.upper()}")