serial_decoder.py

#!/usr/bin/env python3
"""
serial_decoder.py

AERO60492 Coursework 1: Serial Message Decoding
A script to decode a binary data file according to a specified protocol.

Author: 174347826+bt-nav@users.noreply.github.com
Date: February 2026
"""

import csv
import sys
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Iterable, Optional

# Define protocol constants.
STX = 0x7E  # ~
PTX = 0x50  # P
TTX = 0x54  # T
VALUE_RANGE = range(256)  # 0-255
FRAME_LENGTH = 26


@dataclass
class DataFrame:
    """Data class used to represent a single frame of extracted data.

    Attributes:
        stx (str): marks the start of the frame with ~~.
        system_id (int): the device that sent the data frame.
        destination_id (int): the device the data frame is destined for.
        component_id (int): the subcomponent that sent the data frame.
        sequence (int): the sequence number of the data frame.
        payload_type (int): indicates the type of data in the payload.
        ptx (str): marks the start of the payload with P.
        rpm (int): extracted RPM measurement in revolutions per minute.
        voltage (int): extracted voltage in millivolts.
        current (int): extracted current in milliamps.
        mosfet_temp (float): extracted MOSFET temperature in Celsius.
        cap_temp (float): extracted capacitor temperature in Celsius.
        ttx (str): marks the start of timing info with T.
        timestamp_us (int): extracted Unix time code in microseconds.
        checksum (int): checksum value.
        checksum_valid (bool): whether the checksum matches a computed value.
    """

    stx: str
    system_id: int
    destination_id: int
    component_id: int
    sequence: int
    payload_type: int
    ptx: str
    rpm: int
    voltage: int
    current: int
    mosfet_temp: float
    cap_temp: float
    ttx: str
    timestamp_us: int
    checksum: int
    checksum_valid: bool


class Decoder:
    """Class containing the methods used to decode a binary message."""

    def __init__(self) -> None:
        pass

    # Public methods

    def decode_message(
        self, stream, output_path: str
    ) -> tuple[list[DataFrame], list[int], list[int], list[int]]:
        """Decode frames from a binary stream with a buffer.

        Returns the decoded frames, and indices of potentially corrupt frames."
        """

        # Variables to keep track of frames.
        buffer: bytes = b""
        frame_index: int = 0
        decoded_frames: list[DataFrame] = []
        bad_checksum_frames: list[int] = []
        bad_structure_frames: list[int] = []
        bad_sequence_frames: list[int] = []
        prev_seq: Optional[int] = None

        # Read the binary stream in 1024 byte chunks to save memory.
        while True:
            chunk = stream.read(1024)
            if not chunk and len(buffer) < FRAME_LENGTH:
                break
            buffer += chunk

            # Process all data frames in the buffer.
            i = 0
            while i <= len(buffer) - FRAME_LENGTH:
                if buffer[i] == STX and buffer[i + 1] == STX:
                    candidate = buffer[i : i + FRAME_LENGTH]
                    frame_index += 1

                    # Decode the candidate frame.
                    decoded = self._decode_data_frame(frame=candidate)
                    if decoded is not None:
                        decoded_frames.append(decoded)

                        # Checksum, structure, and sequence checks.
                        structure_ok = self._check_structure(frame=candidate)
                        checksum_ok = decoded.checksum_valid

                        if not checksum_ok:
                            bad_checksum_frames.append(frame_index)
                        if not structure_ok:
                            bad_structure_frames.append(frame_index)

                        if prev_seq is not None:
                            bad_sequence = self._check_sequence(
                                previous_seq=prev_seq,
                                current_seq=decoded.sequence,
                                current_frame=frame_index,
                            )
                            if bad_sequence is not None:
                                bad_sequence_frames.append(bad_sequence)
                        prev_seq = decoded.sequence

                        # Append the decoded frame to the output CSV
                        self._output_frame([decoded], output_path)

                        i += FRAME_LENGTH
                        continue
                i += 1

            # Keep any remaining bytes in the buffer for the next chunk.
            buffer = buffer[i:]
            if not chunk:
                break

        return (
            decoded_frames,
            bad_checksum_frames,
            bad_structure_frames,
            bad_sequence_frames,
        )

    # Private methods

    def _output_frame(
        self,
        frames: Iterable[DataFrame],
        output_path: str,
    ) -> None:
        """Write decoded frames to a CSV file."""

        with open(output_path, "a", newline="", encoding="utf-8") as csv_file:
            writer = csv.writer(csv_file)
            for frame in frames:
                writer.writerow(
                    [
                        frame.stx,
                        frame.system_id,
                        frame.destination_id,
                        frame.component_id,
                        frame.sequence,
                        frame.payload_type,
                        frame.ptx,
                        frame.rpm,
                        frame.voltage,
                        frame.current,
                        frame.mosfet_temp,
                        frame.cap_temp,
                        frame.ttx,
                        frame.timestamp_us,
                        frame.checksum,
                    ]
                )

    def _decode_data_frame(self, frame: bytes) -> Optional[DataFrame]:
        """Decodes a data frame and returns a populated DataFrame object."""

        # Decoding the header.
        frame_start = "".join(
            self._decode_marker(b) for b in (frame[0], frame[1])
        )
        sys_id, dest_id, comp_id, seq, payload_type = self._decode_header(
            frame
        )

        # Decoding the payload.
        payload_start = self._decode_marker(frame[7])
        rpm, voltage, current, mosfet_temp, cap_temp = self._decode_payload(
            frame
        )

        # Decoding the timing info.
        timing_start = self._decode_marker(frame[16])
        timestamp = self._decode_timestamp(frame)

        # Decoding the checksum and validating against the computed checksum.
        checksum = frame[25]
        computed_compute_checksum = self._compute_checksum(frame[:25])
        checksum_valid = checksum == computed_compute_checksum

        # Return the decoded data frame.
        return DataFrame(
            stx=frame_start,
            system_id=sys_id,
            destination_id=dest_id,
            component_id=comp_id,
            sequence=seq,
            payload_type=payload_type,
            ptx=payload_start,
            rpm=rpm,
            voltage=voltage,
            current=current,
            mosfet_temp=mosfet_temp,
            cap_temp=cap_temp,
            ttx=timing_start,
            timestamp_us=timestamp,
            checksum=checksum,
            checksum_valid=checksum_valid,
        )

    # Helper methods

    @staticmethod
    def _check_structure(frame: bytes) -> bool:
        """Verify bytes in a frame follow the message protocol.

        Checks frame length, and that each byte has an expected value.
        """
        if len(frame) != FRAME_LENGTH:
            return False
        if not (frame[0] == STX and frame[1] == STX):
            return False
        for b in frame[2:7]:
            if b not in VALUE_RANGE:
                return False
        if frame[7] != PTX:
            return False
        for b in frame[8:16]:
            if b not in VALUE_RANGE:
                return False
        if frame[16] != TTX:
            return False
        for b in frame[17:26]:
            if b not in VALUE_RANGE:
                return False
        return True

    @staticmethod
    def _check_sequence(
        previous_seq: int, current_seq: int, current_frame: int
    ) -> Optional[int]:
        """Check if the current sequence number is expected.

        Returns out-of-sequence frame indices.
        """
        expected = (previous_seq + 1) % 256
        if current_seq == expected:
            return None
        return current_frame

    @staticmethod
    def _compute_checksum(frame_excluding_checksum: bytes) -> int:
        """Computes checksum for a frame using the given algorithm."""
        total = sum(frame_excluding_checksum)
        return 255 - (total % 256)

    @staticmethod
    def _temp_lookup(raw_temp: int) -> float:
        """Convert raw temperature byte to Celsius.

        Temperatures start at 30.0°C for 0xA0, with each byte adding 0.1°C,
        ending at 36.3°C for 0xDF. Values outside this range return NaN.
        """
        if 0xA0 <= raw_temp <= 0xDF:
            return 30.0 + (raw_temp - 0xA0) * 0.1
        return float("nan")

    @staticmethod
    def _decode_marker(value: int) -> str:
        """Turn a byte value into an ASCII character."""
        if 32 <= value <= 126:
            return chr(value)
        return str(value)

    @staticmethod
    def _decode_header(frame: bytes) -> tuple[int, int, int, int, int]:
        """Extract header bytes from a frame."""
        return frame[2], frame[3], frame[4], frame[5], frame[6]

    @staticmethod
    def _decode_payload(frame: bytes) -> tuple[int, int, int, float, float]:
        """Extract payload bytes from a frame and convert."""

        # Use correct endianess and signedness to convert bytes to integers
        rpm = int.from_bytes(frame[8:10], byteorder="big", signed=False)
        voltage = int.from_bytes(frame[10:12], byteorder="big", signed=False)
        current = int.from_bytes(frame[12:14], byteorder="little", signed=True)

        # Use lookup table to convert temperature bytes to Celsius
        mosfet_temp = Decoder._temp_lookup(frame[14])
        capacitor_temp = Decoder._temp_lookup(frame[15])

        return rpm, voltage, current, mosfet_temp, capacitor_temp

    @staticmethod
    def _decode_timestamp(frame: bytes) -> int:
        """Extract timestamp bytes from a frame and convert to an integer."""
        return int.from_bytes(frame[17:25], byteorder="big", signed=False)


