Refactor for some name changes of libraries

util/com.py

@@ -0,0 +1,136 @@
+from typing import override
+import serial
+import struct
+import serial.tools.list_ports
+
+from util.interface import ControllerConnection
+
+
+# ┌                                                ┐
+# │         Main Com Class Implementation          │
+# └                                                ┘
+# Below you can find what you were most likely looking for. This is the implementation of the communication with the microcontroller.
+# You may also be interested in the decoder.py and instructions.py file, as the decoding and the hooking / syncing process are
+# implemented there. It is recommended that you do NOT read the test/com.py file, as that one is only there for simulation purposes
+# and is much more complicated than this here, if you are not well versed with Python or are struggling with the basics
+
+# All variables starting in self are bound to the object and can be changed by any consumer of this library. The Com class
+# inherits from the ControllerConnection class (found in interface.py), which implements some of the methods (functions)
+# this class exposes, namely the constructor, set_port_override and get_error. They are not further relevant for the code below
+# though, so you can safely ignore it.
+
+
+class Com(ControllerConnection):
+    def _connection_check(self) -> bool:
+        if self._serial == None:
+            return self._open()
+        if self._serial != None:
+            if not self._serial.is_open:
+                self._serial.open()
+            return True
+        else:
+            return False
+
+    @override
+    def get_comport(self) -> str:
+        """Find the comport the microcontroller has attached to"""
+        if self._port_override != "":
+            return self._port_override
+
+        # Catch all errors and simply return an empty string if search unsuccessful
+        try:
+            # Get an array of all used comports
+            ports = [comport.device for comport in serial.tools.list_ports.comports()]
+
+            # Filter for specific controller
+            for comport in ports:
+                for filter in self._filters:
+                    if filter in comport:
+                        return comport
+        except Exception as e:
+            self._err = e
+
+        return ""
+
+    def _open(self) -> bool:
+        """Open the connection. Internal function, not to be called directly, use connect instead
+
+        Returns:
+            Boolean indicates if connection was successful or not
+        """
+        # Get the com port the controller has connected to
+        comport = self.get_comport()
+
+        # Comport search returns empty string if search unsuccessful
+        if comport != "":
+            # Try to generate a new Serial object with the configuration of this class
+            # self._baudrate contains the baud rate and defaults to 19200
+            try:
+                self._serial = serial.Serial(comport, self._baudrate, timeout=5)
+            except serial.SerialException as e:
+                # If an error occurs, catch it, handle it and store the error
+                # for the UI and return False to indicate failed connection
+                self._err = e
+                return False
+
+            # Connection succeeded, return True
+            return True
+        else:
+            # Haven't found a comport
+            return False
+
+    @override
+    def connect(self) -> bool:
+        """Try to find a comport and connect to the microcontroller. Returns the success as a boolean"""
+        return self._connection_check()
+
+    @override
+    def close(self) -> None:
+        """Close the serial connection, if possible"""
+        if self._serial != None:
+            try:
+                self._serial.close()
+            except:
+                pass
+
+    @override
+    def receive(self, byte_count: int) -> bytes:
+        """Receive bytes from microcontroller over serial. Returns bytes. Might want to decode using functions from lib.decoder"""
+        # Check connection
+        self._connection_check()
+
+        # Ignore this boilerplate (extra code), the body of the if is the only thing important.
+        # The reason for the boilerplate is that the type checker will notice that self._serial can be
+        # None, thus showing errors.
+        if self._serial != None:
+            return self._serial.read(byte_count)
+        else:
+            raise Exception("ERR_CONNECTING")
+
+    @override
+    def send(self, msg: str) -> None:
+        """Send a string over serial connection. Will open a connection if none is available"""
+        # Check connection
+        self._connection_check()
+
+        # Ignore this boilerplate (extra code), the body of the if is the only thing important.
+        # The reason for the boilerplate is that the type checker will notice that self._serial can be
+        # None, thus showing errors.
+        if self._serial != None:
+            self._serial.write(msg.encode())
+        else:
+            raise Exception("ERR_CONNECTING")
+
+    @override
+    def send_float(self, msg: float) -> None:
+        """Send a float number over serial connection"""
+        # Check connection
+        self._connection_check()
+
+        # Ignore this boilerplate (extra code), the body of the if is the only thing important.
+        # The reason for the boilerplate is that the type checker will notice that self._serial can be
+        # None, thus showing errors.
+        if self._serial != None:
+            self._serial.write(bytearray(struct.pack(">f", msg))[0:3])
+        else:
+            raise Exception("ERR_CONNECTING")

