Adds abstract methods; adds dummy_rx and dummy_tx, make a package for managing imports

config.py

 import logging
-from utils import get_platform
+from OhmPi.utils import get_platform
 
 from paho.mqtt.client import MQTTv31
 
@@ -27,27 +27,23 @@ OHMPI_CONFIG = {
 }  # TODO: add a dictionary with INA models and associated gain values
 
 HARDWARE_CONFIG = {
-    {'controller': {'model' : 'raspberry_pi_3'
-                    }
-     },
-    {'tx' : {'model' : 'mb_2024_rev_0_0',
+    'controller': {'model' : 'raspberry_pi_3'
+                   },
+    'tx' : {'model' : 'mb_2024_rev_0_0',
              'mcp_board_address': 0x20,
              'Imax': 4800 / 50 / 2,  # Maximum current
              'R_shunt': 2  # Shunt resistance in Ohms
-             }
-     },
-    {'rx' : {'model': 'mb_2024_rev_0_0',
+            },
+    'rx' : {'model': 'mb_2024_rev_0_0',
              'coef_p2': 2.50,  # slope for current conversion for ADS.P2, measurement in V/V
              'nb_samples': 20,  # Max value 10 # was named integer before...
-             }
-     },
-    {'mux': {'model' : 'mux_2021',
+            },
+    'mux': {'model' : 'mux_2021',
              'max_elec': 64,
              'board_addresses': {'A': 0x73, 'B': 0x72, 'M': 0x71, 'N': 0x70},  # CHECK IF YOUR BOARDS HAVE THESE ADDRESSES
              'coef_p2': 2.50,  # slope for current conversion for ADS.P2, measurement in V/V
              'nb_samples': 20  # Max value 10 # was named integer before...
-             }
-     }
+            }
 }
 
 # SET THE LOGGING LEVELS, MQTT BROKERS AND MQTT OPTIONS ACCORDING TO YOUR NEEDS

hardware/__init__.py

+from OhmPi.hardware.abstract_hardware import TxAbstract, RxAbstract, MuxAbstract, ControllerAbstract
\ No newline at end of file

hardware/hardware.py → hardware/abstract_hardware.py

-from abc import ABC
-import os
-from ..logging_setup import create_default_logger
+from abc import ABC, abstractmethod
+from OhmPi.logging_setup import create_default_logger
 
 class ControllerAbstract(ABC):
     def __init__(self, **kwargs):
@@ -11,31 +10,49 @@ class MuxAbstract(ABC):
 
 class TxAbstract(ABC):
     def __init__(self, **kwargs):
+        self.board_name = kwargs.pop('board_name', 'unknown TX hardware')
         polarity = kwargs.pop('polarity', 1)
         inj_time = kwargs.pop('inj_time', 1.)
-        self.exec_logger = kwargs.pop('exec_logger', create_default_logger('exec'))
-        self.soh_logger = kwargs.pop('soh_logger', create_default_logger('soh'))
+        self.exec_logger = kwargs.pop('exec_logger', create_default_logger('exec_tx'))
+        self.soh_logger = kwargs.pop('soh_logger', create_default_logger('soh_tx'))
         self._polarity = None
         self._inj_time = None
         self._dps_state = 'off'
+        self._adc_gain = 1.
         self.polarity = polarity
         self.inj_time = inj_time
-        self.board_name = os.path.basename(__file__)
-        self.exec_logger.debug(f'TX {self.board_name} initialization')
+        self.exec_logger.debug(f'{self.board_name} TX initialization')
+
+    @property
+    def adc_gain(self):
+        return self._adc_gain
+
+    @adc_gain.setter
+    def adc_gain(self, value):
+        self._adc_gain = value
+        self.exec_logger.debug(f'Setting TX ADC gain to {value}')
+
+    @abstractmethod
+    def adc_gain_auto(self):
+        pass
 
     @property
+    @abstractmethod
     def current(self):
         # add actions to read the DPS current and return it
         return None
 
     @current.setter
+    @abstractmethod
     def current(self, value, **kwargs):
         # add actions to set the DPS current
         pass
 
+    @abstractmethod
     def current_pulse(self, **kwargs):
         pass
 
+    @abstractmethod
     def inject(self, state='on'):
         assert state in ['on', 'off']
 
@@ -67,15 +84,23 @@ class TxAbstract(ABC):
         self._dps_state = 'on'
 
     @property
+    @abstractmethod
     def voltage(self):
         # add actions to read the DPS voltage and return it
         return None
 
     @voltage.setter