def main(
    input_path: str = "encoded_data.bin",
    output_path: str = "decoded_data.csv",
) -> None:
    decoder = Decoder()
    """Main function to decode the binary file and print answers."""

    # Clear the output file and open the input file.
    with open(output_path, "w", newline="", encoding="utf-8"):
        pass
    with open(input_path, "rb") as input_file:
        (
            frames,
            bad_checksum_frames,
            bad_structure_frames,
            bad_sequence_frames,
        ) = decoder.decode_message(input_file, output_path)

    # Union to find unique potentially corrupt frames
    potentially_corrupt_frames = (
        set(bad_checksum_frames)
        | set(bad_structure_frames)
        | set(bad_sequence_frames)
    )

    # Convert first Unix timestamp (secs since January 1, 1970, UTC) to a date.
    timestamp_secs = float(frames[0].timestamp_us) / 10**6
    calendar_date = (
        datetime.fromtimestamp(timestamp_secs, tz=timezone.utc)
        .date()
        .isoformat()
    )

    # Print info to help answer the questions.
    print(f"Total frames: {len(frames)}")

    print(f"\nFailed checksums: {len(bad_checksum_frames)}")
    if bad_checksum_frames:
        print(f"Rows: {sorted(bad_checksum_frames)}")

    print(f"\nFrames with bad structure: {len(bad_structure_frames)}")
    if bad_structure_frames:
        print(f"Rows: {sorted(bad_structure_frames)}")

    print(f"\nOut-of-sequence frames: {len(bad_sequence_frames)}")
    if bad_sequence_frames:
        print(f"Rows: {sorted(bad_sequence_frames)}")

    print(f"\nPotentially corrupt frames: {len(potentially_corrupt_frames)}")
    if potentially_corrupt_frames:
        print(f"Rows: {sorted(potentially_corrupt_frames)}")

    print(f"\nCalendar date: {calendar_date}")


if __name__ == "__main__":
    # Accept command-line arguments for input and output paths.
    if len(sys.argv) == 3:
        main(sys.argv[1], sys.argv[2])
    else:
        main()