awe/radar_system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

kamil_adc support added

Browse Source
This commit is contained in:
Ayzen
2026-05-08 21:23:52 +03:00
parent bde86813e5
commit 907dbf29ce
36 changed files with 2161 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python_app/tests/test_laser_control_protocol.py
+250
View File
@@ -0,0 +1,250 @@
"""Laser-control protocol compatibility tests."""
from __future__ import annotations
import unittest
from unittest.mock import patch
from python_app.hardware_full.laser_control.controller import DEVICE_MAIN_MESSAGE_ID, LaserController
from python_app.hardware_full.laser_control.models import VariationType
from python_app.hardware_full.laser_control.protocol import Protocol, TaskType
from python_app.hardware_full.kamil_adc_service import apply_kamil_adc_laser_control
from python_app.models.run_config_model import RunConfigModel
DEVICE_MAIN_MANUAL_HEX = "1111ff37ffa518ab000000000000000a8000000a8000ff003d2acc2c163f"
DEVICE_MAIN_CHANGE_CURRENT_LD1_HEX = (
"7777ff3701003d2acc2c10008813ffa5cc2c18ab0a00000a8000000a8000b600"
)
DEVICE_MAIN_CHANGE_CURRENT_LD2_HEX = (
"7777ff3702003d2acc2c0500881318ab3d2affa50a00000a8000000a80005106"
)
class _FakeProtocol:
def __init__(self) -> None:
self.is_connected = True
self.sent: list[bytes] = []
def send_raw(self, data: bytes) -> None:
self.sent.append(bytes(data))
def receive_raw(self, length: int) -> bytes:
return b"\x00\x00" if length == 2 else b""
class _FakeLaserController:
instances: list["_FakeLaserController"] = []
def __init__(self, **kwargs: object) -> None:
self.kwargs = kwargs
self.calls: list[tuple[str, object]] = []
_FakeLaserController.instances.append(self)
def connect(self) -> bool:
self.calls.append(("connect", None))
return True
def reset(self) -> None:
self.calls.append(("reset", None))
def set_manual_mode(self, **kwargs: object) -> None:
self.calls.append(("set_manual_mode", kwargs))
def start_variation(self, **kwargs: object) -> None:
self.calls.append(("start_variation", kwargs))
def disconnect(self) -> None:
self.calls.append(("disconnect", None))
class LaserControlProtocolCompatibilityTest(unittest.TestCase):
def setUp(self) -> None:
_FakeLaserController.instances.clear()
@staticmethod
def _kamil_config(laser_payload: dict[str, object]) -> RunConfigModel:
return RunConfigModel.from_dict(
{
"radar": {
"model": "kamil_adc",
"driver_mode": "native",
"laser_control": laser_payload,
}
}
)
def test_manual_command_matches_device_main_bytes(self) -> None:
command = Protocol.encode_decode_enable(
temp1=28.0,
temp2=28.9,
current1=33.0,
current2=35.0,
pi_coeff1_p=2560,
pi_coeff1_i=128,
pi_coeff2_p=2560,
pi_coeff2_i=128,
message_id=DEVICE_MAIN_MESSAGE_ID,
)
self.assertEqual(command.hex(), DEVICE_MAIN_MANUAL_HEX)
def test_variation_commands_match_device_main_bytes(self) -> None:
ld1_command = Protocol.encode_task_enable(
task_type=TaskType.CHANGE_CURRENT_LD1,
static_temp1=28.0,
static_temp2=28.9,
static_current1=33.0,
static_current2=35.0,
min_value=33.0,
max_value=35.0,
step=0.05,
time_step=50,
delay_time=10,
message_id=DEVICE_MAIN_MESSAGE_ID,
pi_coeff1_p=2560,
pi_coeff1_i=128,
pi_coeff2_p=2560,
pi_coeff2_i=128,
)
ld2_command = Protocol.encode_task_enable(
task_type=TaskType.CHANGE_CURRENT_LD2,
static_temp1=28.0,
static_temp2=28.9,
static_current1=33.0,
static_current2=35.0,
min_value=33.0,
max_value=35.0,
step=0.05,
time_step=50,
delay_time=10,
message_id=DEVICE_MAIN_MESSAGE_ID,
pi_coeff1_p=2560,
pi_coeff1_i=128,
pi_coeff2_p=2560,
pi_coeff2_i=128,
)
self.assertEqual(ld1_command.hex(), DEVICE_MAIN_CHANGE_CURRENT_LD1_HEX)
self.assertEqual(ld2_command.hex(), DEVICE_MAIN_CHANGE_CURRENT_LD2_HEX)
def test_start_sequence_matches_device_main_order(self) -> None:
fake_protocol = _FakeProtocol()
controller = LaserController(pi_coeff1_p=2560, pi_coeff1_i=128, pi_coeff2_p=2560, pi_coeff2_i=128)
controller._protocol = fake_protocol
with patch("python_app.hardware_full.laser_control.controller.time.sleep", return_value=None):
controller.reset()
controller.set_manual_mode(
temp1=28.0,
temp2=28.9,
current1=33.0,
current2=35.0,
message_id=DEVICE_MAIN_MESSAGE_ID,
)
controller.start_variation(
variation_type=VariationType.CHANGE_CURRENT_LD1,
params={
"static_temp1": 28.0,
"static_temp2": 28.9,
"static_current1": 33.0,
"static_current2": 35.0,
"min_value": 33.0,
"max_value": 35.0,
"step": 0.05,
"time_step": 50,
"delay_time": 10,
},
)
self.assertEqual(
[command.hex() for command in fake_protocol.sent],
[
"2222",
DEVICE_MAIN_MANUAL_HEX,
DEVICE_MAIN_CHANGE_CURRENT_LD1_HEX,
],
)
def test_stop_sequence_restores_device_main_manual_bytes(self) -> None:
fake_protocol = _FakeProtocol()
controller = LaserController(pi_coeff1_p=2560, pi_coeff1_i=128, pi_coeff2_p=2560, pi_coeff2_i=128)
controller._protocol = fake_protocol
with patch("python_app.hardware_full.laser_control.controller.time.sleep", return_value=None):
controller.set_manual_mode(
temp1=28.0,
temp2=28.9,
current1=33.0,
current2=35.0,
message_id=DEVICE_MAIN_MESSAGE_ID,
)
fake_protocol.sent.clear()
controller.stop_task(restore_message_id=DEVICE_MAIN_MESSAGE_ID)
self.assertEqual(
[command.hex() for command in fake_protocol.sent],
[
"2222",
DEVICE_MAIN_MANUAL_HEX,
],
)
def test_apply_radar_variation_sequence_matches_device_main_order(self) -> None:
config = self._kamil_config(
{
"enabled": True,
"port": "/dev/ttyUSB0",
"mode": "variation",
"pi_coeff1_p": 2560,
"pi_coeff1_i": 128,
"pi_coeff2_p": 2560,
"pi_coeff2_i": 128,
"variation": {
"variation_type": "CHANGE_CURRENT_LD1",
"static_temp1": 28.0,
"static_temp2": 28.9,
"static_current1": 33.0,
"static_current2": 35.0,
"min_value": 33.0,
"max_value": 35.0,
"step": 0.05,
"time_step": 50,
"delay_time": 10,
},
}
)
with patch("python_app.hardware_full.laser_control.controller.LaserController", _FakeLaserController):
applied = apply_kamil_adc_laser_control(config)
self.assertTrue(applied)
controller = _FakeLaserController.instances[0]
self.assertEqual(
[name for name, _payload in controller.calls],
["connect", "reset", "set_manual_mode", "start_variation", "disconnect"],
)
manual_payload = controller.calls[2][1]
self.assertEqual(
manual_payload,
{
"temp1": 28.0,
"temp2": 28.9,
"current1": 33.0,
"current2": 35.0,
"message_id": DEVICE_MAIN_MESSAGE_ID,
},
)
def test_apply_radar_skips_disabled_laser_control(self) -> None:
config = self._kamil_config({"enabled": False})
with patch("python_app.hardware_full.laser_control.controller.LaserController", _FakeLaserController):
applied = apply_kamil_adc_laser_control(config)
self.assertFalse(applied)
self.assertEqual(_FakeLaserController.instances, [])
if __name__ == "__main__":
unittest.main()