+    @abstractmethod
     def voltage(self, value, **kwargs):
         # add actions to set the DPS voltage
         pass
 
+    @property
+    @abstractmethod
+    def tx_bat(self):
+        pass
+
+    @abstractmethod
     def voltage_pulse(self, voltage, length, polarity):
         """ Generates a square voltage pulse
 
@@ -95,6 +120,27 @@ class RxAbstract(ABC):
     def __init__(self, **kwargs):
         self.exec_logger = kwargs.pop('exec_logger', create_default_logger('exec'))
         self.soh_logger = kwargs.pop('soh_logger', create_default_logger('soh'))
-        self.board_name = os.path.basename(__file__)
-        self.exec_logger.debug(f'RX {self.board_name} initialization')
+        self.board_name = kwargs.pop('board_name', 'unknown RX hardware')
+        self.exec_logger.debug(f'{self.board_name} RX initialization')
+        self._adc_gain = 1.
 
+    @property
+    def adc_gain(self):
+        return self._adc_gain
+
+
+    @adc_gain.setter
+    def adc_gain(self, value):
+        self._adc_gain = value
+        self.exec_logger.debug(f'Setting RX ADC gain to {value}')
+
+    @abstractmethod
+    def adc_gain_auto(self):
+        pass
+
+    @property
+    @abstractmethod
+    def voltage(self):
+        """ Gets the voltage VMN in Volts
+        """
+        pass
\ No newline at end of file

hardware/dummy_rx.py

+from OhmPi.config import HARDWARE_CONFIG
+import numpy as np
+import os
+from OhmPi.hardware import RxAbstract
+RX_CONFIG = HARDWARE_CONFIG['rx']
+
+# hardware limits
+voltage_min = 10.  # mV
+voltage_max = 4500.
+RX_CONFIG['voltage_min'] = voltage_min  # mV
+RX_CONFIG['voltage_max'] = voltage_max
+
+class Rx(RxAbstract):
+    def __init__(self, **kwargs):
+        kwargs.update({'board_name': os.path.basename(__file__).rstrip('.py')})
+        super().__init__(**kwargs)
+        self._adc_gain = 1.
+
+    @property
+    def adc_gain(self):
+        return self._adc_gain
+
+    @adc_gain.setter
+    def adc_gain(self, value):
+        self.exec_logger.debug(f'Setting RX ADC gain to {value}')
+
+    def adc_gain_auto(self):
+        gain = 1
+        self.exec_logger.debug(f'Setting TX ADC gain automatically to {gain}')
+        self.adc_gain = gain
+
+    @property
+    def voltage(self):
+        """ Gets the voltage VMN in Volts
+        """
+        u = np.random.uniform(-.200,.200) # gets the max between u0 & u2 and set the sign
+        self.exec_logger.debug(f'Reading random voltage on RX. Returning {u} V')
+        return u
\ No newline at end of file

hardware/dummy_tx.py

+from OhmPi.config import HARDWARE_CONFIG
+import time
+import os
+import numpy as np
+from OhmPi.hardware import TxAbstract
+
+TX_CONFIG = HARDWARE_CONFIG['tx']
+
+# hardware limits
+voltage_min = 10.  # mV
+voltage_max = 4500.
+
+TX_CONFIG['current_min'] = voltage_min / (TX_CONFIG['R_shunt'] * 50)  # mA
+TX_CONFIG['current_max'] = voltage_max / (TX_CONFIG['R_shunt'] * 50)
+TX_CONFIG['default_voltage'] = 5.  # V
+TX_CONFIG['voltage_max'] = 50.  # V
+TX_CONFIG['dps_switch_on_warm_up'] = 4. # 4 seconds
+
+class Tx(TxAbstract):
+    def inject(self, state='on'):
+        pass
+
+    def __init__(self, **kwargs):
+        kwargs.update({'board_name': os.path.basename(__file__).rstrip('.py')})
+        super().__init__(**kwargs)
+        self._voltage = kwargs.pop('voltage', TX_CONFIG['default_voltage'])
+
+        self._adc_gain = 1
+
+        self.polarity = 0
+        self.turn_on()
+        time.sleep(TX_CONFIG['dps_switch_on_warm_up'])
+        self.exec_logger.info(f'TX battery: {self.tx_bat:.1f} V')
+        self.turn_off()
+
+    @property
+    def adc_gain(self):
+        return self._adc_gain
+
+    @adc_gain.setter
+    def adc_gain(self, value):
+        self._adc_gain = value
+        self.exec_logger.debug(f'Setting TX ADC gain to {value}')
+
+    def adc_gain_auto(self):
+        gain = 1.
+        self.exec_logger.debug(f'Setting TX ADC gain automatically to {gain}')
+        self.adc_gain = gain
+
+    def current_pulse(self, **kwargs):
+        super().current_pulse(**kwargs)
+        self.exec_logger.warning(f'Current pulse is not implemented for the {TX_CONFIG["model"]} board')
+
+    @property
+    def current(self):
+        """ Gets the current IAB in Amps
+        """
+        current = np.abs(np.random.normal(0.7, 0.2))
+        self.exec_logger.debug(f'Reading random current on TX. Returning {current} A')
+        return current
+
+    @ current.setter
+    def current(self, value):
+        assert TX_CONFIG['current_min'] <= value <= TX_CONFIG['current_max']
+        self.exec_logger.warning(f'Current pulse is not implemented for the {TX_CONFIG["model"]} board')
+
+    @property
+    def voltage(self):
+        return self._voltage
+    @voltage.setter
+    def voltage(self, value):
+        if value > TX_CONFIG['voltage_max']:
+            self.exec_logger.warning(f'Sorry, cannot inject more than {TX_CONFIG["voltage_max"]} V, '
+                                     f'set it back to {TX_CONFIG["default_voltage"]} V (default value).')
+            value = TX_CONFIG['default_voltage']
+        if value < 0.:
+            self.exec_logger.warning(f'Voltage should be given as a positive number. '
+                                     f'Set polarity to -1 to reverse voltage...')
+            value = np.abs(value)
+
+
+    @property
+    def tx_bat(self):
+        tx_bat = np.random.uniform(10.9, 13.4)
+        if tx_bat < 12.:
+            self.soh_logger.debug(f'Low TX Battery: {tx_bat:.1f} V')
+        return tx_bat
+
+    def voltage_pulse(self, voltage=TX_CONFIG['default_voltage'], length=None, polarity=None):
+        """ Generates a square voltage pulse
+
+        Parameters
+        ----------
+        voltage: float, optional
+            Voltage to apply in volts, tx_v_def is applied if omitted.
+        length: float, optional
+            Length of the pulse in seconds
+        polarity: 1,0,-1
+            Polarity of the pulse
+        """
+
+        if length is None:
+            length = self.inj_time
+        if polarity is None:
+            polarity = self.polarity
+        self.polarity = polarity
+        self.voltage = voltage
+        self.exec_logger.debug(f'Voltage pulse of {polarity*voltage:.3f} V for {length:.3f} s')
+        self.inject(state='on')
+        time.sleep(length)
+        self.inject(state='off')

hardware/mb_2024_rev_0_0.py

 import importlib
-from ..config import OHMPI_CONFIG
+from OhmPi.config import OHMPI_CONFIG
 import adafruit_ads1x15.ads1115 as ads  # noqa
 from adafruit_ads1x15.analog_in import AnalogIn  # noqa
 from adafruit_mcp230xx.mcp23008 import MCP23008  # noqa
@@ -7,11 +7,12 @@ from digitalio import Direction  # noqa
 import minimalmodbus  # noqa
 import time
 import numpy as np
-from hardware import TxAbstract, RxAbstract
+import os
+from OhmPi.hardware import TxAbstract, RxAbstract
 controller_module = importlib.import_module(f'{OHMPI_CONFIG["hardware"]["controller"]["model"]}')
 
-TX_CONFIG = OHMPI_CONFIG['hardware']['TX']
-RX_CONFIG = OHMPI_CONFIG['hardware']['RX']
+TX_CONFIG = OHMPI_CONFIG['rx']
+RX_CONFIG = OHMPI_CONFIG['tx']
 
 # hardware limits
 voltage_min = 10.  # mV
@@ -51,6 +52,7 @@ def _gain_auto(channel):
 
 class Tx(TxAbstract):
     def __init__(self, **kwargs):
+        kwargs.update({'board_name': os.path.basename(__file__).rstrip('.py')})
         super().__init__(**kwargs)
         self._voltage = kwargs.pop('voltage', TX_CONFIG['default_voltage'])
         self.controller = kwargs.pop('controller', controller_module.Controller())
@@ -60,9 +62,9 @@ class Tx(TxAbstract):
 
         # ADS1115 for current measurement (AB)
         self._adc_gain = 2/3
-        self.ads_current_address = 0x48
-        self.ads_current = ads.ADS1115(self.controller.bus, gain=self.adc_gain, data_rate=860,
-                                       address=self.ads_current_address)
+        self._ads_current_address = 0x48
+        self._ads_current = ads.ADS1115(self.controller.bus, gain=self.adc_gain, data_rate=860,
+                                        address=self._ads_current_address)
 
         # Relays for pulse polarity
         self.pin0 = self.mcp_board.get_pin(0)
@@ -85,7 +87,7 @@ class Tx(TxAbstract):
         self.DPS.debug = False  #
         self.DPS.serial.parity = 'N'  # No parity
         self.DPS.mode = minimalmodbus.MODE_RTU  # RTU mode
-        self.DPS.write_register(0x0001, 1000, 0)  # max current allowed (100 mA for relays)
+        self.DPS.write_register(0x0001, 1000, 0)  # max current allowed (100 mA for relays) :
         # (last number) 0 is for mA, 3 is for A
 
         # I2C connexion to MCP23008, for current injection
@@ -93,12 +95,7 @@ class Tx(TxAbstract):
         self.pin4.direction = Direction.OUTPUT
         self.pin4.value = True
 
-        tx_bat = self.DPS.read_register(0x05, 2)
-        if self.exec_logger is not None:
-            self.exec_logger.info(f'TX battery: {tx_bat:.1f} V')
-        if tx_bat < 12.:
-            if self.soh_logger is not None:
-                self.soh_logger.debug(f'Low TX Battery: {tx_bat:.1f} V')  # TODO: SOH logger
+        self.exec_logger.info(f'TX battery: {self.tx_bat:.1f} V')
         self.turn_off()
 
     @property
@@ -109,12 +106,12 @@ class Tx(TxAbstract):
     def adc_gain(self, value):
         assert value in [2/3, 2, 4, 8, 16]
         self._adc_gain = value
-        self.ads_current = ads.ADS1115(self.controller.bus, gain=self.adc_gain, data_rate=860,
-                                       address=self.ads_current_address)
+        self._ads_current = ads.ADS1115(self.controller.bus, gain=self.adc_gain, data_rate=860,
+                                        address=self._ads_current_address)
         self.exec_logger.debug(f'Setting TX ADC gain to {value}')
 
     def adc_gain_auto(self):
-        gain = _gain_auto(AnalogIn(self.ads_current, ads.P0))
+        gain = _gain_auto(AnalogIn(self._ads_current, ads.P0))
         self.exec_logger.debug(f'Setting TX ADC gain automatically to {gain}')
         self.adc_gain = gain
 
@@ -126,7 +123,7 @@ class Tx(TxAbstract):
     def current(self):
         """ Gets the current IAB in Amps
         """
-        return AnalogIn(self.ads_current, ads.P0).voltage * 1000. / (50 * TX_CONFIG['R_shunt'])  # noqa measure current
+        return AnalogIn(self._ads_current, ads.P0).voltage * 1000. / (50 * TX_CONFIG['R_shunt'])  # noqa measure current
 
     @ current.setter
     def current(self, value):
@@ -184,6 +181,13 @@ class Tx(TxAbstract):
         self.pin2.value = True
         self.pin3.value = True
 
+    @property
+    def tx_bat(self):
+        tx_bat = self.DPS.read_register(0x05, 2)
+        if tx_bat < 12.:
+            self.soh_logger.debug(f'Low TX Battery: {tx_bat:.1f} V')
+        return tx_bat
+
     def voltage_pulse(self, voltage=TX_CONFIG['default_voltage'], length=None, polarity=None):
         """ Generates a square voltage pulse
 
