1"""
  2This module provides simple simulations for the hardware found on the Raspberry Pi.
  3This allows dry running test cases without owning or remoting into a HITL.
  4
  5# (c) 2020-2024 TurnAround Factor, Inc.
  6#
  7# CUI DISTRIBUTION CONTROL
  8# Controlled by: DLA J68 R&D SBIP
  9# CUI Category: Small Business Research and Technology
 10# Distribution/Dissemination Controls: PROTECTED BY SBIR DATA RIGHTS
 11# POC: GOV SBIP Program Manager Denise Price, 571-767-0111
 12# Distribution authorized to U.S. Government Agencies only, to protect information not owned by the
 13# U.S. Government and protected by a contractor’s ‘limited rights’ statement, or received with the understanding that
 14# it is not routinely transmitted outside the U.S. Government (determination made September 14, 2024). Other requests
 15# for this document shall be referred to ACOR DLA Logistics Operations (J-68), 8725 John J. Kingman Rd., Suite 4317,
 16# Fort Belvoir, VA 22060-6221
 17#
 18# SBIR DATA RIGHTS
 19# Contract No.:SP4701-23-C-0083
 20# Contractor Name: TurnAround Factor, Inc.
 21# Contractor Address: 10365 Wood Park Ct. Suite 313 / Ashland, VA 23005
 22# Expiration of SBIR Data Rights Period: September 24, 2029
 23# The Government's rights to use, modify, reproduce, release, perform, display, or disclose technical data or computer
 24# software marked with this legend are restricted during the period shown as provided in paragraph (b)(4) of the Rights
 25# in Noncommercial Technical Data and Computer Software--Small Business Innovative Research (SBIR) Program clause
 26# contained in the above identified contract. No restrictions apply after the expiration date shown above. Any
 27# reproduction of technical data, computer software, or portions thereof marked with this legend must also reproduce
 28# the markings.
 29"""
 30
 31from __future__ import annotations
 32
 33import atexit
 34import random
 35import time
 36from datetime import datetime
 37
 38from hitl_tester.modules.bms_types import TimeoutExceededError
 39from hitl_tester.modules.bms.korad import Status
 40from hitl_tester.modules.logger import logger
 41
 42
 43# TODO(JA): would it be easier to simulate pyvisa for cell/load/dmm/charger/chamber.
 44# We would pass in a fake resource
 45
 46
 47class M300Dmm:
 48    """M300 Dmm."""
 49
 50    def __init__(self, scan_list: list[dict[str, int]]):
 51        self._volts = 1.4
 52        self.scan_list = scan_list
 53
 54    def configure_voltage_trigger(self):
 55        """Set up trigger parameters. Up to 10,000 readings."""
 56
 57    def send_trigger(self):
 58        """Send a software trigger."""
 59
 60    def read_internal_memory(self) -> list[float]:
 61        """Return measured data from internal memory."""
 62        return []
 63
 64    def configure_voltage_normal(self):
 65        """The voltage settings for normal measurements."""
 66
 67    @property
 68    def volts(self) -> float:
 69        """Measures DC voltage."""
 70        self._volts -= 0.001
 71        return float(self._volts)
 72
 73    @property
 74    def amps_ac(self) -> float:
 75        """Measures AC current."""
 76        return 1.0
 77
 78    def reset(self):
 79        """Resets the instrument."""
 80
 81
 82class ADC:
 83    """An ADC plate from PiPlates."""
 84
 85    VOLTAGE = 1.1
 86    MEASUREMENTS_TAKEN = 0
 87
 88    @staticmethod
 89    def setMODE(addr: int, mode: str):  # pylint: disable=unused-argument,invalid-name  # third-party library
 90        """Change mode of plate."""
 91
 92    @staticmethod
 93    def configINPUT(
 94        addr: int, channel: str, sample_rate: int, enable: bool
 95    ):  # pylint: disable=unused-argument,invalid-name  # third-party library
 96        """Configure plate input."""
 97
 98    @staticmethod
 99    def readSINGLE(  # pylint: disable=unused-argument,invalid-name  # From a third-party library
100        addr: int, channel: str
101    ) -> float | None:
102        """Read a plate channel."""
103        if ADC.MEASUREMENTS_TAKEN % 16 == 0:
104            ADC.VOLTAGE -= 0.0005
105        ADC.MEASUREMENTS_TAKEN -= 1
106        return ADC.VOLTAGE + (ADC.MEASUREMENTS_TAKEN % 16) * 0.01
107
108
109class RELAY2:
110    """A RELAY2 plate from PiPlates."""
111
112    @staticmethod
113    def relayON(board_id: int, channel_id: int):  # pylint: disable=unused-argument,invalid-name
114        """Activate relay."""
115
116    @staticmethod
117    def relayOFF(board_id: int, channel_id: int):  # pylint: disable=unused-argument,invalid-name
118        """Activate relay."""
119
120
121class Korad:
122    """Korad KA6003P power supply command wrapper."""
123
124    VOLTAGE_MAX = 60
125    VOLTAGE_MIN = 1.2
126    CURRENT_MAX = 3
127
128    def __init__(self, korad_id: int):
129        """
130        Initialize the KA6003P wrapper with a specific PyVISA resource.
131        This class does NOT open the resource, you have to open it for yourself!
132        """
133        self.id: int = korad_id
134        self._overvoltage_protection = False
135        self._overcurrent_protection = False
136        self._beep = False
137        self._volts = 3.0
138        self._amps = 1.0
139
140        @atexit.register
141        def __atexit__():
142            """Configure a safe shut down for when the class instance is destroyed."""
143            self.disable()
144
145    @property
146    def status(self) -> Status:
147        """Get the power supply status."""
148        return Status(Status.OUTPUT_ON | Status.OVERVOLTAGE_OVERCURRENT_PROTECTION)
149
150    def set_profile(self, volts: float, amps: float):
151        """Sets charging profile"""
152        self.volts = volts
153        self.amps = amps
154
155    @property
156    def overcurrent_protection(self) -> float:
157        """Get the last set overcurrent protection state."""
158        return self._overcurrent_protection
159
160    @overcurrent_protection.setter
161    def overcurrent_protection(self, enabled: bool):
162        """Enable or disable overcurrent protection."""
163        self._overcurrent_protection = enabled
164
165    @property
166    def overvoltage_protection(self) -> float:
167        """Get the last set overvoltage protection state."""
168        return self._overvoltage_protection
169
170    @overvoltage_protection.setter
171    def overvoltage_protection(self, enabled: bool):
172        """Enable or disable overvoltage protection."""
173
174    @property
175    def measured_amps(self) -> float:
176        """Measures current."""
177        return self._amps
178
179    @property
180    def amps(self) -> float:
181        """Get the target current."""
182        return self._amps
183
184    @amps.setter
185    def amps(self, new_amps: float):
186        """Set the target current."""
187        if new_amps > Korad.CURRENT_MAX:
188            raise RuntimeError(f"Current of {new_amps}A exceeds maximum of {Korad.CURRENT_MAX}A.")
189        self._amps = new_amps
190
191    @property
192    def measured_volts(self) -> float:
193        """Measures voltage."""
194        return self._volts
195
196    @property
197    def volts(self) -> float:
198        """Get the target voltage."""
199        return self._volts
200
201    @volts.setter
202    def volts(self, new_volts: float):
203        """Set the target voltage"""
204        if not Korad.VOLTAGE_MIN <= new_volts <= Korad.VOLTAGE_MAX:
205            raise RuntimeError(f"{new_volts}V is out of range {Korad.VOLTAGE_MIN}V to {Korad.VOLTAGE_MAX}V.")
206        self._volts = new_volts
207
208    @property
209    def beep(self) -> bool:
210        """Get the beep setting."""
211        return self._beep
212
213    @beep.setter
214    def beep(self, enable_beep: bool):
215        """Turns on or off the beep."""
216
217    def recall(self, memory_number: int):
218        """Recalls a panel setting from memory 1 to 5."""
219
220    def store(self, memory_number: int):
221        """Stores a panel setting to memory 1 to 5."""
222
223    def enable(self):
224        """Enable the output."""
225
226    def disable(self):
227        """Disable the output."""
228
229
230class Cell:
231    """Basic Agilent 66321 Battery Simulator simulator."""
232
233    def __init__(self, cell_id, cell_chemistry):  # pylint: disable=unused-argument
234        self.id = cell_id
235        self.volts = 0
236        self.ohms = 0.030
237        self.amps = 0
238        self.compensation_mode = None
239        self.state_of_charge = 0
240        self.measured_volts = 0
241
242        @atexit.register
243        def __atexit__():
244            """Configure a safe shut down for when the class instance is destroyed."""
245            logger.write_info_to_report(f"Disabling cell {self.id}")
246            self.disable()
247
248    def enable(self):
249        """Enables power supply output"""
250
251    def disable(self):
252        """Enables power supply output"""
253
254    def reset(self):
255        """Resets the instrument"""
256
257    def volts_to_soc(self, voltage):
258        """Convert voltage to soc based on temp."""
259        return (voltage or 5) / 10
260
261
262class Charger:
263    """Basic Rigol DL711 Power Supply simulator."""
264
265    def __init__(self):
266        self.volts = 0
267        self._amps = 2
268        self._amps_counter = 0
269        self.amps_limit = 0
270        self.volts_limit = 0
271        self.target_amps = 0
272        self.target_volts = 0
273
274        @atexit.register
275        def __atexit__():
276            """Configure a safe shut down for when the class instance is destroyed."""
277            self.disable()
278
279    @property
280    def amps(self):
281        """Simulate amps dropping"""
282        # self._amps_counter += 1
283        # self._amps -= 0.1
284        # if self._amps_counter in (3, 8, 9):
285        #     return 0.084
286        return 0 if 0.100 > self._amps > 0.021 else self._amps
287
288    def set_profile(self, volts, amps):
289        """Sets charging profile"""
290        self.target_volts = volts
291        self.target_amps = amps
292        self.volts = volts
293        self._amps = amps
294
295    def enable(self):
296        """Enables power supply output"""
297        logger.write_info_to_report("Enabling charger")
298
299    def disable(self):
300        """Disables power supply output"""
301        logger.write_info_to_report("Disabling charger")
302
303    def reset(self):
304        """Resets the instrument"""
305
306
307class Load:
308    """Basic Rigol DL3000 Electronic Load simulator."""
309
310    def __init__(self):
311        self.volts = 3.0
312        self.amps = 2.0
313        self.ohms = 10
314        self.amps_range = 0
315        self.volts_range = 0
316        self.ohms_r_range = 0
317        self.ohms_v_range = 0
318        self.ohms_i_range = 0
319
320        @atexit.register
321        def __atexit__():
322            """Configure a safe shut down for when the class instance is destroyed."""
323            self.disable()
324
325    def configure_pulse_trigger(self, pulse_current: int = 1, base_current: int = 0, duration: float = 0.25):
326        """Set up pulse trigger values."""
327
328    def wait_for_pulse(self):
329        """Wait for load pulse to complete."""
330
331    def send_trigger(self):
332        """Send a software trigger."""
333
334    def mode_cc(self):
335        """Sets load to constant current mode."""
336
337    def mode_cr(self):
338        """Sets load to constant resistance mode."""
339
340    def mode_cv(self):
341        """Sets load to constant voltage mode"""
342
343    def enable(self):
344        """Enables the load input"""
345        logger.write_info_to_report("Enabling load")
346
347    def disable(self):
348        """Disables the load input"""
349        logger.write_info_to_report("Disabling load")
350
351    def reset(self):
352        """Resets the instrument"""
353
354
355class Dmm:
356    """Basic Rigol DM3068 DMM simulator."""
357
358    def __init__(self):
359        self._helper = 0
360        self._volts = 4
361        self._amps_ac = 2
362        self._sample_count = 2000
363
364    def configure_voltage_trigger(self, sample_count: int = 2000):
365        """Set up trigger parameters."""
366        self._sample_count = min(2000, sample_count)
367
368    def send_trigger(self):
369        """Send a software trigger."""
370
371    def read_internal_memory(self) -> list[float]:
372        """Return measured data from internal memory."""
373        sample_data = [
374            3.37148454,
375            3.3714377,
376            3.36943853,
377            3.34266015,
378            3.33230059,
379            3.32955,
380            3.32771727,
381            3.32534937,
382            3.32193206,
383            3.32062852,
384        ]
385        for i in range(995):  # Add some noise to the first half of the measurements
386            sample_data.insert(0, sample_data[i] + random.random() / 10000 * random.choice((-1, 1)))
387        for _ in range(995):  # Decrease linearly
388            sample_data.append(sample_data[-1] - 0.0000309725902335)  # Magic number comes from real measurements
389        return sample_data[: self._sample_count]
390
391    @property
392    def volts(self) -> float:
393        """The voltage from the DMM"""
394        # self._helper += 1
395        # if False and self._helper >= 70:  # Trigger undervoltage after 50 readings
396        #     self._volts = 2.3
397        # else:
398        # self._volts = max(2.5, self._volts - 0.05)
399        return float(self._volts)
400
401    @property
402    def amps_ac(self) -> float:
403        """Measures AC current."""
404        return float(self._amps_ac)
405
406    def reset(self):
407        """Resets the instrument"""
408
409
410class THERMO:
411    """Basic thermocouple simulator"""
412
413    @staticmethod
414    def getTEMP(board_id, channel):  # pylint: disable=unused-argument,invalid-name  # From a third-party library
415        """Get the temperature as a float"""
416        return 24.0
417
418    @staticmethod
419    def getADDR(board_id):  # pylint: disable=invalid-name  # From a third-party library
420        """Returns board_id if board is present."""
421        return board_id
422
423
424class DAQC2:
425    """Basic Pi-Plates DAQC2 Plate simulator"""
426
427    @staticmethod
428    def setDAC(address, channel, vdac):  # pylint: disable=unused-argument,invalid-name  # From a third-party library
429        """Set Digital-to-Analogue converter."""
430
431    @staticmethod
432    def setDOUTbit(address, channel):  # pylint: disable=unused-argument,invalid-name  # From a third-party library
433        """Set data out bit."""
434
435    @staticmethod
436    def clrDOUTbit(address, channel):  # pylint: disable=unused-argument,invalid-name  # From a third-party library
437        """Clear data out bit."""
438
439    @staticmethod
440    def getADC(address, channel):  # pylint: disable=unused-argument,invalid-name  # From a third-party library
441        """Temperature in celsius / 100."""
442        return 24 / 100
443
444
445class ThermalChamber:
446    """Watlow F4T controller command wrapper."""
447
448    def __init__(self) -> None:  # pylint: disable=unused-argument  # Dummy function
449        """Initialize the Watlow F4T controller."""
450        self._display_units = "F"
451        self._internal_units = "F"
452        self.date = datetime.today()
453        self._air_temperature = 26.0
454        self._set_point_temperature = 24.0
455
456        @atexit.register
457        def __atexit__():
458            """Configure a safe shut down for when the class instance is destroyed."""
459            logger.write_info_to_report("Stopping thermal chamber loop")
460            self.stop_loop()
461
462    @property
463    def air_temperature(self) -> float:
464        """Get process value
465        :returns: the current chamber temperature
466        """
467        try:
468            self._air_temperature += (
469                (self._set_point_temperature - self._air_temperature)
470                / abs(self._set_point_temperature - self._air_temperature)
471                / 2
472            )
473        except ZeroDivisionError:
474            pass
475        return self._air_temperature  # Working Process Value, reg 2820 [CHAMBER TEMPERATURE]
476
477    @property
478    def set_point_temperature(self) -> float:
479        """Get set point
480        :returns: the set point temperature
481        """
482        return float(self._set_point_temperature)  # Set point/Closed-Loop Set Point, reg 2782/2810, default 75.0°F
483
484    @set_point_temperature.setter
485    def set_point_temperature(self, new_temp: float):
486        """Apply new set point
487        :param new_temp: the new set point
488        """
489        self._set_point_temperature = new_temp  # Set point, reg 2782, default 75.0°F
490
491    def set_room_temperature(self):
492        """Set the target temperature to room temperature"""
493        old_units = self.internal_units
494        self.internal_units = "C"
495        self.set_point_temperature = 23.0  # Room temperature
496        self.internal_units = old_units
497
498    def block_until_set_point_reached(self, timeout: float | None = None, period: float = 1, buffer: float = 1):
499        """Wait until the thermal chamber has reached its set point
500        :param timeout: raise TimeoutError after a certain amount of seconds has passed
501        :param period: how often the temperature is checked in seconds
502        :param buffer: Since measured temperature may not be precise, we actually wait until set_point +- buffer
503        """
504        logger.write_debug_to_report(f"Blocking until {self.set_point_temperature} {self.internal_units}")
505        start_time = time.perf_counter()
506        while not self.set_point_temperature - buffer <= self.air_temperature <= self.set_point_temperature + buffer:
507            if timeout is not None and time.perf_counter() - start_time >= timeout:
508                raise TimeoutExceededError(f"Temperature was not reached after {timeout} seconds")
509            logger.write_debug_to_report(
510                f"Current temp: {self.air_temperature} - Target temp: {self.set_point_temperature} +- {buffer}"
511            )
512            time.sleep(period)
513
514    @property
515    def status(self) -> str:
516        """Read controller status
517        :returns: the chamber status
518        """
519        return "Constant"
520
521    def start_loop(self):
522        """Start loop 1 on chamber"""
523
524    def stop_loop(self):
525        """Stop loop 1 on chamber"""
526
527    @property
528    def date_and_time(self) -> datetime:
529        """Read current date on controller"""
530        return self.date
531
532    @date_and_time.setter
533    def date_and_time(self, new_date: datetime):
534        """Write new date to controller
535        :param new_date: the date to set the controller to
536        """
537        self.date = new_date
538
539    @property
540    def internal_units(self) -> str:
541        """Read internal temperature units
542        :returns: 'C' for Celsius, 'F' for Fahrenheit
543        """
544        return self._internal_units
545
546    @internal_units.setter
547    def internal_units(self, new_units: str):
548        """Write internal temperature units
549        :param new_units: 'C' for Celsius, 'F' for Fahrenheit
550        """
551        self._internal_units = new_units
552
553    @property
554    def display_units(self) -> str:
555        """Read display temperature units
556        :returns: 'C' for Celsius, 'F' for Fahrenheit
557        """
558        return self._display_units
559
560    @display_units.setter
561    def display_units(self, new_units: str):
562        """Write display temperature units
563        :param new_units: 'C' for Celsius, 'F' for Fahrenheit
564        """
565        self._display_units = new_units