util/config.py

@@ -0,0 +1,147 @@
+# This library is used to validate the config file
+
+import configparser
+from typing import List
+
+# Load the config
+config = configparser.ConfigParser()
+config.read("./config.ini")
+
+global first_error
+first_error = True
+
+global is_verbose
+is_verbose = True
+
+
+# Set the verbosity if needed
+def set_verbosity(verbose: bool):
+    global is_verbose
+    is_verbose = verbose
+
+    print("\n", "-" * 20, "\nValidating configuration...\n")
+
+
+def str_to_bool(val: str) -> bool | None:
+    """Convert a string to boolean, converting "True" and "true" to True, same for False
+
+    Args:
+        val: The value to try to convert
+
+    Returns:
+        Returns either a boolean if conversion was successful, or None if not a boolean
+    """
+    return {"True": True, "true": True, "False": False, "false": False}.get(val, None)
+
+
+def read_config(
+    key_0: str,
+    key_1: str,
+    default: str,
+    valid_entries: List[str] = [],
+    type_to_validate: str = "",
+) -> str:
+    """Read the configuration, report potential configuration issues and validate each entry
+
+    Args:
+        key_0: The first key (top level)
+        key_1: The second key (where the actual key-value pair is)
+        default: The default value to return if the check fails
+        valid_entries: [Optiona] The entries that are valid ones to check against
+        type_to_validate: [Optional] Data type to validate
+
+    Returns:
+        [TODO:return]
+    """
+    # Try loading the keys
+    tmp = {}
+    try:
+        tmp = config[key_0]
+    except KeyError:
+        print_config_error(key_0, key_1, "", default, "unknown", index=1)
+        return default
+
+    value = ""
+    try:
+        value = tmp[key_1]
+    except KeyError:
+        print_config_error(key_0, key_1, "", default, "unknown")
+        return default
+
+    if len(value) == 0:
+        print_config_error(key_0, key_1, value, default, "not_empty")
+
+    # Validate input
+    if type_to_validate != "":
+        # Need to validate
+        if type_to_validate == "int":
+            try:
+                int(value)
+            except ValueError:
+                print_config_error(key_0, key_1, value, default, "int")
+                return default
+        if type_to_validate == "float":
+            try:
+                float(value)
+            except ValueError:
+                print_config_error(key_0, key_1, value, default, "float")
+                return default
+
+        if type_to_validate == "bool":
+            if str_to_bool(value) == None:
+                print_config_error(key_0, key_1, value, default, "bool")
+                return default
+
+    if len(valid_entries) > 0:
+        # Need to validate the names
+        try:
+            valid_entries.index(value)
+        except ValueError:
+            print_config_error(
+                key_0, key_1, value, default, "oneof", valid_entries=valid_entries
+            )
+            return default
+
+    return value
+
+
+def print_config_error(
+    key_0: str,
+    key_1: str,
+    value: str,
+    default: str,
+    expected: str,
+    valid_entries: List[str] = [],
+    msg: str = "",
+    index: int = 1,
+):
+    """Print configuration errors to the shell
+
+    Args:
+        key_0: The first key (top level)
+        key_1: The second key (where the actual value is to be found)
+        expected: The data type expected. If unknown key, set to "unknown" and set index; If should be one of, use "oneof" and set valid_entries list
+        msg: The message to print
+        index: The index in the chain (i.e. if key_0 or key_1)
+    """
+    if not is_verbose:
+        return
+
+    print(f" ==> Using default setting ({default}) for {key_0}.{key_1}")
+
+    if expected == "unknown":
+        # The field was unknown
+        print(f'    -> Unknown field "{key_0 if index == 0 else key_1}"')
+    elif expected == "oneof":
+        print(
+            f'    -> Invalid name "{value}". Has to be one of', ", ".join(valid_entries)
+        )
+    elif expected == "not_empty":
+        print("    -> Property is unexpectedly None")
+    elif expected == "bool":
+        print(f'    -> Boolean property expected, but instead found "{value}".')
+    else:
+        print(f"    -> Expected a config option of type {expected}.")
+
+    if msg != "":
+        print(msg)