@@ -210,31 +214,30 @@ class Tx(TxAbstract):
 
 class Rx(RxAbstract):
     def __init__(self, **kwargs):
-        self.controller = kwargs.pop('controller', controller_module.Controller())
-        self._adc_gain = [2/3, 2/3]
+        kwargs.update({'board_name': os.path.basename(__file__).rstrip('.py')})
         super().__init__(**kwargs)
-        self.ads_voltage_address = 0x49
-        # ADS1115 for voltage measurement (MN)
-        self.ads_voltage = ads.ADS1115(self.controller.bus, gain=2/3, data_rate=860, address=self.ads_voltage_address)
+        self.controller = kwargs.pop('controller', controller_module.Controller())
 
+        # ADS1115 for voltage measurement (MN)
+        self._ads_voltage_address = 0x49
+        self._adc_gain = 2/3
+        self._ads_voltage = ads.ADS1115(self.controller.bus, gain=self._adc_gain, data_rate=860, address=self._ads_voltage_address)
 
     @property
     def adc_gain(self):
         return self._adc_gain
 
-
     @adc_gain.setter
     def adc_gain(self, value):
-        assert value in [2 / 3, 2, 4, 8, 16]
+        assert value in [2/3, 2, 4, 8, 16]
         self._adc_gain = value
