Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 39 additions & 8 deletions src/openlifu_sdk/ui/simulated_interface.py
Original file line number Diff line number Diff line change
Expand Up @@ -578,22 +578,32 @@ def set_preset(self, preset_index: int, sequence_duration_index: int,
:meth:`get_preset`-based and happens separately in
:func:`verify_preset` on the host.

Does not push acoustic parameters -- that is the job of
:meth:`set_solution`, which runs against a solution that
was itself derived from the loaded preset.
Applies the loaded machine_config's trigger parameters
to :attr:`_sequence` and :attr:`_pulse` so a subsequent
:meth:`SimulatedLIFUInterface.start_sonication` runs the
correct sequence. Reads these fields from the
machine_config: ``pulse_interval_ms``, ``pulse_count``,
``pulse_train_interval_s``,
``pulse_train_count_selections[sequence_duration_index]``,
and ``pulse_length_us``. Mirrors the real firmware model
where ``set_preset`` loads the firmware-baked machine_config
and no separate ``set_trigger`` call is required before
starting the sonication (per preset_pseudo_code.md).

Raises
------
IndexError
From :meth:`get_machine_config` if ``preset_index`` is
out of range.
ValueError
If ``expected_crc`` does not match the flash
machine_config's CRC (this is the CRC-mismatch fault
path the operator UI must annunciate per SR-003) or if
``sequence_duration_index`` is negative.
* If ``expected_crc`` does not match the flash
machine_config's CRC (SR-003 CRC-mismatch fault
path).
* If ``sequence_duration_index`` is negative or
beyond the loaded preset's
``pulse_train_count_selections`` length.
"""
_machine_config, flash_mc_crc = self.get_machine_config(preset_index)
machine_config, flash_mc_crc = self.get_machine_config(preset_index)
if expected_crc != flash_mc_crc:
raise ValueError(
f"set_preset CRC mismatch: caller sent "
Expand All @@ -606,6 +616,27 @@ def set_preset(self, preset_index: int, sequence_duration_index: int,
f"sequence_duration_index must be >= 0, got "
f"{sequence_duration_index}"
)
selections = machine_config.get("pulse_train_count_selections", [])
if sequence_duration_index >= len(selections):
raise ValueError(
f"sequence_duration_index {sequence_duration_index} out "
f"of range for preset's pulse_train_count_selections "
f"(length {len(selections)})"
)

# Apply the loaded preset's trigger + pulse configuration
# so a subsequent start_sonication runs the correct
# sequence. Convert ms/us to seconds to match the units
# the engine uses.
self._sequence = {
"pulse_interval": float(machine_config.get("pulse_interval_ms", 0.0)) / 1000.0,
"pulse_count": int(machine_config.get("pulse_count", 1)),
"pulse_train_interval": float(machine_config.get("pulse_train_interval_s", 0.0)),
"pulse_train_count": int(selections[sequence_duration_index]),
}
pulse_length_us = float(machine_config.get("pulse_length_us", 100.0))
self._pulse["duration"] = pulse_length_us / 1_000_000.0

self._loaded_preset_index = preset_index
self._loaded_duration_index = sequence_duration_index

Expand Down
19 changes: 17 additions & 2 deletions unit-test/test_simulated_preset_flash.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,23 @@

def _mc(preset_id: str, settings_crc: int, **extra) -> dict:
"""Minimal machine_config for tests -- ``id`` + ``settings_crc``
are what ``SimulatedTxDevice.get_preset`` reads."""
return {"id": preset_id, "settings_crc": settings_crc, **extra}
are what ``SimulatedTxDevice.get_preset`` reads;
``pulse_train_count_selections`` (+ supporting trigger-config
fields) are what ``set_preset`` reads to configure the
sim's ``_sequence`` so a subsequent ``start_sonication`` runs
the loaded preset. Overridable via ``**extra`` for tests that
care about specific values."""
base = {
"id": preset_id,
"settings_crc": settings_crc,
"pulse_interval_ms": 10.0,
"pulse_count": 1,
"pulse_train_interval_s": 0.0,
"pulse_train_count_selections": [1, 2, 3, 4, 5],
"pulse_length_us": 100.0,
}
base.update(extra)
return base


