Adding functionality to include ADC and GPIO responses as objects stored in the device pins.

Author: CreatingNull

CHANGES.rst

@@ -30,6 +30,8 @@ Version 0.5.0
   ``abstractions.NPCPacket``
 * Including the TX packet in the RX response.
   This allows us more freedom to build higher level decoding functions.
+* Adding functionality to include ADC and GPIO responses as objects stored
+  in the device pins.
 
 Version 0.4.0
 -------------

uoshardware/__init__.py

@@ -38,6 +38,10 @@ class UOSCommunicationError(UOSError):
     """Exception while communicating with a UOS Device."""
 
 
+class UOSRuntimeError(UOSError):
+    """General exception for runtime failure, usually indicates misuse."""
+
+
 # Configures the global logger for the library
 # Note: Clients need to initialize logging otherwise no output will be visible.
 logger = logging.getLogger(__name__)

uoshardware/abstractions.py

@@ -1,8 +1,9 @@
 """Module defining the base class and static func for interfaces."""
 from abc import ABCMeta, abstractmethod
 from dataclasses import dataclass, field
+from datetime import datetime
 
-from uoshardware import Persistence, UOSUnsupportedError
+from uoshardware import Persistence, UOSRuntimeError, UOSUnsupportedError, logger
 
 
 @dataclass(frozen=True)
@@ -153,6 +154,15 @@ class ComResult:
     rx_packets: list = field(default_factory=list)
     tx_packet: NPCPacket | None = None
 
+    def get_rx_payload(self, packet_index: int) -> list[int]:
+        """Return just the payload portion of a rx packet."""
+        if len(self.rx_packets) <= packet_index:
+            raise UOSRuntimeError(
+                f"Can't index payload {packet_index} of "
+                f"{len(self.rx_packets)} rx packet(s)."
+            )
+        return self.rx_packets[packet_index][4:-2]
+
 
 @dataclass
 class InstructionArguments:
@@ -255,7 +265,44 @@ def enumerate_devices() -> list:
         )
 
 
-@dataclass(frozen=True)
+@dataclass(init=False)
+class Sample:
+    """A converted response from a reading on a pin."""
+
+    raw_value: int
+    value: float
+    time: datetime
+
+
+@dataclass(init=False)
+class ADCSample(Sample):
+    """ADC specific Sample constructor for ADC readings."""
+
+    def __init__(self, raw_value: list[int], steps: int, reference: float):
+        """Create an ADC Sample."""
+        self.raw_value = int(bytes(raw_value).hex(), 16)  # convert bytes to int
+        self.value = (self.raw_value / steps) * reference
+        self.time = datetime.now()
+        logger.debug(
+            "Constructing ADCSample from `%s` with adc steps `%s` and ref `%s`",
+            self.raw_value,
+            steps,
+            reference,
+        )
+
+
+@dataclass(init=False)
+class DigitalSample(Sample):
+    """Digital specific sample constructor for gpio reads."""
+
+    def __init__(self, raw_value: int):
+        """Create a Digital GPIO Sample."""
+        self.raw_value = raw_value
+        self.value = raw_value
+        self.time = datetime.now()
+
+
+@dataclass
 class Pin:
     """Defines supported features of the pin."""
 
@@ -271,6 +318,10 @@ class Pin:
     pull_down: bool = False
     aliases: list = field(default_factory=list)
 
+    # Values updated during runtime.
+    gpio_reading: DigitalSample | None = None
+    adc_reading: ADCSample | None = None
+
 
 @dataclass(frozen=True)
 class Device:
@@ -279,7 +330,7 @@ class Device:
     name: str
     interfaces: list
     functions_enabled: dict
-    pins: dict = field(default_factory=dict)
+    pins: dict[int, Pin] = field(default_factory=dict)
     aux_params: dict = field(default_factory=dict)
 
     def get_compatible_pins(self, function: UOSFunction) -> dict:
@@ -302,3 +353,55 @@ def get_compatible_pins(self, function: UOSFunction) -> dict:
                 getattr(pin, requirement) for requirement in function.pin_requirements
             )
         }
