diff --git a/BaseController/create_controllers.py b/BaseController/create_controllers.py index ac2530b..639dca5 100644 --- a/BaseController/create_controllers.py +++ b/BaseController/create_controllers.py @@ -23,3 +23,22 @@ def flight_computer_rev2_controller() -> Controller: sensor_frame_size=120, sensor_data_file="output/flight_comp_rev2_sensor_data.txt" ) + +def engine_controller_rev5_controller() -> Controller: + poll_codes = { + sensor.poll_code: BaseSensor( + sensor.short_name, + sensor.name, + sensor.size, + sensor.data_type, + sensor.unit + ) for sensor in create_sensors.engine_controller_rev5_sensors() + } + + return Controller( + id=b"\x08", + name="Liquid Engine Controller (L0002 Rev 5.0)", + poll_codes=poll_codes, + sensor_frame_size=44, + sensor_data_file="output/engine_ctrl_rev5_sensor_data.txt" + ) diff --git a/EngineController/__init__.py b/EngineController/__init__.py new file mode 100644 index 0000000..18f382a --- /dev/null +++ b/EngineController/__init__.py @@ -0,0 +1,12 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2025 Sun Devil Rocketry + +from .engine_state import EngineState, ENGINE_STATE_NAMES, Valve, VALVES, parse_valve_byte +from .engine_controller import ( + CONTROLLER_ID, CONTROLLER_NAME, + TelemetryData, + hotfire_abort, preflight_purge, fill_chill, standby, hotfire, + stop_hotfire, stop_purge, lox_purge, manual_mode, + get_state, telemetry_request, +) +from .flash_extract import EngineControllerFlashFrame, flash_extract diff --git a/EngineController/engine_controller.py b/EngineController/engine_controller.py new file mode 100644 index 0000000..a579163 --- /dev/null +++ b/EngineController/engine_controller.py @@ -0,0 +1,202 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2025 Sun Devil Rocketry + +from dataclasses import dataclass +from typing import Dict, Optional + +from SDECv2.SerialController import SerialObj +from SDECv2.Sensor import create_sensors +from SDECv2.Sensor.util import process_data_bytes +from .engine_state import EngineState, parse_valve_byte + +CONTROLLER_ID = b"\x08" +CONTROLLER_NAME = "Liquid Engine Controller (L0002 Rev 5.0)" + +_ACK = b"\x95" +_NO_ACK = b"\x98" + +_OPCODE_ABORT = b"\x90" +_OPCODE_PREFLIGHT_PURGE = b"\x91" +_OPCODE_FILL_CHILL = b"\x92" +_OPCODE_STANDBY = b"\x93" +_OPCODE_HOTFIRE = b"\x94" +_OPCODE_TELREQ = b"\x96" +_OPCODE_STOP_PURGE = b"\x97" +_OPCODE_GET_STATE = b"\x99" +_OPCODE_STOP_HOTFIRE = b"\x9A" +_OPCODE_LOX_PURGE = b"\x9B" +_OPCODE_MANUAL = b"\x9E" + +_SENSOR_DUMP_SIZE = 40 # 10 sensors × 4 bytes each + + +@dataclass +class TelemetryData: + """Snapshot of engine controller telemetry. + + :param sensor_readings: Mapping of sensor short-name to converted + value in physical units (psi / lb / °C). + :type sensor_readings: Dict[str, float] + :param valve_states: Mapping of valve name to its current physical + state string (``"OPEN"`` or ``"CLOSED"``). + :type valve_states: Dict[str, str] + """ + sensor_readings: Dict[str, float] + valve_states: Dict[str, str] + + +def _send_state_command(serial: SerialObj, opcode: bytes) -> bool: + """Send a single-byte opcode and return whether the controller acknowledged. + + :param serial: Open serial connection to the engine controller. + :type serial: SerialObj + :param opcode: 1-byte command opcode. + :type opcode: bytes + :returns: ``True`` if the controller replied with ``ACK`` (``0x95``). + :rtype: bool + """ + serial.send(opcode) + return serial.read() == _ACK + + +def hotfire_abort(serial: SerialObj) -> bool: + """Send the abort command, transitioning the engine to ``ABORT`` state. + + :param serial: Open serial connection to the engine controller. + :type serial: SerialObj + :returns: ``True`` on acknowledgement. + :rtype: bool + """ + return _send_state_command(serial, _OPCODE_ABORT) + +def preflight_purge(serial: SerialObj) -> bool: + """Initiate the pre-fire purge sequence (``PRE_FIRE_PURGE`` state). + + :param serial: Open serial connection to the engine controller. + :type serial: SerialObj + :returns: ``True`` on acknowledgement. + :rtype: bool + """ + return _send_state_command(serial, _OPCODE_PREFLIGHT_PURGE) + +def fill_chill(serial: SerialObj) -> bool: + """Initiate the fill-and-chill sequence (``FILL_AND_CHILL`` state). + + :param serial: Open serial connection to the engine controller. + :type serial: SerialObj + :returns: ``True`` on acknowledgement. + :rtype: bool + """ + return _send_state_command(serial, _OPCODE_FILL_CHILL) + +def standby(serial: SerialObj) -> bool: + """Transition the engine to ``STANDBY`` state. + + :param serial: Open serial connection to the engine controller. + :type serial: SerialObj + :returns: ``True`` on acknowledgement. + :rtype: bool + """ + return _send_state_command(serial, _OPCODE_STANDBY) + +def hotfire(serial: SerialObj) -> bool: + """Initiate ignition, transitioning the engine to ``FIRE`` state. + + :param serial: Open serial connection to the engine controller. + :type serial: SerialObj + :returns: ``True`` on acknowledgement. + :rtype: bool + """ + return _send_state_command(serial, _OPCODE_HOTFIRE) + +def stop_hotfire(serial: SerialObj) -> bool: + """Terminate the engine burn. + + :param serial: Open serial connection to the engine controller. + :type serial: SerialObj + :returns: ``True`` on acknowledgement. + :rtype: bool + """ + return _send_state_command(serial, _OPCODE_STOP_HOTFIRE) + +def stop_purge(serial: SerialObj) -> bool: + """Stop the post-fire purge, transitioning the engine to ``DISARM`` state. + + :param serial: Open serial connection to the engine controller. + :type serial: SerialObj + :returns: ``True`` on acknowledgement. + :rtype: bool + """ + return _send_state_command(serial, _OPCODE_STOP_PURGE) + +def lox_purge(serial: SerialObj) -> bool: + """Initiate the LOX tank purge sequence. + + :param serial: Open serial connection to the engine controller. + :type serial: SerialObj + :returns: ``True`` on acknowledgement. + :rtype: bool + """ + return _send_state_command(serial, _OPCODE_LOX_PURGE) + +def manual_mode(serial: SerialObj) -> bool: + """Enter manual valve control mode (``MANUAL`` state). + + :param serial: Open serial connection to the engine controller. + :type serial: SerialObj + :returns: ``True`` on acknowledgement. + :rtype: bool + """ + return _send_state_command(serial, _OPCODE_MANUAL) + + +def get_state(serial: SerialObj) -> Optional[EngineState]: + """Query the current engine state from the controller. + + :param serial: Open serial connection to the engine controller. + :type serial: SerialObj + :returns: The current :class:`~SDECv2.EngineController.engine_state.EngineState`, + or ``None`` on timeout, ``NACK``, or an unrecognized state byte. + :rtype: Optional[EngineState] + """ + serial.send(_OPCODE_GET_STATE) + response = serial.read() + if not response or response == _NO_ACK: + return None + try: + return EngineState(response[0]) + except ValueError: + return None + + +def telemetry_request(serial: SerialObj) -> Optional[TelemetryData]: + """Request a snapshot of all sensor readings and valve states. + + Sends opcode ``0x96``, waits for ``ACK`` (``0x95``), then reads + 40 bytes of sensor data followed by 1 byte of valve state. + Conversion functions are applied, so all sensor values are in + physical units (psi / lb / °C). + + :param serial: Open serial connection to the engine controller. + :type serial: SerialObj + :returns: :class:`TelemetryData` containing converted sensor readings + and decoded valve states, or ``None`` if the controller did not + acknowledge. + :rtype: Optional[TelemetryData] + """ + serial.send(_OPCODE_TELREQ) + if serial.read() != _ACK: + return None + + sensor_data_bytes = serial.read(_SENSOR_DUMP_SIZE) + valve_byte = serial.read() + + sensors = create_sensors.engine_controller_rev5_sensors() + sensor_readings: Dict[str, float] = {} + for sensor in sensors: + raw = sensor_data_bytes[sensor.offset: sensor.offset + sensor.size] + value = process_data_bytes(raw, sensor.data_type, sensor.convert_data) + sensor_readings[sensor.short_name] = value + + valve_states = parse_valve_byte(valve_byte) + return TelemetryData(sensor_readings=sensor_readings, valve_states=valve_states) diff --git a/EngineController/engine_state.py b/EngineController/engine_state.py new file mode 100644 index 0000000..e708f71 --- /dev/null +++ b/EngineController/engine_state.py @@ -0,0 +1,91 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2025 Sun Devil Rocketry + +from dataclasses import dataclass +from enum import IntEnum +from typing import Dict + + +class EngineState(IntEnum): + """Engine controller state machine states. + + States progress through the nominal hotfire sequence: + ``INITIALIZATION`` → ``READY`` → ``PRE_FIRE_PURGE`` → ``FILL_AND_CHILL`` + → ``STANDBY`` → ``FIRE`` → ``DISARM`` → ``POST_FIRE``. + + ``MANUAL`` and ``ABORT`` are reachable from any state. + """ + INITIALIZATION = 0x00 + READY = 0x01 + PRE_FIRE_PURGE = 0x02 + FILL_AND_CHILL = 0x03 + STANDBY = 0x04 + FIRE = 0x05 + DISARM = 0x06 + POST_FIRE = 0x07 + MANUAL = 0x08 + ABORT = 0x09 + + +#: Human-readable display names for each :class:`EngineState`. +ENGINE_STATE_NAMES: Dict[EngineState, str] = { + EngineState.INITIALIZATION: "Initialization", + EngineState.READY: "Ready", + EngineState.PRE_FIRE_PURGE: "Pre-Fire Purge", + EngineState.FILL_AND_CHILL: "Fill and Chill", + EngineState.STANDBY: "Standby", + EngineState.FIRE: "Fire", + EngineState.DISARM: "Disarm", + EngineState.POST_FIRE: "Post-Fire", + EngineState.MANUAL: "Manual", + EngineState.ABORT: "Abort", +} + + +@dataclass(frozen=True) +class Valve: + """Metadata for a single solenoid valve. + + :param name: Valve identifier string (e.g. ``"oxPress"``). + :param on_state: Physical state when the valve bit is **set** + (e.g. ``"OPEN"`` or ``"CLOSED"``). + :param off_state: Physical state when the valve bit is **clear**. + """ + name: str + on_state: str + off_state: str + + +#: All eight valves keyed by 1-indexed bit position in the telemetry valve byte. +#: Bit *n* maps to ``(1 << (n - 1))`` in the byte returned by +#: :func:`~SDECv2.EngineController.engine_controller.telemetry_request`. +VALVES: Dict[int, Valve] = { + 1: Valve("oxPress", "OPEN", "CLOSED"), + 2: Valve("fuelPress", "OPEN", "CLOSED"), + 3: Valve("oxVent", "CLOSED", "OPEN"), + 4: Valve("fuelVent", "CLOSED", "OPEN"), + 5: Valve("oxPurge", "CLOSED", "OPEN"), + 6: Valve("fuelPurge", "CLOSED", "OPEN"), + 7: Valve("oxMain", "OPEN", "CLOSED"), + 8: Valve("fuelMain", "OPEN", "CLOSED"), +} + + +def parse_valve_byte(valve_byte: bytes) -> Dict[str, str]: + """Decode the 1-byte valve bitmask from telemetry into a valve name → state mapping. + + :param valve_byte: Single byte received from the engine controller where + each bit represents one valve's state (bit 0 = valve 1, etc.). + :type valve_byte: bytes + :returns: Mapping of valve name to its physical state string + (``"OPEN"`` or ``"CLOSED"``). + :rtype: Dict[str, str] + """ + valve_int = valve_byte[0] + states: Dict[str, str] = {} + for bit_pos, valve in VALVES.items(): + if valve_int & (1 << (bit_pos - 1)): + states[valve.name] = valve.on_state + else: + states[valve.name] = valve.off_state + return states diff --git a/EngineController/flash_extract.py b/EngineController/flash_extract.py new file mode 100644 index 0000000..55d48ed --- /dev/null +++ b/EngineController/flash_extract.py @@ -0,0 +1,122 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2025 Sun Devil Rocketry + +import struct +import pandas as pd +from dataclasses import dataclass +from typing import Dict, List + +from SDECv2.SerialController import SerialObj +from SDECv2.Sensor import create_sensors +from SDECv2.Sensor.util import process_data_bytes + +FLASH_SIZE = 524288 # 512 KB + +# Frame layout: 4-byte uint32 timestamp + 10 × int32 sensor values = 44 bytes +_FRAME_FORMAT = " List[EngineControllerFlashFrame]: + """Download and parse the full flash contents of the engine controller. + + Sends opcode ``0x22`` with sub-command ``0xC0``, then reads 512 KB of + flash in 512-byte chunks. Each 44-byte frame is structured as:: + + uint32 timestamp_ms (4 bytes) + int32 pt0 … lc (10 × 4 bytes) + + Conversion functions are applied so all returned sensor values are in + physical units (psi / lb / °C). Frames whose timestamp equals + ``0xFFFFFFFF`` (erased flash) are discarded. + + :param serial: Open serial connection to the engine controller. + :type serial: SerialObj + :param store_data: When ``True``, write the parsed frames to *output_path* + as a CSV file. + :type store_data: bool + :param output_path: Destination path for the CSV output. + :type output_path: str + :returns: List of parsed :class:`EngineControllerFlashFrame` objects in + chronological order. + :rtype: List[EngineControllerFlashFrame] + """ + serial.send(_OPCODE_FLASH) + serial.send(_SUBCODE_EXTRACT) + + num_chunks = FLASH_SIZE // 512 + flash_bytes = bytearray() + for i in range(num_chunks): + if i % 128 == 0: + print(f"Reading block {i} ...") + chunk = serial.read(num_bytes=512) + num_received = len(chunk) + if num_received == 0: + print(f"[{i + 1}/{num_chunks}] Timeout: flash appears empty") + elif num_received != 512: + print(f"[{i + 1}/{num_chunks}] Partial read ({num_received} bytes)") + flash_bytes.extend(chunk) + + print(f"{len(flash_bytes)} bytes received from flash") + + # Consume the trailing status byte + serial.read() + + sensors = create_sensors.engine_controller_rev5_sensors() + frames: List[EngineControllerFlashFrame] = [] + frame_dicts = [] + + offset = 0 + while offset + _FRAME_SIZE <= len(flash_bytes): + raw = struct.unpack(_FRAME_FORMAT, flash_bytes[offset: offset + _FRAME_SIZE]) + offset += _FRAME_SIZE + + timestamp_ms = raw[0] + if timestamp_ms == _EMPTY_TIMESTAMP: + continue + + raw_sensor_values = raw[1:] + sensor_readings: Dict[str, float] = {} + for sensor, raw_val in zip(sensors, raw_sensor_values): + value = process_data_bytes( + raw_val.to_bytes(sensor.size, byteorder="little", signed=True), + sensor.data_type, + sensor.convert_data, + ) + sensor_readings[sensor.short_name] = value + + frames.append(EngineControllerFlashFrame(timestamp_ms, sensor_readings)) + if store_data: + frame_dicts.append({"timestamp_ms": timestamp_ms, **sensor_readings}) + + if store_data and frame_dicts: + pd.DataFrame(frame_dicts).to_csv(output_path, index=False) + print(f"Flash data saved to {output_path}") + + print(f"{len(frames)} valid frames parsed") + return frames diff --git a/README.md b/README.md index 497c9db..063dea8 100644 --- a/README.md +++ b/README.md @@ -1,16 +1,31 @@ +### Summary SDECv2 is a recreation/upgrade of legacy SDEC. Replacing SDEC as of v2.0.0, with the following features: - sensor dump - dashboard dump - preset management - flash extract -Before running: -- pip install -e . -- add the following to your .vscode/settings.json to prevent errors on SDECv2 internal imports +### Setup +Before running, compile the code: +``` +pip install -e . +``` +add the following to your .vscode/settings.json to prevent errors on SDECv2 internal imports ``` { "python.analysis.extraPaths": [ "${workspaceFolder}/.." ] } +``` + +### Testing + +#### Run individual file test: +``` +python -m Testing.. +``` +Example: This line below runs test_engine_controller.py test +``` +python -m Testing.EngineController.test_engine_controller ``` \ No newline at end of file diff --git a/Sensor/conv_functions.py b/Sensor/conv_functions.py index e9c6891..23b19ca 100644 --- a/Sensor/conv_functions.py +++ b/Sensor/conv_functions.py @@ -31,4 +31,80 @@ def imu_gyro(readout): return float(signed_int) / (gyro_sensitivity) def baro_press(readout): - return readout * 0.001 \ No newline at end of file + return readout * 0.001 + +def _adc_to_voltage(readout: int) -> float: + """Convert a 16-bit ADC readout to a voltage (3.3 V reference). + + :param readout: Raw 16-bit unsigned ADC value. + :type readout: int + :returns: Voltage in volts. + :rtype: float + """ + return readout * (3.3 / 65536.0) + +def pt_pressure(readout: int) -> float: + """Convert a raw ADC reading from an amplified pressure transducer to psi. + + Uses an instrumentation amplifier with Rref = 100 kΩ and Rgain = 3.3 kΩ + (gain ≈ 31.3×), and a full-scale transducer output of 0.1 V → 1000 psi. + + :param readout: Raw 16-bit unsigned ADC value. + :type readout: int + :returns: Pressure in psi. + :rtype: float + """ + voltage = _adc_to_voltage(readout) + gain = 1.0 + (100.0 / 3.3) + max_voltage = gain * 0.1 + return voltage * (1000.0 / max_voltage) + +def pt_pressure_5V(readout: int) -> float: + """Convert a raw ADC reading from a 5 V pressure transducer to psi. + + Assumes a linear 0–5 V output corresponding to 0–2000 psi. + + :param readout: Raw 16-bit unsigned ADC value. + :type readout: int + :returns: Pressure in psi. + :rtype: float + """ + voltage = _adc_to_voltage(readout) + return voltage * (2000.0 / 5.0) + +def tc_temp(readout: int) -> float: + """Convert a raw thermocouple readout to degrees Celsius. + + The thermocouple IC encodes temperature across two bytes: + the upper byte carries the integer part (×16) and the lower + byte carries the fractional part (÷16). Bit 15 is the sign. + + :param readout: Raw 16-bit thermocouple value (upper byte = integer, + lower byte = fraction). + :type readout: int + :returns: Temperature in °C. + :rtype: float + """ + upper_byte = (readout & 0xFF00) >> 8 + lower_byte = readout & 0x00FF + negative = (upper_byte & 0x80) == 0x80 + temp = float(upper_byte) * 16.0 + float(lower_byte) / 16.0 + if negative: + temp -= 4096.0 + return temp + +def loadcell_force(readout: int) -> float: + """Convert a raw ADC reading from the load cell to force in lb. + + Uses the same amplifier circuit as :func:`pt_pressure` + (Rref = 100 kΩ, Rgain = 3.3 kΩ) and assumes a sensitivity of + 1000 lb at 0.1 V pre-amplification. + + :param readout: Raw 16-bit unsigned ADC value. + :type readout: int + :returns: Force in lb. + :rtype: float + """ + voltage = _adc_to_voltage(readout) + gain = 1.0 + (100.0 / 3.3) + return (voltage / gain) * (1000.0 / 0.1) \ No newline at end of file diff --git a/Sensor/create_sensors.py b/Sensor/create_sensors.py index 1aa3fe3..1ac8d25 100644 --- a/Sensor/create_sensors.py +++ b/Sensor/create_sensors.py @@ -1,7 +1,7 @@ # SPDX-License-Identifier: BSD-3-Clause # Copyright (c) 2025 Sun Devil Rocketry -from .conv_functions import imu_accel, imu_gyro, baro_press +from .conv_functions import imu_accel, imu_gyro, baro_press, pt_pressure, pt_pressure_5V, tc_temp, loadcell_force from .sensor import Sensor from typing import List @@ -84,6 +84,50 @@ def rev2_dashboard_dump_sensors() -> List[Sensor]: sensors: List[Sensor] = [] offset = 0 + for poll_code, short_name, name, size, data_type, unit, conv_func in sensor_tuples: + sensors.append(Sensor(short_name, name, size, data_type, unit, conv_func, poll_code, offset)) + offset += size + + return sensors + +def engine_controller_rev5_sensors() -> List[Sensor]: + """Create the sensor list for the Liquid Engine Controller (L0002 Rev 5.0). + + Returns 10 sensors in poll-code order (``0x00``–``0x09``):: + + pt0 LOX Pressure psi + pt1 LOX Flow Upstream psi + pt2 LOX Flow Downstream psi + pt3 PT3 psi + pt4 Engine Pressure psi + pt5 Fuel Flow Downstream psi + pt6 Fuel Flow Upstream psi + pt7 Fuel Pressure psi + tc Thermocouple °C + lc Load Cell lb + + All sensors carry raw ``int`` values (4 bytes each) with their + respective conversion functions pre-attached. + + :returns: Ordered list of :class:`~SDECv2.Sensor.sensor.Sensor` objects. + :rtype: List[Sensor] + """ + sensor_tuples = [ + (b"\x00", "pt0", "LOX Pressure", 4, int, "psi", pt_pressure), + (b"\x01", "pt1", "LOX Flow Upstream", 4, int, "psi", pt_pressure), + (b"\x02", "pt2", "LOX Flow Downstream", 4, int, "psi", pt_pressure), + (b"\x03", "pt3", "PT3", 4, int, "psi", pt_pressure), + (b"\x04", "pt4", "Engine Pressure", 4, int, "psi", pt_pressure), + (b"\x05", "pt5", "Fuel Flow Downstream", 4, int, "psi", pt_pressure), + (b"\x06", "pt6", "Fuel Flow Upstream", 4, int, "psi", pt_pressure), + (b"\x07", "pt7", "Fuel Pressure", 4, int, "psi", pt_pressure), + (b"\x08", "tc", "Thermocouple", 4, int, "C", tc_temp), + (b"\x09", "lc", "Load Cell", 4, int, "lb", loadcell_force), + ] + + sensors: List[Sensor] = [] + offset = 0 + for poll_code, short_name, name, size, data_type, unit, conv_func in sensor_tuples: sensors.append(Sensor(short_name, name, size, data_type, unit, conv_func, poll_code, offset)) offset += size diff --git a/Testing/EngineController/Sensor/test_sentry_dump.py b/Testing/EngineController/Sensor/test_sentry_dump.py new file mode 100644 index 0000000..2125310 --- /dev/null +++ b/Testing/EngineController/Sensor/test_sentry_dump.py @@ -0,0 +1,30 @@ +from Sensor import SensorSentry +from Sensor import create_sensors +from SerialController import SerialObj + +def test_sensor(): + # Use premade functions to create the Engine Sensors + engine_controller_rev5_sensors = create_sensors.engine_controller_rev5_sensors() + + # Extract the Sensor objects for the Acc{x,y,z} and gyro{x,y,z}conv sensors + sensor_sentry = SensorSentry(sensors=engine_controller_rev5_sensors) + + # Create the serial connection + serial_connection = SerialObj() + serial_connection.init_comport("COM6", 921600, 5) + serial_connection.open_comport() + + # Get the sensor dump from the sentry + print("Sentry Dump:") + sensor_dump = sensor_sentry.dump(serial_connection) + + for sensor, readout in sensor_dump.items(): + if readout: + print(f"{sensor.name}: {readout:.2f} {sensor.unit}") + else: + print(f"{sensor.name}: 0.0 {sensor.unit}") + + serial_connection.close_comport() + +if __name__ == "__main__": + test_sensor() \ No newline at end of file diff --git a/Testing/EngineController/Sensor/test_sentry_poll.py b/Testing/EngineController/Sensor/test_sentry_poll.py new file mode 100644 index 0000000..ffcb40a --- /dev/null +++ b/Testing/EngineController/Sensor/test_sentry_poll.py @@ -0,0 +1,45 @@ +import time + +from BaseController import Firmware, BaseController +from BaseController import create_controllers +from Sensor import Sensor, SensorSentry +from Sensor import conv_functions, create_sensors +from SerialController import SerialSentry, SerialObj, Comport + +def test_sensor(): + # Use premade functions to create the Engine Sensors + engine_controller_rev5_sensors = create_sensors.engine_controller_rev5_sensors() + + # Extract the Sensor objects for the PT0, PT1, PT2 + sensor_sentry = SensorSentry() + + for sensor in engine_controller_rev5_sensors: + if sensor.poll_code in {b"\x00", b"\x01", b"\x02"}: + sensor_sentry.add_sensor(sensor) + + # Create the serial connection + serial_connection = SerialObj() + serial_connection.init_comport("COM6", 921600, 5) + serial_connection.open_comport() + + # Get the sensor poll from the sentry + print("Sentry Poll (2 seconds):") + for sensor_poll in sensor_sentry.poll(serial_connection, timeout=2): + for sensor, readout in sensor_poll.items(): + if readout: + print(f"{sensor.name}: {readout:.2f} {sensor.unit}") + else: + print(f"{sensor.name}: 0.0 {sensor.unit}") + + print("Sentry Poll (count 10):") + for sensor_poll in sensor_sentry.poll(serial_connection, count=10): + for sensor, readout in sensor_poll.items(): + if readout: + print(f"{sensor.name}: {readout:.2f} {sensor.unit}") + else: + print(f"{sensor.name}: 0.0 {sensor.unit}") + + serial_connection.close_comport() + +if __name__ == "__main__": + test_sensor() \ No newline at end of file diff --git a/Testing/EngineController/test_engine_controller.py b/Testing/EngineController/test_engine_controller.py new file mode 100644 index 0000000..986b18c --- /dev/null +++ b/Testing/EngineController/test_engine_controller.py @@ -0,0 +1,50 @@ +from BaseController import create_controllers +from EngineController import ( + EngineState, ENGINE_STATE_NAMES, + CONTROLLER_ID, CONTROLLER_NAME, + get_state, telemetry_request, +) +from SerialController import SerialObj + +def test_engine_controller(): + ec_rev5 = create_controllers.engine_controller_rev5_controller() + + print(f"Controller: {ec_rev5.name}") + print(f"Controller ID: {ec_rev5.id}") + print(f"Sensor frame size: {ec_rev5.sensor_frame_size} bytes") + print() + print(ec_rev5.get_formatted_poll_codes()) + +def test_get_state(): + serial = SerialObj() + serial.init_comport("COM3", 921600, 5) + serial.open_comport() + + state = get_state(serial) + if state is not None: + print(f"Engine state: {ENGINE_STATE_NAMES[state]}") + else: + print("Failed to get engine state") + + serial.close_comport() + +def test_telemetry_request(): + serial = SerialObj() + serial.init_comport("COM3", 921600, 5) + serial.open_comport() + + data = telemetry_request(serial) + if data is not None: + print("Sensor readings:") + for name, value in data.sensor_readings.items(): + print(f" {name}: {value:.3f}") + print("Valve states:") + for name, state in data.valve_states.items(): + print(f" {name}: {state}") + else: + print("Telemetry request failed") + + serial.close_comport() + +if __name__ == "__main__": + test_engine_controller() diff --git a/Testing/EngineController/test_flash_extract.py b/Testing/EngineController/test_flash_extract.py new file mode 100644 index 0000000..9cee21d --- /dev/null +++ b/Testing/EngineController/test_flash_extract.py @@ -0,0 +1,21 @@ +from EngineController import flash_extract, EngineControllerFlashFrame +from SerialController import SerialObj + +def test_flash_extract(): + serial = SerialObj() + serial.init_comport("COM3", 921600, 30) + serial.open_comport() + + frames = flash_extract(serial, store_data=True) + + print(f"Total frames: {len(frames)}") + if frames: + first = frames[0] + print(f"First frame — timestamp: {first.timestamp_ms} ms") + for name, value in first.sensor_readings.items(): + print(f" {name}: {value:.3f}") + + serial.close_comport() + +if __name__ == "__main__": + test_flash_extract() diff --git a/__init__.py b/__init__.py index 3197e69..1902bb7 100644 --- a/__init__.py +++ b/__init__.py @@ -5,4 +5,11 @@ from .BaseController import BaseController, BaseSensor, Controller, Firmware, create_controllers from .Parser import Parser, PresetConfig, PresetData, Feature, FeatureBitmask, DataBitmask, Data, create_configs from .Sensor import Sensor, SensorSentry, create_sensors, util, conv_functions -from .SerialController import SerialObj, SerialSentry, Comport \ No newline at end of file +from .SerialController import SerialObj, SerialSentry, Comport +from .EngineController import ( + EngineState, ENGINE_STATE_NAMES, Valve, VALVES, parse_valve_byte, + TelemetryData, EngineControllerFlashFrame, + hotfire_abort, preflight_purge, fill_chill, standby, hotfire, + stop_hotfire, stop_purge, lox_purge, manual_mode, + get_state, telemetry_request, flash_extract, +) \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 35a13cd..a1f161d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,6 +5,6 @@ requires-python = ">=3.12.0" version = "2.0.0" [tool.setuptools] -packages = ["SDECv2.BaseController", "SDECv2.Parser", "SDECv2.Sensor", "SDECv2.SerialController"] +packages = ["SDECv2.BaseController", "SDECv2.Parser", "SDECv2.Sensor", "SDECv2.SerialController", "SDECv2.EngineController"] [tool.setuptools.package-dir] "" = ".." \ No newline at end of file