util/decoder.py

@@ -0,0 +1,24 @@
+import struct
+
+
+# Decoder to decode various sent values from the microcontroller
+class Decoder:
+    # Decode an ascii character
+    def decode_ascii(self, value: bytes) -> str:
+        try:
+            return value.decode()
+        except:
+            return "Error"
+
+    # Decode a float (6 bits)
+    def decode_float(self, value: bytes) -> float:
+        return struct.unpack(">f", bytes.fromhex(str(value, "ascii") + "00"))[0]
+
+    # Decode a float, but with additional offsets
+    def decode_float_long(self, value: bytes) -> float:
+        return struct.unpack(">f", bytes.fromhex(str(value, "ascii") + "0000"))[0]
+
+    # Decode an int
+    def decode_int(self, value: bytes) -> int:
+        # return int.from_bytes(value, 'big')
+        return int(value, base=16)

util/instructions.py

@@ -0,0 +1,139 @@
+import util.decoder
+import time
+
+from util.interface import ControllerConnection
+
+decoder = util.decoder.Decoder()
+
+
+# Class that supports sending instructions to the microcontroller,
+# as well as hooking to data stream according to protocol
+class Instructions:
+    def __init__(self, com: ControllerConnection) -> None:
+        self._com = com
+
+    # Helper method to hook to the data stream according to protocol.
+    # You can specify the sequence that the program listens to to sync up,
+    # as an array of strings, that should each be of length one and only contain
+    # ascii characters
+    def hook(self, instruction: str, sequence: list[str]) -> bool:
+        # Add protection: If we cannot establish connection, refuse to run
+        if not self._com.connect():
+            return False
+
+        # Send instruction to microcontroller to start hooking process
+        self._com.send(instruction)
+
+        # Record start time to respond to timeout
+        start = time.time()
+
+        # The pointer below points to the element in the array which is the next expected character to be received
+        pointer = 0
+
+        # Simply the length of the sequence, since it is both cheaper and cleaner to calculate it once
+        sequence_max = len(sequence)
+
+        # Only run for a limited amount of time
+        while time.time() - start < 5:
+            # Receive and decode a single byte and decode as ASCII
+            data = decoder.decode_ascii(self._com.receive(1))
+            if data == sequence[pointer]:
+                # Increment the pointer (move to next element in the List)
+                pointer += 1
+            else:
+                # Jump back to start
+                pointer = 0
+
+            # If the pointer has reached the end of the sequence, return True, as now the hook was successful
+            if pointer == sequence_max:
+                return True
+
+        # If we time out, which is the only way in which this code can be reached, return False
+        return False
+
+    # Used to hook to the main data stream, as that hooking mechanism is different
+    def hook_main(self) -> bool:
+        # Record start time to respond to timeout
+        start = time.time()
+
+        # Wait to find a CR character (enter)
+        char = decoder.decode_ascii(self._com.receive(1))
+        while char != "\n":
+            # Check for timeout
+            if time.time() - start > 3:
+                return False
+
+            # Set the next character by receiving and decoding it as ASCII
+            char = decoder.decode_ascii(self._com.receive(1))
+
+        # Store the position in the hooking process
+        state = 0
+        distance = 0
+
+        # While we haven't timed out and have not reached the last state execute
+        # The last state indicates that the sync was successful
+        while time.time() - start < 5 and state < 3:
+            # Receive the next char and decode it as ASCII
+            char = decoder.decode_ascii(self._com.receive(1))
+
+            # The character we look for when syncing is Space (ASCII char 32 (decimal))
+            # It is sent every 4 bits. If we have received 3 with the correct distance from
+            # the previous in a row, we are synced
+            if char == " ":
+                if distance == 4:
+                    state += 1
+                distance = 0
+            else:
+                if distance > 4:
+                    state = 0
+                    distance = 0
+                else:
+                    distance += 1
+
+        # Read 5 more bits to correctly sync up
+        self._com.receive(5)
+
+        return state == 3
+
+    # Private helper method to transmit data using the necessary protocols
+    def _change_data(
+        self,
+        instruction: str,
+        readback: list[str],
+        data: list[float],
+        readback_length: int,
+    ) -> None:
+        # Hook to stream
+        if self.hook(instruction, readback):
+            # Transmit data
+            while len(data) > 0:
+                # If we received data back, we can send more data, i.e. from this we know
+                # the controller has received the data
+                # If not, we close the connection and create an exception
+                if self._com.receive(readback_length) != "":
+                    self._com.send_float(data.pop(0))
+                else:
+                    self._com.close()
+                    raise Exception(
+                        "Failed to transmit data. No response from controller"
+                    )
+            self._com.close()
+        else:
+            self._com.close()
+            raise ConnectionError(
+                "Failed to hook to controller data stream. No fitting response received"
+            )
+
+    # Abstraction of the _change_data method specifically designed to change the entire config
+    def change_config(self, new_config: list[float]) -> None:
+        try:
+            self._change_data("PR", ["\n", "P", "R", "\n"], new_config, 3)
+        except Exception as e:
+            raise e
+
+    # Abstraction of the _change_data method specifically designed to change only the configured temperature
+    def change_temperature(self, temperatures: list[float]) -> None:
+        try:
+            self._change_data("PT", ["\n", "P", "T", "\n"], temperatures, 3)
+        except Exception as e:
+            raise e