# ------------------------------------------------------------------
Expand Down
211 changes: 211 additions & 0 deletions unit-test/test_simulated_set_preset_applies_trigger.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,211 @@
# SPDX-License-Identifier: Apache-2.0
"""Tests for the trigger-config application side of
:meth:`SimulatedTxDevice.set_preset`.

Mirrors the real firmware model: ``set_preset`` loads the
flash-baked ``machine_config`` and applies its trigger +
pulse parameters so a subsequent
:meth:`SimulatedLIFUInterface.start_sonication` runs the
correct sequence without a separate ``set_trigger`` call.

Added 2026-10-01 (companion to the OI #78 smoke test
revealing that ``set_preset`` didn't update ``_sequence``,
so an immediate ``start_sonication`` ran the default 1-train
sequence instead of the operator's selected duration).
"""

from __future__ import annotations

import zlib

import pytest

pytest.importorskip("PyQt6")

from openlifu_sdk.ui.simulated_interface import SimulatedTxDevice # noqa: E402


def _machine_config_crc(mc: dict) -> int:
"""Replicates openlifu_operator_interface.preset_schema.machine_config_crc
without introducing a cross-repo import."""
import json
canonical = json.dumps(mc, sort_keys=True, separators=(",", ":"),
ensure_ascii=True).encode("utf-8")
return zlib.crc32(canonical)


def _build_mc(
*,
label: str = "example",
pulse_interval_ms: float = 10.0,
pulse_count: int = 100,
pulse_train_interval_s: float = 0.5,
pulse_train_count_selections=(5, 10, 20),
pulse_length_us: float = 50.0,
) -> dict:
"""Build a minimal machine_config in the same shape the OI
builds in preset_schema.to_machine_config."""
return {
"id": label,
"voltage": 12.0,
"sensitivity": 1.0,
"voltage_range": [10.0, 15.0],
"start_C": 25.0,
"shutoff_C": 55.0,
"pulse_length_us": pulse_length_us,
"pulse_interval_ms": pulse_interval_ms,
"pulse_count": pulse_count,
"pulse_train_interval_s": pulse_train_interval_s,
"pulse_train_count_selections": list(pulse_train_count_selections),
"settings_crc": 0,
}


def _tx_with_flash(mc_list):
"""Build a SimulatedTxDevice with a flash seeded by the given
machine_configs (CRC computed on the fly)."""
flash = [(mc, _machine_config_crc(mc)) for mc in mc_list]
return SimulatedTxDevice(preset_flash=flash)


# ------------------------------------------------------------------
# Trigger config application
# ------------------------------------------------------------------


def test_set_preset_updates_sequence_from_machine_config():
mc = _build_mc(
pulse_interval_ms=25.0,
pulse_count=50,
pulse_train_interval_s=1.5,
pulse_train_count_selections=[10, 20, 40],
)
tx = _tx_with_flash([mc])
expected_crc = _machine_config_crc(mc)

# Pick the middle duration (index 1 -> 20 trains).
tx.set_preset(0, sequence_duration_index=1, expected_crc=expected_crc)

assert tx._sequence["pulse_interval"] == pytest.approx(0.025)
assert tx._sequence["pulse_count"] == 50
assert tx._sequence["pulse_train_interval"] == pytest.approx(1.5)
assert tx._sequence["pulse_train_count"] == 20


def test_set_preset_updates_pulse_duration_from_machine_config():
mc = _build_mc(pulse_length_us=75.0)
tx = _tx_with_flash([mc])
tx.set_preset(0, sequence_duration_index=0,
expected_crc=_machine_config_crc(mc))
# pulse_length_us=75 -> _pulse["duration"] = 75e-6 s
assert tx._pulse["duration"] == pytest.approx(75e-6)