-        self.ads_voltage = ads.ADS1115(self.controller.bus, gain=self.adc_gain, data_rate=860,
-                                       address=self.ads_voltage_address)
+        self._ads_voltage = ads.ADS1115(self.controller.bus, gain=self.adc_gain, data_rate=860,
+                                        address=self._ads_voltage_address)
         self.exec_logger.debug(f'Setting RX ADC gain to {value}')
 
-
     def adc_gain_auto(self):
-        gain_0 = _gain_auto(AnalogIn(self.ads_voltage, ads.P0))
-        gain_2 = _gain_auto(AnalogIn(self.ads_voltage, ads.P2))
+        gain_0 = _gain_auto(AnalogIn(self._ads_voltage, ads.P0))
+        gain_2 = _gain_auto(AnalogIn(self._ads_voltage, ads.P2))
         gain = np.min([gain_0, gain_2])[0]
         self.exec_logger.debug(f'Setting TX ADC gain automatically to {gain}')
         self.adc_gain = gain
@@ -243,8 +246,8 @@ class Rx(RxAbstract):
     def voltage(self):
         """ Gets the voltage VMN in Volts
         """
-        u0 = AnalogIn(self.ads_voltage, ads.P0).voltage * 1000.
-        u2 = AnalogIn(self.ads_voltage, ads.P2).voltage * 1000.
+        u0 = AnalogIn(self._ads_voltage, ads.P0).voltage * 1000.
+        u2 = AnalogIn(self._ads_voltage, ads.P2).voltage * 1000.
         u = np.max([u0,u2]) * (np.heaviside(u0-u2, 1.) * 2 - 1.) # gets the max between u0 & u2 and set the sign
         self.exec_logger.debug(f'Reading voltages {u0} V and {u2} V on RX. Returning {u} V')
         return u