util/interface.py

@@ -0,0 +1,66 @@
+from abc import ABC, abstractmethod
+from typing import Optional
+import serial
+
+# If you don't know what OOP is, you can safely ignore this file
+#
+# The below class is abstract to have a consistent, targetable interface
+# for both the real connection module and the simulation module
+#
+# For the interested, a quick rundown of what the benefits are of doing it this way is:
+# This class provides a way to have two wholly different implementations that have
+# the same function interface (i.e. all functions take the same arguments)
+#
+# Another benefit of having classes is that we can pass a single instance around to
+# various components and have one shared instance that all can modify, reducing some
+# overhead.
+#
+# The actual implementation of most functions (called methods in OOP) are implemented
+# in the Com class below.
+
+
+class ControllerConnection(ABC):
+    def __init__(
+        self, baudrate: Optional[int] = 19200, filters: Optional[list[str]] = None
+    ) -> None:
+        self._serial: Optional[serial.Serial] = None
+        self._filters = (
+            filters
+            if filters != None
+            else ["USB-Serial Controller", "Prolific USB-Serial Controller"]
+        )
+        self._port_override = ""
+        self._baudrate = baudrate if baudrate != None else 19200
+        self._err = None
+
+    def set_port_override(self, override: str) -> None:
+        """Set the port override, to disable port search"""
+        if override != "" and override != "None":
+            self._port_override = override
+
+    def get_error(self) -> serial.SerialException | None:
+        return self._err
+
+    @abstractmethod
+    def get_comport(self) -> str:
+        pass
+
+    @abstractmethod
+    def connect(self) -> bool:
+        pass
+
+    @abstractmethod
+    def close(self) -> None:
+        pass
+
+    @abstractmethod
+    def receive(self, byte_count: int) -> bytes:
+        pass
+
+    @abstractmethod
+    def send(self, msg: str) -> None:
+        pass
+
+    @abstractmethod
+    def send_float(self, msg: float) -> None:
+        pass

util/test/com.py