+
+    def update_adc_samples(self, result: ComResult):
+        """Update the pin samples with the response of a get_adc_input."""
+        if not result.status or len(result.exception) != 0:
+            raise UOSRuntimeError("Can't update ADC samples from a failed response.")
+        if result.tx_packet is None or len(result.rx_packets) < 1:
+            raise UOSRuntimeError("Can't update ADC samples without a valid result.")
+        if (
+            "adc_reference" not in self.aux_params
+            or "adc_resolution" not in self.aux_params
+        ):
+            raise UOSRuntimeError("Device not properly defined for ADC updates.")
+        sample_values = result.get_rx_payload(0)
+        logger.debug("Device returned sampled adc values %s", sample_values)
+        for sample_index, pin in enumerate(result.tx_packet.payload):
+            if pin not in self.pins:
+                raise UOSRuntimeError(
+                    f"Can't update ADC samples on pin {pin} as it's invalid for {self.name}."
+                )
+            self.pins[pin].adc_reading = ADCSample(
+                sample_values[sample_index * 2 : sample_index * 2 + 2],
+                steps=pow(2, self.aux_params["adc_resolution"]),
+                reference=self.aux_params["adc_reference"],
+            )
+            logger.debug(
+                "Setting pin %s adc reading to %s",
+                pin,
+                # This is a false call as it can't be None here.
+                self.pins[pin].adc_reading.value,  # type: ignore
+            )
+
+    def update_gpio_samples(self, result: ComResult):
+        """Update the pin samples with the response of a get_gpio_inpout."""
+        if not result.status or len(result.exception) != 0:
+            raise UOSRuntimeError("Can't update GPIO samples from a failed responsee.")
+        if result.tx_packet is None or len(result.rx_packets) < 1:
+            raise UOSRuntimeError("Can't update GPIO samples without a valid result.")
+        sample_values = result.get_rx_payload(0)
+        logger.debug("Device returned sampled gpio values %s", sample_values)
+        for sample_index, pin in enumerate(sample_values):
+            pin = result.tx_packet.payload[2 * sample_index]
+            if pin not in self.pins:
+                raise UOSRuntimeError(
+                    f"Can't update GPIO samples on pin {pin} as it's invalid for {self.name}."
+                )
+            self.pins[pin].gpio_reading = DigitalSample(sample_values[sample_index])
+            logger.debug(
+                "Setting pin %s gpio reading to %s",
+                pin,
+                # This is a false call as it can't be None here.
+                self.pins[pin].gpio_reading.value,  # type: ignore
+            )

uoshardware/api.py

@@ -156,15 +156,18 @@ def get_gpio_input(
         :param volatility: How volatile should the command be, use constants from uoshardware.
         :return: ComResult object.
         """
-        return self.__execute_instruction(
+        result = self.__execute_instruction(
             UOSFunctions.get_gpio_input,
             InstructionArguments(
-                payload=(pin, 0 if pull_up else 1),
+                payload=(pin, 1 if pull_up else 0),
                 expected_rx_packets=2,
                 check_pin=pin,
                 volatility=volatility,
             ),
         )
+        if result.status:
+            self.device.update_gpio_samples(result)
+        return result
 
     def get_adc_input(
         self,
@@ -175,12 +178,15 @@ def get_adc_input(
         :param pin: The index of the analog pin to read
         :return: ComResult object containing the ADC readings.
         """
-        return self.__execute_instruction(
+        result = self.__execute_instruction(
             UOSFunctions.get_adc_input,
             InstructionArguments(
                 payload=tuple([pin]), expected_rx_packets=2, check_pin=pin
             ),
         )
+        if result.status:  # update the samples in the device.
+            self.device.update_adc_samples(result)
+        return result
 
     def get_system_info(self) -> ComResult:
         """Read the UOS version and device type.

uoshardware/devices/_arduino.py

@@ -86,7 +86,7 @@
             aliases=["A7"],
         ),
     },
-    aux_params={"default_baudrate": 115200},
+    aux_params={"default_baudrate": 115200, "adc_reference": 5, "adc_resolution": 10},
 )
 
 
@@ -99,5 +99,5 @@
         for pin_index, value in _ARDUINO_NANO_3.pins.items()
         if pin_index not in {20, 21}
     },
-    aux_params={"default_baudrate": 115200},
+    aux_params={"default_baudrate": 115200, "adc_reference": 5, "adc_resolution": 10},
 )