把 Studio 主机协议改成单向端口,并让「从视频重算」走分析入口

控制器经 AcquireGate / ImageSink / ShellNotes 与信号接线,采集 tick 用 acquire_presenter 上屏;三维测试证明分析结果已喂 3D 后,重算并入 on_analysis_requested。
验证:apps/studio 下相关 pytest 与 tools/ui_check.py(634 checks)通过。
This commit is contained in:
ykcheungab
2026-10-05 14:00:24 +08:00
parent 25ccffd814
commit 28e955a817
27 changed files with 1764 additions and 745 deletions
+22 -5
View File
@@ -123,18 +123,35 @@ class _Host:
self.rail_updates = 0
self.markers = 0
self.repaints = 0
self._latest_fps = latest_fps
self._latest_fps_value = latest_fps
def _persist_prefs(self) -> None:
@property
def latest_fps(self):
return self._latest_fps_value
@latest_fps.setter
def latest_fps(self, value) -> None:
self._latest_fps_value = value
# compatibility for tests that still assign ``_latest_fps``
@property
def _latest_fps(self):
return self._latest_fps_value
@_latest_fps.setter
def _latest_fps(self, value) -> None:
self._latest_fps_value = value
def persist_prefs(self) -> None:
self.persisted += 1
def _refresh_status_rail(self) -> None:
def refresh_status_rail(self) -> None:
self.rail_updates += 1
def _drop_render_marker(self) -> None:
def drop_render_marker(self) -> None:
self.markers += 1
def _repaint_live_frame(self) -> None:
def repaint_live_frame(self) -> None:
self.repaints += 1
+154 -42
View File
@@ -165,11 +165,12 @@ class _Host:
def __init__(self, *, view: str = "replay") -> None:
self.prefs: dict = {}
self.view = view
self._running = False
self._preview_open = False
self._session_active = False
self._running_flag = False
self._preview_open_flag = False
self._session_active_flag = False
self.notes: list[str] = []
self.frames: list = []
self.analysis_packets: list = []
self.presets: list[str] = []
self.units: list[str] = []
self.source_modes: list[str] = []
@@ -178,46 +179,105 @@ class _Host:
self.run_updates = 0
self.rail_updates = 0
self.markers = 0
#: What the setup prompt answers for a bare video: a workspace directory, or
#: ``None`` for "the operator declined". The window builds a real dialog; a fake
#: has no dialog, so the answer is the field a test sets.
self.workspace = None
self.workspace_requests: list = []
# reads
def _in_replay_view(self) -> bool:
return self.view == "replay"
# AcquireGate reads (legacy private names kept as writeable fields for tests)
@property
def running(self) -> bool:
return self._running_flag
def _current_source_kind(self) -> str:
@running.setter
def running(self, value: bool) -> None:
self._running_flag = bool(value)
@property
def preview_open(self) -> bool:
return self._preview_open_flag
@preview_open.setter
def preview_open(self, value: bool) -> None:
self._preview_open_flag = bool(value)
@property
def session_active(self) -> bool:
return self._session_active_flag
@session_active.setter
def session_active(self, value: bool) -> None:
self._session_active_flag = bool(value)
# compatibility for tests that still assign ``_running`` / friends
@property
def _running(self) -> bool:
return self._running_flag
@_running.setter
def _running(self, value: bool) -> None:
self._running_flag = bool(value)
@property
def _preview_open(self) -> bool:
return self._preview_open_flag
@_preview_open.setter
def _preview_open(self, value: bool) -> None:
self._preview_open_flag = bool(value)
@property
def _session_active(self) -> bool:
return self._session_active_flag
@_session_active.setter
def _session_active(self, value: bool) -> None:
self._session_active_flag = bool(value)
def current_source_kind(self) -> str:
return "recompute"
# writes
def _apply_view_preset(self, preset: str) -> None:
self.presets.append(preset)
self.view = preset
def bare_video_workspace(self, video):
self.workspace_requests.append(video)
return self.workspace
def _set_display_unit(self, unit: str) -> None:
def analysis_service(self, recording, cancelled=None, *, on_progress=None):
raise AssertionError("the test did not provide an analysis service")
def set_display_unit(self, unit: str) -> None:
self.units.append(unit)
def _set_source_mode(self, mode: str) -> None:
def set_source_mode(self, mode: str) -> None:
self.source_modes.append(mode)
def _show_image_frame(self, frame) -> None:
def show_frame(self, frame) -> None:
self.frames.append(frame)
def _drop_render_marker(self) -> None:
def show_analysis_packet(self, packet) -> None:
self.analysis_packets.append(packet)
def drop_render_marker(self) -> None:
self.markers += 1
def _notify(self, text: str, ms: int = 0) -> None:
def notify(self, text: str, ms: int = 0) -> None:
self.notes.append(str(text))
def _start_run(self) -> None:
def start_acquire(self) -> None:
self.started += 1
def _stop_run(self, *, reason: str = "user_stop", wait: bool = False) -> None:
def stop_acquire(self, *, reason: str = "user_stop", wait: bool = False) -> None:
self.stopped.append(reason)
def _update_run_enabled(self) -> None:
def update_run_enabled(self) -> None:
self.run_updates += 1
def _refresh_status_rail(self) -> None:
def refresh_status_rail(self) -> None:
self.rail_updates += 1
def persist_prefs(self) -> None:
pass
class _Reader:
def __init__(self, *, newtons=None) -> None:
@@ -276,6 +336,10 @@ class _Engine:
def reader(self):
return self._reader
def directory(self):
"""No recording directory: a stub opens a video, not a recorded session."""
return None
def video_path(self):
return self._path
@@ -309,6 +373,23 @@ class _Engine:
return self.accept
def _wire_replay(controller, host):
"""Signals replace private start/stop/view calls on the shell."""
controller.view_requested.connect(
lambda preset: (
host.presets.append(preset),
setattr(host, "view", preset),
controller.set_active_view(preset),
)
)
controller.acquire_requested.connect(host.start_acquire)
controller.acquire_stop_requested.connect(
lambda reason: host.stop_acquire(reason=reason)
)
controller.set_active_view(host.view)
return controller
def _controller(**kwargs):
"""A controller wired to fakes; returns ``(controller, host, engine, parts)``."""
engine = kwargs.pop("engine", None) or _Engine()
@@ -317,13 +398,16 @@ def _controller(**kwargs):
charts = kwargs.pop("charts", None) or _Charts()
panel = kwargs.pop("image_panel", None) or _ImagePanel()
mesh = kwargs.pop("mesh_panel", None) or _Mesh()
controller = ReplayController(
engine=engine,
host=host,
toolbar=toolbar,
charts=charts,
image_panel=panel,
mesh_panel=mesh,
controller = _wire_replay(
ReplayController(
engine=engine,
host=host,
toolbar=toolbar,
charts=charts,
image_panel=panel,
mesh_panel=mesh,
),
host,
)
actions = {
"open_video": _Action(),
@@ -380,20 +464,19 @@ def test_the_playhead_follows_the_wall_clock_and_stops_at_the_end(qt_app, monkey
assert not controller.is_playing(), "the transport kept running past the end"
def test_the_transport_runs_a_recompute_pass_instead_of_playing(qt_app):
"""With the recompute switch on, play starts the *pipeline*, not the cursor."""
def test_the_transport_plays_even_when_the_recompute_switch_is_on(qt_app):
"""Analysis is a separate pass; the rail's play button always moves the cursor."""
engine = _Engine(duration=1.0, has_video=True)
controller, host, engine, parts = _controller(engine=engine)
parts[4]["recompute"].setChecked(True)
controller.on_transport_toggled(True)
assert controller.recompute_running
assert host.started == 1
assert not controller.is_playing(), "the recompute pass also started playback"
assert controller.is_playing(), "play must scrub the recording"
assert host.started == 0, "play must not open an acquisition chain"
controller.on_transport_toggled(False)
assert not controller.recompute_running
assert host.stopped == ["user_stop"]
assert not controller.is_playing()
assert host.stopped == []
def test_the_replay_policy_owns_the_image_only_in_the_replay_view(qt_app):
@@ -402,12 +485,12 @@ def test_the_replay_policy_owns_the_image_only_in_the_replay_view(qt_app):
)
assert controller.owns_image()
host.view = "live"
controller.set_active_view("live")
assert not controller.owns_image(), "a live view does not hand the pane to a replay"
host.view = "replay"
controller.recompute_running = True
assert not controller.owns_image(), "the pass is drawing, not the replay"
controller.set_active_view("replay")
# An analysis pass is drawn by this controller too, so ownership stays with replay.
assert controller.owns_image()
def test_a_bare_video_is_opened_as_a_video_and_refused_outside_the_view(qt_app, tmp_path):
@@ -416,16 +499,45 @@ def test_a_bare_video_is_opened_as_a_video_and_refused_outside_the_view(qt_app,
engine = _Engine()
controller, host, engine, parts = _controller(engine=engine)
host.view = "live"
controller.set_active_view("live")
assert controller.load_video(bare) is False
assert host.notes[-1] == t("toolbar.replay_view_only")
assert engine.opened == [], "a video was opened from the live view"
host.view = "replay"
controller.set_active_view("replay")
assert controller.load_video(bare) is True, "a playable clip was refused"
assert engine.opened == [("video", bare)]
assert host.source_modes[-1] == "video"
assert host.prefs["video_path"] == str(bare)
# Opening a bare video asks how to analyse it — once, after it is on screen. The answer
# here is "declined", and that must leave today's behaviour intact: the video plays,
# and no workspace is claimed as this timeline's input.
assert host.workspace_requests == [bare], "the setup prompt was never asked"
assert controller.recording_directory() is None
assert controller.result_state()[1] == t("playback.analysis_setup_declined")
def test_a_bare_video_uses_the_workspace_the_prompt_answers_with(qt_app, tmp_path):
"""The generated workspace *is* the recording directory for everything downstream."""
bare = tmp_path / "clip.mp4"
bare.write_bytes(b"not really a video")
workspace = tmp_path / "workspace"
workspace.mkdir()
(workspace / "session.json").write_text("{}", encoding="utf-8")
host = _Host()
host.workspace = workspace
controller, host, engine, parts = _controller(host=host)
assert controller.load_video(bare) is True
assert host.workspace_requests == [bare]
assert controller.recording_directory() == workspace, (
"the pass would have read the video's own folder instead of the workspace"
)
# Dropping what is open must drop the claim with it: the workspace is keyed by the
# video, so reopening finds it, but it is not this timeline's input any more.
controller.on_recording_changed(False)
assert controller.recording_directory() is None
def test_the_charts_fall_back_to_raw_when_a_recording_has_no_newtons(qt_app):
@@ -449,7 +561,7 @@ def test_the_replay_entries_exist_only_in_the_replay_view(qt_app):
controller, host, engine, parts = _controller()
actions = parts[4]
host.view = "live"
controller.set_active_view("live")
controller.sync_actions()
assert [
actions[name].enabled for name in ("open_video", "playback", "latest")
@@ -459,7 +571,7 @@ def test_the_replay_entries_exist_only_in_the_replay_view(qt_app):
assert parts[0].recompute_enabled[-1] is False
assert actions["open_video"].tooltips[-1] == t("toolbar.replay_view_only")
host.view = "replay"
controller.set_active_view("replay")
engine._has_video = True
controller.sync_actions()
assert actions["open_video"].enabled is True
@@ -110,7 +110,7 @@ def test_a_shared_profile_id_is_listed_twice_and_can_be_refused(mixed_picker):
assert picker.config_combo.count() == 2
def test_a_single_valid_entry_is_listed_with_its_identity_and_not_blocked(
def test_a_single_valid_entry_is_listed_with_what_it_is_and_not_blocked(
qt_app, tmp_path, monkeypatch
):
def build(root: Path) -> None:
@@ -121,9 +121,41 @@ def test_a_single_valid_entry_is_listed_with_its_identity_and_not_blocked(
assert picker.type_combo.findData("mini") >= 0
picker.set_sensor_type("mini")
label = picker.config_combo.itemText(0)
assert "orisys.mini.lab-r1" in label
assert "mini" in label and "user" in label and "degraded" in label
# What it is, where it came from, what it gives — and not the profile_id the operator
# cannot act on, nor the SDK's verdict word, which says nothing about the force.
assert label == "Mini lab · User · No calibration selected — raw force"
assert picker.start_blocked_reason() is None
entry = picker.selected_entry()
assert entry is not None and entry.profile_id == "orisys.mini.lab-r1"
# The identity the row dropped is one hover away.
tooltip = picker.config_combo.itemData(0, Qt.ItemDataRole.ToolTipRole)
assert "orisys.mini.lab-r1" in tooltip
assert "descriptor bundle" in tooltip
picker.deleteLater()
def test_no_valid_row_shows_internal_identity_or_a_severity_word(mixed_picker):
"""Every row of every family, not only the one the fixture is built around."""
picker = mixed_picker
seen = 0
for family in (FAMILY_UNIDENTIFIED, "mini"):
picker.set_sensor_type(family)
for index in range(picker.config_combo.count()):
picker.config_combo.setCurrentIndex(index)
entry = picker.selected_entry()
if entry is None or not entry.valid:
continue
seen += 1
label = picker.config_combo.itemText(index)
assert entry.profile_id not in label, label
for word in ("ok", "degraded", "failed", "legacy"):
assert word not in label, f"{word!r} reached the row: {label}"
assert seen, "the fixture offered no valid entry, so nothing was checked"
def test_a_duplicate_row_names_its_twin(mixed_picker):
picker = mixed_picker
picker.set_sensor_type("mini")
first, second = (picker.config_combo.itemText(i) for i in (0, 1))
assert first == "Mini lab · User · Also provided by user/mini-b"
assert second == "Mini lab · User · Also provided by user/mini-a"
@@ -50,6 +50,17 @@ EXPECTED_SHIPPED: dict[str, tuple[str, str, str, bool, str]] = {
"mini": ("mini", "orisys.mini.legacy", "compound_eye", True, "degraded"),
}
#: What an operator reads in the picker, one row per shipped bundle: the product, where it came
#: from, and what the force numbers will be. ``box`` is the only shipped bundle that carries a
#: ``force_calibration.json``, so it is the only row naming a calibration; the other three are
#: stated as "no calibration selected — raw force" rather than as the SDK's ``degraded``.
EXPECTED_LABELS: dict[str, str] = {
"box": "Box · Shipped · Calibration: force_calibration.json",
"finger": "Finger R01 · Shipped · No calibration selected — raw force",
"finger-mems": "Finger (MEMS) · Shipped · No calibration selected — raw force",
"mini": "Mini · Shipped · No calibration selected — raw force",
}
@pytest.fixture(scope="module")
def shipped():
@@ -91,6 +102,55 @@ def test_the_shipped_families_keep_the_picker_order(shipped):
assert shipped.families() == ["box", "mini", "finger", "finger-mems"]
def test_every_shipped_row_states_what_it_is_where_from_and_what_it_gives(shipped):
"""The four real bundles, exactly as the picker renders them."""
for folder, expected in EXPECTED_LABELS.items():
entry = shipped.by_ref(f"shipped/{folder}")
assert entry is not None, folder
assert entry.label() == expected, folder
def test_no_shipped_row_carries_identity_severity_or_the_word_legacy(shipped):
"""The row is the first thing an operator reads; none of these belong in it.
``legacy`` means "resolved through the compatibility adapter because it has no descriptor",
which is not what the word says to a reader — and the finger bundle's ``profile_id`` is not
even that shape, so the column mixed two kinds of string.
"""
for entry in shipped.entries:
label = entry.label()
assert entry.profile_id not in label, label
for word in ("ok", "degraded", "failed", "legacy"):
assert word not in label, f"{word!r} reached the row: {label}"
def test_every_shipped_row_says_what_the_calibration_selection_is(shipped):
"""Calibration is a selection, so an absent one reads as raw force, not as a defect."""
for entry in shipped.entries:
label = entry.label()
selected = entry.selected_calibration
if selected:
assert f"Calibration: {selected}" in label, label
else:
assert "No calibration selected — raw force" in label, label
# The SDK calls an absent calibration a warning and the bundle "degraded"; the row
# says what that means for the operator instead of repeating the verdict.
assert entry.state == "degraded", entry.ref
def test_the_shipped_tooltips_keep_the_facts_the_rows_dropped(shipped):
for entry in shipped.entries:
tooltip = entry.tooltip()
assert entry.profile_id in tooltip, entry.ref
assert f"SDK validation: {entry.state}" in tooltip, entry.ref
shape = (
"descriptor bundle"
if entry.profile_id == FINGER_PROFILE_ID
else "compatibility bundle, no descriptor"
)
assert shape in tooltip, entry.ref
def test_box_mini_and_finger_mems_are_legacy_sensor_json_bundles(shipped):
for folder in ("box", "mini", "finger-mems"):
bundle = sensors_dir() / folder
+56 -38
View File
@@ -2003,10 +2003,10 @@ def test_the_replay_transport_plays_and_pauses(make_window, replay_recording):
assert window.replay.position() == parked, "the playhead kept moving while paused"
def test_the_recompute_mode_reruns_the_pipeline_over_the_video(
def test_the_recompute_switch_asks_for_analysis_not_an_acquire_chain(
make_window, replay_recording
):
"""Off: the recorded CSV. On: the recorded video through the vision chain."""
"""「从视频重算」 shares ``on_analysis_requested`` with opening a recording."""
from PySide6.QtWidgets import QApplication
window = _show(make_window(), (1440, 880))
@@ -2019,27 +2019,30 @@ def test_the_recompute_mode_reruns_the_pipeline_over_the_video(
assert window.act_recompute.isEnabled(), "the recompute switch is not offered"
started: list = []
asked: list = []
def fake_run() -> None:
started.append(True)
window.session.set_running(True)
original = window.replay_ctl.on_analysis_requested
def spy_analysis() -> None:
asked.append(True)
return original()
window._start_run = fake_run
window.replay_ctl.on_analysis_requested = spy_analysis
window.act_recompute.setChecked(True)
QApplication.processEvents()
assert started == [True], f"the pass started {len(started)} times"
assert not panel.is_recording_source(), "the charts still read the CSV"
assert window.toolbar.source_mode() == "video"
assert window.toolbar.video_path().endswith(".mp4"), window.toolbar.video_path()
assert window.toolbar.btn_run.isChecked(), (
"the transport does not show the pass as running"
)
assert asked == [True], "the switch did not ask for analysis"
assert started == [], "the switch must not open an acquisition chain"
# A recording's force measurement stays on forces.csv; analysis feeds 2D/3D.
assert panel.is_recording_source(), "analysis must not replace the recorded CSV"
window._recompute_running = True
window.act_recompute.setChecked(False)
QApplication.processEvents()
assert panel.is_recording_source(), "turning the recompute off did not restore the CSV"
assert not window.toolbar.btn_run.isChecked(), "the transport stayed checked"
assert panel.is_recording_source()
def test_the_replay_view_offers_a_stop_that_always_works(make_window, replay_recording):
@@ -2083,25 +2086,23 @@ def test_the_replay_view_offers_a_stop_that_always_works(make_window, replay_rec
window.statusBar().currentMessage()
)
# A recompute pass: the same button stops it, without stopping the recording.
stopped: list = []
window._stop_run = lambda **kwargs: stopped.append(kwargs.get("reason"))
window._recompute_running = True
window._update_run_enabled()
assert window.toolbar.btn_stop.isEnabled(), "no stop offered while a pass runs"
window.toolbar.btn_stop.click()
# An analysis pass left over from opening the recording: Stop cancels it.
# Opening may already have finished the pass; start one so the escape hatch is real.
window.replay_ctl.cancel_analysis()
window.replay_ctl.wait_for_analysis()
window.replay_ctl.on_analysis_requested()
QApplication.processEvents()
assert stopped == ["user_stop"], stopped
assert t("toolbar.stopped_pass") in window.statusBar().currentMessage(), (
window.statusBar().currentMessage()
)
if window.replay_ctl.analysis_running():
window._update_run_enabled()
assert window.toolbar.btn_stop.isEnabled(), "no stop offered while a pass runs"
window.toolbar.btn_stop.click()
QApplication.processEvents()
assert not window.replay_ctl.analysis_running()
assert t("toolbar.stopped_pass") in window.statusBar().currentMessage(), (
window.statusBar().currentMessage()
)
# Nothing running: it says so instead of doing nothing at all.
# (The pass is over, so the stop button is correctly disabled — the escape hatch is
# therefore exercised the way Esc reaches it, through the controller.)
assert not window.toolbar.btn_stop.isEnabled(), (
"a stop is still offered after the pass was stopped"
)
window.replay_ctl.stop_activity()
QApplication.processEvents()
assert t("toolbar.nothing_to_stop") in window.statusBar().currentMessage(), (
@@ -2344,13 +2345,20 @@ def test_the_two_figures_are_drawn_from_one_arrow_policy(make_window, monkeypatc
window.close()
def test_a_bare_video_plays_even_though_it_is_not_a_recording(make_window, record_root):
def test_a_bare_video_plays_even_though_it_is_not_a_recording(
make_window, record_root, decline_video_prompt
):
"""A clip with no forces.csv beside it is still something to watch.
An operator who points at a video expects the transport to play it. Opening the
video's own folder as a *recording* used to "succeed" against the tolerant reader
and then play a zero-length timeline ("pressing 回放 does nothing"), so the fix is
not to refuse the file but to play it as what it is: a video, with empty charts.
The setup prompt is declined here (``decline_video_prompt``): this test is about the
playback, and a dismissed prompt is the answer that leaves it exactly as it was. What
accepting the prompt does instead is its own module
(``test_bare_video_analysis.py``).
"""
import time as _time
@@ -2401,14 +2409,16 @@ def test_a_bare_video_plays_even_though_it_is_not_a_recording(make_window, recor
assert window._play_timer.isActive()
image = window.replay.video_image()
assert image is not None and not image.isNull(), "no frame was decoded"
window._stop_replay_activity()
window.replay_ctl.stop_activity()
QApplication.processEvents()
assert not window._play_timer.isActive()
finally:
window.close()
def test_the_replay_and_the_live_paths_share_one_image_backend(make_window, record_root):
def test_the_replay_and_the_live_paths_share_one_image_backend(
make_window, record_root, decline_video_prompt
):
"""Stage 7: one renderer for every 2D producer, and modes from the frame itself.
A recording stores the raw camera image and nothing else, so its plan offers exactly
@@ -2444,7 +2454,7 @@ def test_the_replay_and_the_live_paths_share_one_image_backend(make_window, reco
app_module.render_image = spy
try:
window._apply_view_preset("replay")
window.layout_ctl.apply_preset("replay")
assert window._load_replay_video(bare) is True
window._render_replay_frame()
QApplication.processEvents()
@@ -2503,7 +2513,9 @@ def test_the_application_converts_bgr_to_qimage_in_one_place():
assert others == [], f"another BGR→QImage conversion exists: {others}"
def test_the_replay_entries_exist_only_in_the_replay_view(make_window, record_root):
def test_the_replay_entries_exist_only_in_the_replay_view(
make_window, record_root, decline_video_prompt
):
"""Stage 7: the live view has no replay entry, and says where they live.
Opening a video from the live view used to set the acquisition source to a file
@@ -2523,7 +2535,7 @@ def test_the_replay_entries_exist_only_in_the_replay_view(make_window, record_ro
window = _show(make_window(), (1440, 880))
try:
window._apply_view_preset("live")
window.layout_ctl.apply_preset("live")
QApplication.processEvents()
for action in (window.act_open_video, window.act_playback, window.act_playback_latest):
assert not action.isEnabled(), "a replay entry is offered in the live view"
@@ -2542,7 +2554,7 @@ def test_the_replay_entries_exist_only_in_the_replay_view(make_window, record_ro
# In the replay view they are offered again, and the recompute entry follows
# whether there is anything to recompute from.
window._apply_view_preset("replay")
window.layout_ctl.apply_preset("replay")
QApplication.processEvents()
for action in (window.act_open_video, window.act_playback, window.act_playback_latest):
assert action.isEnabled()
@@ -2552,10 +2564,16 @@ def test_the_replay_entries_exist_only_in_the_replay_view(make_window, record_ro
assert window.toolbar.btn_recompute.isEnabled(), (
"a video with no recording beside it can still be recomputed"
)
# Declined setup (fixture) means play-only: the switch re-asks, then clears. It
# must not open an acquisition chain over the clip.
started: list = []
window._start_run = lambda: started.append(True)
window.toolbar.btn_recompute.click()
QApplication.processEvents()
assert window.act_recompute.isChecked(), "the rail's recompute entry did not toggle"
window.act_recompute.setChecked(False)
assert started == [], "recompute must not open an acquisition chain"
assert not window.act_recompute.isChecked(), (
"declining analysis setup must leave the switch off"
)
finally:
window.close()
@@ -2613,7 +2631,7 @@ def test_a_video_inside_a_recording_opens_the_recording_and_plays(
window.toolbar.btn_run.setChecked(True)
QApplication.processEvents()
assert window._play_timer.isActive(), "pressing 回放 did not start playback"
window._pause_playback()
window.replay_ctl.pause_playback()
window.close()
+8 -5
View File
@@ -146,6 +146,9 @@ class _Replay:
def sync_actions(self) -> None:
self.synced += 1
def set_active_view(self, preset: str) -> None:
self.active_view = str(preset)
class _Host:
def __init__(self, prefs: dict) -> None:
@@ -155,19 +158,19 @@ class _Host:
self.rail_updates = 0
self.source_kind = "live"
def _persist_prefs(self) -> None:
def persist_prefs(self) -> None:
self.persisted += 1
def _update_run_enabled(self) -> None:
def update_run_enabled(self) -> None:
self.run_updates += 1
def _refresh_status_rail(self) -> None:
def refresh_status_rail(self) -> None:
self.rail_updates += 1
def _current_source_kind(self) -> str:
def current_source_kind(self) -> str:
return self.source_kind
def _device_geometry_provider(self):
def device_geometry_provider(self):
return None