@@ -0,0 +1,274 @@
+"""
+Library to be used in standalone mode (without microcontroller, for testing functionality)
+It simulates the behviour of an actual microcontroller being connected
+"""
+
+
+# ────────────────────────────────────────────────────────────────────
+# ┌                                                ┐
+# │             Testing Module For Com             │
+# └                                                ┘
+# This file contains a Com class that can be used to test the functionality
+# even without a microcontroller. It is not documented in a particularly
+# beginner-friendly way, nor is the code written with beginner-friendliness
+# in mind. It is the most complicated piece of code of the entire application
+# ────────────────────────────────────────────────────────────────────
+
+
+
+
+
+
+
+
+
+
+
+# Just be warned, more OOP concepts and less documentation can be found here.
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+# Code starts here
+# ────────────────────────────────────────────────────────────────────
+
+from typing import List, Optional, override
+import queue
+import random
+import time
+import struct
+
+from util.interface import ControllerConnection
+
+# All double __ prefixed properties and methods are not available in the actual impl
+
+instruction_lut: dict[str, list[str]] = {
+    "PR": ["\n", "P", "R", "\n"],
+    "PT": ["\n", "P", "T", "\n"],
+    "RD": ["\n", "R", "D", "\n"],
+    "NM": ["\n", "N", "M", "\n"],
+    "FM": ["\n", "F", "M", "\n"],
+}
+
+reconfig = ["a", "b", "c", "t"]
+
+
+class SimulationError(Exception):
+    pass
+
+
+class SensorConfig:
+    a: float
+    b: float
+    c: float
+    t: float
+
+    def __init__(
+        self, a: float = 20, b: float = 30, c: float = 10, t: float = 55
+    ) -> None:
+        self.a = a
+        self.b = b
+        self.c = c
+        self.t = t
+
+
+class Com(ControllerConnection):
+    def __init__(
+        self, fail_sim: int, baudrate: int = 19200, filters: Optional[list[str]] = None
+    ) -> None:
+        # Calling the constructor of the super class to assign defaults
+        print("\n\nWARNING: Using testing library for communication!\n\n")
+        super().__init__(baudrate, filters)
+
+        # Initialize queue with values to be sent on call of recieve
+        self.__simulated_data: queue.Queue[bytes] = queue.Queue()
+        self.__simulated_data_remaining = 0
+
+        self.__reconf_sensor = 0
+        self.__reconf_step = 0
+        self.__fail_sim = fail_sim
+
+        self.__config: List[SensorConfig] = [
+            SensorConfig(),
+            SensorConfig(),
+            SensorConfig(),
+            SensorConfig(),
+        ]
+
+        # Initially, we are in normal mode (which leads to slower data intervals)
+        self.__mode = "NM"
+
+    @override
+    def get_comport(self) -> str:
+        return "Sim" if self._port_override == "" else self._port_override
+
+    @override
+    def connect(self) -> bool:
+        # Randomly return false in 1 in fail_sim ish cases
+        if random.randint(0, self.__fail_sim) == 0:
+            print("Simulating error to connect")
+            return False
+        return True
+
+    @override
+    def close(self) -> None:
+        pass
+
+    @override
+    def receive(self, byte_count: int) -> bytes:
+        data = []
+        # If queue is too short, refill it
+        if self.__simulated_data_remaining < byte_count:
+            self.__fill_queue()
+
+        for _ in range(byte_count):
+            if self.__mode == "NM":
+                time.sleep(0.005)
+            try:
+                data.append(self.__simulated_data.get_nowait())
+                self.__simulated_data_remaining -= 1
+            except Exception as e:
+                print("ERROR: Simulation could not continue")
+                raise SimulationError(
+                    "Simulation encountered an error with the simulation queue. The error encountered: \n"
+                    + str(e)
+                )
+        return b"".join(data)
+
+    @override
+    def send(self, msg: str) -> None:
+        # Using LUT to reference
+        readback = instruction_lut.get(msg)
+        if readback != None:
+            for i in range(len(readback)):
+                self.__add_ascii_char(readback[i])
+        if msg == "RD":
+            self.__set_read_data_data()
+        elif msg == "PR":
+            self.__reconf_sensor = 0
+            self.__reconf_step = 0
+            self.__add_ascii_char("a")
+            self.__add_ascii_char("0")
+            self.__add_ascii_char("\n")
+
+    def __set_read_data_data(self) -> None:
+        # Send data for all four sensors
+        for i in range(4):
+            self.__add_float_as_hex(self.__config[i].a)
+            self.__add_ascii_char(" ")
+            self.__add_float_as_hex(self.__config[i].b)
+            self.__add_ascii_char(" ")
+            self.__add_float_as_hex(self.__config[i].c)
+            self.__add_ascii_char(" ")
+            self.__add_float_as_hex(self.__config[i].t)
+            self.__add_ascii_char("\n")
+
+    @override
+    def send_float(self, msg: float) -> None:
+        if self.__reconf_step == 0:
+            self.__config[self.__reconf_sensor].a = msg
+        elif self.__reconf_step == 1:
+            self.__config[self.__reconf_sensor].b = msg
+        elif self.__reconf_step == 2:
+            self.__config[self.__reconf_sensor].c = msg
+        elif self.__reconf_step == 3:
+            self.__config[self.__reconf_sensor].t = msg
+
+        if self.__reconf_step == 3:
+            self.__reconf_step = 0
+            self.__reconf_sensor += 1
+        else:
+            self.__reconf_step += 1
+
+        if self.__reconf_sensor == 4:
+            return
+
+        self.__add_ascii_char(reconfig[self.__reconf_step])
+        self.__add_ascii_char(str(self.__reconf_sensor))
+        self.__add_ascii_char("\n")
+
+    def __fill_queue(self):
+        # Simulate a full cycle
+        for _ in range(4):
+            self.__add_integer_as_hex(self.__generate_random_int(200))
+            self.__simulated_data.put(bytes(" ", "ascii"))
+            self.__add_float_as_hex(self.__generate_random_float(50))
+            self.__simulated_data.put(bytes(" ", "ascii"))
+            self.__simulated_data_remaining += 2
+        for _ in range(3):
+            self.__add_integer_as_hex(self.__generate_random_int(65535))
+            self.__simulated_data.put(bytes(" ", "ascii"))
+            self.__simulated_data_remaining += 1
+        self.__add_integer_as_hex(self.__generate_random_int(65535))
+        self.__simulated_data.put(bytes("\n", "ascii"))
+        self.__simulated_data_remaining += 1
+
+    def __generate_random_int(self, max: int) -> int:
+        return random.randint(0, max)
+
+    def __generate_random_float(self, max: int) -> float:
+        return random.random() * max
+
+    def __add_ascii_char(self, ascii_string: str):
+        self.__simulated_data.put(ord(ascii_string).to_bytes(1))
+        self.__simulated_data_remaining += 1
+
+    def __add_two_byte_value(self, c: int):
+        """putchhex
+
+        Args:
+            c: The char (as integer)
+        """
+        # First nibble (high)
+        high_nibble = (c >> 4) & 0x0F
+        high_char = chr(high_nibble + 48 if high_nibble < 10 else high_nibble + 55)
+        self.__simulated_data.put(high_char.encode())
+
+        # Second nibble (low)
+        low_nibble = c & 0x0F
+        low_char = chr(low_nibble + 48 if low_nibble < 10 else low_nibble + 55)
+        self.__simulated_data.put(low_char.encode())
+        self.__simulated_data_remaining += 2
+
+    def __add_integer_as_hex(self, c: int):
+        """Writes the hexadecimal representation of the high and low bytes of integer `c` (16-bit) to the simulated serial port."""
+        if not (0 <= c <= 0xFFFF):
+            raise ValueError("Input must be a 16-bit integer (0–65535)")
+
+        # Get high byte (most significant byte)
+        hi_byte = (c >> 8) & 0xFF
+        # Get low byte (least significant byte)
+        lo_byte = c & 0xFF
+
+        # Call putchhex for the high byte and low byte
+        self.__add_two_byte_value(hi_byte)
+        self.__add_two_byte_value(lo_byte)
+
+    def __add_float_as_hex(self, f: float):
+        """Converts a float to its byte representation and sends the bytes using putchhex."""
+        # Pack the float into bytes (IEEE 754 format)
+        packed = struct.pack(">f", f)  # Big-endian format (network byte order)
+
+        # Unpack the bytes into 3 bytes: high, mid, low
+        high, mid, low = packed[0], packed[1], packed[2]
+
+        # Send each byte as hex
+        self.__add_two_byte_value(high)
+        self.__add_two_byte_value(mid)
+        self.__add_two_byte_value(low)