def test_set_preset_reflects_through_get_trigger():
mc = _build_mc(
pulse_interval_ms=5.0,
pulse_count=10,
pulse_train_interval_s=0.1,
pulse_train_count_selections=[2, 4, 8],
)
tx = _tx_with_flash([mc])
tx.set_preset(0, sequence_duration_index=2,
expected_crc=_machine_config_crc(mc))

trig = tx.get_trigger()
assert trig["pulse_interval"] == pytest.approx(0.005)
assert trig["pulse_count"] == 10
assert trig["pulse_train_interval"] == pytest.approx(0.1)
assert trig["pulse_train_count"] == 8
assert trig["train_count"] == 0
assert trig["trigger_status"] == "STOPPED"


# ------------------------------------------------------------------
# Error paths
# ------------------------------------------------------------------


def test_set_preset_rejects_duration_index_beyond_selections():
mc = _build_mc(pulse_train_count_selections=[5, 10])
tx = _tx_with_flash([mc])
with pytest.raises(ValueError, match="out of range"):
tx.set_preset(0, sequence_duration_index=5,
expected_crc=_machine_config_crc(mc))


def test_set_preset_rejects_negative_duration_index():
mc = _build_mc()
tx = _tx_with_flash([mc])
with pytest.raises(ValueError, match=">= 0"):
tx.set_preset(0, sequence_duration_index=-1,
expected_crc=_machine_config_crc(mc))


def test_set_preset_crc_mismatch_leaves_sequence_unchanged():
"""SR-003 fault path: a CRC mismatch rejects the call and
should not leak partial state into the sequence."""
mc = _build_mc(
pulse_interval_ms=10.0,
pulse_count=5,
pulse_train_interval_s=0.2,
pulse_train_count_selections=[3],
)
tx = _tx_with_flash([mc])
sequence_before = dict(tx._sequence)
pulse_before = dict(tx._pulse)
with pytest.raises(ValueError, match="CRC mismatch"):
tx.set_preset(0, sequence_duration_index=0, expected_crc=0xDEADBEEF)
assert tx._sequence == sequence_before
assert tx._pulse == pulse_before


# ------------------------------------------------------------------
# Integration: start_sonication runs the loaded preset's sequence
# ------------------------------------------------------------------


def test_start_sonication_runs_preset_selected_sequence(qt_app):
"""End-to-end verification: load preset -> start_sonication ->
engine ticks run the selected pulse_train_count, not the
1-train default."""
import time
from openlifu_sdk.ui.simulated_interface import SimulatedLIFUInterface

mc = _build_mc(
pulse_interval_ms=1.0,
pulse_count=1,
pulse_train_interval_s=0.02,
pulse_train_count_selections=[5],
)
flash = [(mc, _machine_config_crc(mc))]
iface = SimulatedLIFUInterface(preset_flash=flash)
try:
iface.txdevice.set_preset(0, sequence_duration_index=0,
expected_crc=_machine_config_crc(mc))
iface.start_sonication(turn_hv_on=False, wait_for_settle=False)

deadline = time.monotonic() + 1.0
while time.monotonic() < deadline:
qt_app.processEvents()
if iface.txdevice.get_trigger()["trigger_status"] == "STOPPED":
break
time.sleep(0.02)

final = iface.txdevice.get_trigger()
assert final["train_count"] == 5
assert final["pulse_train_count"] == 5
assert final["trigger_status"] == "STOPPED"
finally:
iface.close()


# ------------------------------------------------------------------
# Fixtures
# ------------------------------------------------------------------


@pytest.fixture(scope="module")
def qt_app():
from PyQt6.QtCore import QCoreApplication
return QCoreApplication.instance() or QCoreApplication([])