Get test library running

This commit is contained in:
2025-06-15 12:01:59 +02:00
parent b00466c5dd
commit b01232b552
5 changed files with 165 additions and 48 deletions

View File

@@ -3,7 +3,7 @@ Library to be used in standalone mode (without microcontroller, for testing func
It simulates the behviour of an actual microcontroller being connected
"""
from typing import Optional
from typing import List, Optional
import queue
import random
import time
@@ -26,11 +26,28 @@ instruction_lut: dict[str, list[str]] = {
"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(ComSuperClass):
def __init__(
self, baudrate: int = 19200, filters: Optional[list[str]] = None
@@ -43,6 +60,16 @@ class Com(ComSuperClass):
self.__simulated_data: queue.Queue[bytes] = queue.Queue()
self.__simulated_data_remaining = 0
self.__reconf_sensor = 0
self.__reconf_step = 0
self.__config: List[SensorConfig] = [
SensorConfig(),
SensorConfig(),
SensorConfig(),
SensorConfig(),
]
# Initially, we are in normal mode (which leads to slower data intervals)
self.__mode = "NM"
@@ -71,7 +98,7 @@ class Com(ComSuperClass):
for _ in range(byte_count):
if self.__mode == "NM":
time.sleep(0.001)
time.sleep(0.005)
try:
data.append(self.__simulated_data.get_nowait())
self.__simulated_data_remaining -= 1
@@ -81,41 +108,57 @@ class Com(ComSuperClass):
"Simulation encountered an error with the simulation queue. The error encountered: \n"
+ str(e)
)
return b''.join(data)
return b"".join(data)
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.__simulated_data.put(bytes(readback[i], "ascii"))
self.__add_ascii_char(readback[i])
if msg == "RD":
# Handle ReadData readback
# self.__simulated_data.put(ord(""))
pass
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")
def send_float(self, msg: float) -> None:
# Encode float as 8 bytes (64 bit)
ba = struct.pack("d", msg)
for byte in ba:
self.__simulated_data.put(byte.to_bytes())
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
def __fill_queue_alternative(self):
for _ in range(4):
for _ in range(4):
self.__simulated_data.put(random.randbytes(1))
self.__simulated_data.put(bytes(" ", "ascii"))
for _ in range(6):
self.__simulated_data.put(random.randbytes(1))
self.__simulated_data.put(bytes(" ", "ascii"))
for _ in range(3):
for _ in range(4):
self.__simulated_data.put(random.randbytes(1))
self.__simulated_data.put(bytes(" ", "ascii"))
for _ in range(4):
self.__simulated_data.put(random.randbytes(1))
self.__simulated_data.put(bytes("\n", "ascii"))
self.__simulated_data_remaining = 68
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):
for _ in range(4):
@@ -127,10 +170,10 @@ class Com(ComSuperClass):
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 += 4
print("Length:", self.__simulated_data_remaining)
self.__simulated_data_remaining += 1
def __generate_random_int(self, max: int) -> int:
return random.randint(0, max)
@@ -138,11 +181,50 @@ class Com(ComSuperClass):
def __generate_random_float(self, max: int) -> float:
return random.random() * max
def __add_character_as_hex(self, data: str):
pass
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_integer_as_hex(self, data: int):
pass
def __add_two_byte_value(self, c: int):
"""putchhex
def __add_float_as_hex(self, data: float):
pass
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 (065535)")
# 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)