\ No newline at end of file

logging_setup.py

 import json
-from config import EXEC_LOGGING_CONFIG, DATA_LOGGING_CONFIG, MQTT_LOGGING_CONFIG, MQTT_CONTROL_CONFIG
+from OhmPi.config import EXEC_LOGGING_CONFIG, DATA_LOGGING_CONFIG, MQTT_LOGGING_CONFIG, MQTT_CONTROL_CONFIG
 from os import path, mkdir, statvfs
 from time import gmtime
 import logging
-from mqtt_handler import MQTTHandler
-from compressed_sized_timed_rotating_handler import CompressedSizedTimedRotatingFileHandler
+from OhmPi.mqtt_handler import MQTTHandler
+from OhmPi.compressed_sized_timed_rotating_handler import CompressedSizedTimedRotatingFileHandler
 import sys
 from termcolor import colored
 
@@ -17,7 +17,7 @@ def create_default_logger(name):
     handler = logging.StreamHandler(sys.stdout)
     handler.setFormatter(formatter)
     logger.addHandler(handler)
-    logger.setLevel('debug')
+    logger.setLevel(logging.DEBUG)
     return logger
 
 def setup_loggers(mqtt=True):

measure.py

 import importlib
 import numpy as np
-from logging_setup import create_default_logger
-from config import OHMPI_CONFIG
-controller_module = importlib.import_module(f'{OHMPI_CONFIG["hardware"]["controller"]["model"]}')
-tx_module = importlib.import_module(f'{OHMPI_CONFIG["hardware"]["tx"]["model"]}')
-rx_module = importlib.import_module(f'{OHMPI_CONFIG["hardware"]["rx"]["model"]}')
-mux_module = importlib.import_module(f'{OHMPI_CONFIG["hardware"]["mux"]["model"]}')
+from OhmPi.logging_setup import create_default_logger
+from OhmPi.config import OHMPI_CONFIG
+controller_module = importlib.import_module(f'OhmPi.{OHMPI_CONFIG["hardware"]["controller"]["model"]}')
+tx_module = importlib.import_module(f'OhmPi.{OHMPI_CONFIG["hardware"]["tx"]["model"]}')
+rx_module = importlib.import_module(f'OhmPi.{OHMPI_CONFIG["hardware"]["rx"]["model"]}')
+mux_module = importlib.import_module(f'OhmPi.{OHMPI_CONFIG["hardware"]["mux"]["model"]}')
 TX_CONFIG = tx_module.TX_CONFIG
 RX_CONFIG = rx_module.RX_CONFIG
 MUX_CONFIG = mux_module.MUX_CONFIG

utils.py

 import io
 
-
 def get_platform():
     """Gets platform name and checks if it is a raspberry pi