improved logging

This commit is contained in:
Ayzen
2026-06-06 00:52:52 +03:00
parent af6005d68f
commit aea49f6128
65 changed files with 1206 additions and 240 deletions
@@ -1,11 +1,10 @@
"""
Communication protocol for laser control module.
"""Communication protocol for the laser control module.
Encodes commands to bytes and decodes device responses.
Faithful re-implementation of the logic in device_commands.py,
refactored into a clean, testable class-based API.
Encodes commands to wire bytes, decodes device responses, and manages the
serial port connection to the laser control board.
"""
import logging
import struct
from typing import Optional
from enum import IntEnum
@@ -38,6 +37,8 @@ from .exceptions import (
ProtocolError,
)
logger = logging.getLogger(__name__)
# Re-export enums so tests can import from protocol module
class CommandCode(IntEnum):
@@ -77,11 +78,11 @@ def _int_to_hex4(value: int) -> str:
return f"{value:04x}"
def _flipfour(s: str) -> str:
"""Swap two byte-pairs: 'aabb' 'bbaa' (little-endian word)."""
if len(s) != 4:
raise ValueError(f"Expected 4-char hex string, got '{s}'")
return s[2:4] + s[0:2]
def _flipfour(hex_word: str) -> str:
"""Swap the two byte-pairs of a 4-char hex word: 'aabb' -> 'bbaa' (little-endian)."""
if len(hex_word) != 4:
raise ValueError(f"Expected 4-char hex string, got '{hex_word}'")
return hex_word[2:4] + hex_word[0:2]
def _xor_crc(words: list) -> str:
@@ -183,7 +184,7 @@ class Protocol:
# ---- Connection management
def connect(self) -> None:
"""Open the serial port. Auto-detects if port is None."""
"""Open the serial port. Auto-detects the device path when port is None."""
port = self._port_name or self._detect_port()
try:
self._serial = serial.Serial(
@@ -192,13 +193,16 @@ class Protocol:
timeout=SERIAL_TIMEOUT_SEC,
)
except Exception as exc:
logger.error("Cannot open laser serial port '%s': %s", port, exc)
raise CommunicationError(
f"Cannot connect to port '{port}': {exc}"
) from exc
logger.debug("Laser serial port opened: %s @ %d baud", port, BAUDRATE)
def disconnect(self) -> None:
"""Close the serial port if open."""
if self._serial and self._serial.is_open:
logger.debug("Closing laser serial port")
self._serial.close()
@property
@@ -241,13 +245,14 @@ class Protocol:
@staticmethod
def calculate_crc(data: bytes) -> int:
"""
XOR CRC over all 16-bit words except the last two bytes (CRC field).
Mirrors the original CalculateCRC logic.
"""Return the XOR CRC over all 16-bit words except word 0 and the CRC field.
The command-code word (word 0) is excluded, matching the firmware's CRC
expectation.
"""
hex_str = data.hex()
words = [hex_str[i:i+4] for i in range(0, len(hex_str), 4)]
# Skip word 0 (command code) per original firmware expectation
# Word 0 (command code) is excluded from the CRC.
crc_words = words[1:]
result = int(crc_words[0], 16)
for w in crc_words[1:]:
@@ -342,9 +347,8 @@ class Protocol:
case TaskType.CHANGE_CURRENT_LD2:
data += _flipfour(_int_to_hex4(current_ma_to_n(min_value))) # Word 3
data += _flipfour(_int_to_hex4(current_ma_to_n(max_value))) # Word 4
# Word 5: current step encoded like LD1 and like min/max (current_ma_to_n),
# NOT int(step*100) — the latter was a copy/paste from temperature scaling
# and produced a different wire value than LD1 for the same physical step.
# Word 5: current step uses the same current_ma_to_n scaling as
# min/max (and as LD1) so equal physical steps map to equal wire values.
data += _flipfour(_int_to_hex4(current_ma_to_n(step))) # Word 5
data += _flipfour(_int_to_hex4(int(time_step * 100))) # Word 6: Delta_Time_µs × 100
data += _flipfour(_int_to_hex4(temp_c_to_n(static_temp2))) # Word 7