From c256870ee24602c6c9472eae5feb9fb560d1496b Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Tue, 20 Oct 2026 20:33:23 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=8C=89=E5=A4=8D=E4=BD=8D=E8=AF=B7?= =?UTF-8?q?=E6=B1=82=E7=9B=AE=E6=A0=87=E7=A1=AE=E8=AE=A4=E5=8F=82=E8=80=83?= =?UTF-8?q?=E7=89=88=E6=9C=AC=E5=B9=B6=E9=9A=94=E7=A6=BB=E5=A4=8D=E4=BD=8D?= =?UTF-8?q?=E7=8A=B6=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 防止在途旧帧提前触发归零，并保留旧 reset 用法。SDK 101 项针对性回归与 lint 通过。 --- docs/force-workflow.zh.md | 10 ++ docs/sensor-bundles.md | 16 ++- src/orisys/force_readout.py | 27 ++++- src/orisys/force_session.py | 27 +++-- src/orisys/pipeline.py | 122 ++++++++++++++------ src/orisys/sensor.py | 50 +++++++- tests/test_force_session.py | 63 ++++++++++ tests/test_pipeline.py | 222 ++++++++++++++++++++++++++++++++++++ 8 files changed, 477 insertions(+), 60 deletions(-) diff --git a/docs/force-workflow.zh.md b/docs/force-workflow.zh.md index d5f5151..efb4846 100644 --- a/docs/force-workflow.zh.md +++ b/docs/force-workflow.zh.md @@ -65,6 +65,16 @@ session = ForceSession( 基线未完成、无效帧、窗口不足或已取消时 N 为 `NaN`、状态显式；窗口不足不越界补帧。 录像重算用 `replay_window=(start, end)` 在声明区间执行同一操作。 +三级流水线已有同帧测量时，用 `process_measurement(row, stamp, tracker_reset_seq=...)` +传入快照自带的参考版本，不读流水线的实时计数。外部归零先调用 +`target = pipeline.request_reset()`，再调用 +`request_zero_external_reset(reference_id=..., declared_unloaded=True, target_reference_version=target)`。 +`target` 为正整数，帧版本达到它才开始采样；较早复位后已在途的帧不能确认本次请求。 +未被 flow 领取的请求合并为同一目标，领取后的新请求另分配目标；共用目标的复位与归零只确认一次。 +该接缝尚未发布，本轮将原 `tracker_reset_seq` 请求参数改为明确的目标版本参数； +逐帧输入仍用 `tracker_reset_seq`。已发布 `Sensor.reset()` / `StreamPipeline.reset()` 用法不变。 +实现见 `pipeline.py`、`force_session.py`，回归见 `tests/test_pipeline.py`、`tests/test_force_session.py`。 + ## 后处理与有效设置 `OutputSettings(sample_rate_hz, median_window=1, lowpass_hz=0.0, lowpass_order=2, diff --git a/docs/sensor-bundles.md b/docs/sensor-bundles.md index c51e8de..c373fb3 100644 --- a/docs/sensor-bundles.md +++ b/docs/sensor-bundles.md @@ -165,6 +165,16 @@ compute the same numbers. ## Resetting the reference Both backends process externally supplied frames through the same profile. After a gap in the -input - a pause, a seek, a scene change - call `session.reset()`. It is queued to the stage that -owns the tracker, so the reference frame is rebased on the frame about to be computed rather than -on a buffer the pipeline is still reading. +input - a pause, a seek, a scene change - call `session.reset()`. It is queued to the stages that +own the sensor state, so the reference frame is rebased on the frame about to be computed rather +than on a buffer the pipeline is still reading: the flow stage resets the image/tracker state and +bumps the reference version, and the HHD stage resets the force/readout state (filters, contact +flags, calibration session) on the first frame carrying the new version, before decomposing it. +`FrameSnapshot.tracker_reset_seq` is the version that frame was measured with. + +When driving `StreamPipeline` directly, call `target = pipeline.request_reset()` to obtain the +future reference version reserved by that request. Confirm it only when a snapshot carries +`tracker_reset_seq >= target`; the current output frame can trail an earlier reset and cannot +identify a new request. Requests share a target until flow claims it; a later request gets a new +target even while the earlier reset is still being applied. The existing `reset()` returns `None`. +Regression coverage lives in `tests/test_pipeline.py` (request coalescing and reset-in-progress). diff --git a/src/orisys/force_readout.py b/src/orisys/force_readout.py index 2a8103b..b7692cc 100644 --- a/src/orisys/force_readout.py +++ b/src/orisys/force_readout.py @@ -374,14 +374,37 @@ class ForceReadout: ) return filter_changed - def reset(self) -> None: + def reset_flow_ema(self) -> None: + """Drop the optical-flow EMA history (image side). + + The EMA smooths the flow field before HHD, so it belongs to the stage + that computes flow. A streaming pipeline resets this half on its flow + thread, separately from the readout history the HHD stage owns. + """ + self._ema_flow = None + + def reset_readout(self) -> None: + """Drop the force-filter history and the last reading (force side). + + The median/low-pass memory and the last reading belong to the stage that + filters each measurement; this half does not touch the flow EMA. + """ self._force_filter.reset() self._force_median.reset() - self._ema_flow = None self.filtered_raw = None self.raw_unfiltered = None self.unfiltered_newton = None + def reset(self) -> None: + """Reset both halves at once (single-threaded or frame-boundary use). + + A streaming pipeline applies :meth:`reset_flow_ema` on the flow thread + and :meth:`reset_readout` on the HHD thread; calling this keeps the + historical behaviour for callers that own the whole sensor. + """ + self.reset_readout() + self.reset_flow_ema() + def state(self, *, dis_mode=None) -> dict: """Effective filter chain, for records and audits.""" return { diff --git a/src/orisys/force_session.py b/src/orisys/force_session.py index 8cd6167..866cdf3 100644 --- a/src/orisys/force_session.py +++ b/src/orisys/force_session.py @@ -359,9 +359,8 @@ class ForceSession: ) self._pending_reset: dict | None = None #: A queued zero whose tracker reset is performed by an external stage - #: (the App delegates it to ``StreamPipeline.reset``). The value is the - #: pipeline reset serial current when the request was made; the first - #: processed frame with a newer serial begins the baseline window. + #: (the App delegates it to ``StreamPipeline.request_reset``). The value + #: is the reserved target version, not an observed frame/live counter. self._pending_external_reset: int | None = None self._zero_samples: list[tuple[float, float, float]] = [] self._replay_window: tuple[int, int] | None = None @@ -757,16 +756,16 @@ class ForceSession: reference_id: str, declared_unloaded: bool, replay_window: tuple[int, int] | None = None, - tracker_reset_seq: int, + target_reference_version: int, ) -> ZeroProgress: """Request the one unloaded zero when another stage performs the tracker reset. Additive variant of :meth:`request_zero` for a consumer that drives its measurement through a concurrent pipeline: the caller queues the tracker - reset on the pipeline's flow stage (``StreamPipeline.reset``) and passes - the pipeline's reset serial at request time. Frames published with that - same serial were measured before the reset and are never baseline frames; - the first processed frame with a newer serial begins the 30-frame window. + reset with ``StreamPipeline.request_reset`` and passes its returned target + version. Older versions cannot start the 30-frame window, including frames + from an earlier reset already in flight when this request was queued. + Collection begins at the first frame whose version reaches the target. Generation, the declared-unloaded requirement and the filter reset are identical to :meth:`request_zero`. """ @@ -776,13 +775,13 @@ class ForceSession: ) if not isinstance(reference_id, str) or not reference_id.strip(): raise ValueError("reference_id must be a non-empty string") - if type(tracker_reset_seq) is not int or tracker_reset_seq < 0: - raise ValueError("tracker_reset_seq must be a non-negative integer") + if type(target_reference_version) is not int or target_reference_version <= 0: + raise ValueError("target_reference_version must be a positive integer") window = _coerce_replay_window(replay_window) self._generation += 1 self._clear_frame_caches() self._pending_reset = None - self._pending_external_reset = int(tracker_reset_seq) + self._pending_external_reset = target_reference_version self._zero_state = "resetting" self._zero_samples = [] self._replay_window = window @@ -1277,8 +1276,8 @@ class ForceSession: ``tracker_reset_seq`` is the publishing pipeline's reset serial (:attr:`~orisys.pipeline.FrameSnapshot.tracker_reset_seq`). It is only consulted by a zero requested through - :meth:`request_zero_external_reset`: frames with the recorded serial (or an - older one) were measured before the reset and are not baseline frames. + :meth:`request_zero_external_reset`: versions below its reserved target + were measured before that reset and are not baseline frames. """ if not isinstance(row, FrameMeasurement): raise TypeError("row must be a FrameMeasurement") @@ -1304,7 +1303,7 @@ class ForceSession: raise ValueError( "tracker_reset_seq is required while an external-reset zero is queued" ) - if int(tracker_reset_seq) <= external: + if int(tracker_reset_seq) < external: # Measured before the externally queued tracker reset: not a # baseline frame, and the session-owned reset path is not involved. normal_reset_after = False diff --git a/src/orisys/pipeline.py b/src/orisys/pipeline.py index e17c694..0cb2acf 100644 --- a/src/orisys/pipeline.py +++ b/src/orisys/pipeline.py @@ -157,11 +157,13 @@ class FrameSnapshot: force_units: tuple[str, str, str] | None = None calibration_status: dict[str, str] | None = None contact_found: bool | None = None - #: Monotonic count of tracker resets the flow stage has applied. A consumer - #: that delegates a frame-boundary reset to :meth:`StreamPipeline.reset` can - #: use this to tell frames measured before that reset from frames measured - #: after it (the reset is queued, not synchronous). Additive field: producers - #: that do not use it leave it at 0. + #: Reference version this frame was measured with: the count of image-side + #: tracker resets the flow stage had applied when it computed this frame's + #: flow. A consumer that delegates a frame-boundary reset to + #: :meth:`StreamPipeline.request_reset` compares it with the returned target to + #: tell frames measured before that reset from frames measured after it (the + #: reset is queued, not synchronous). Additive field: producers that do not + #: use it leave it at 0. tracker_reset_seq: int = 0 @@ -234,13 +236,19 @@ class StreamPipeline: self._threads: list[threading.Thread] = [] self._error: BaseException | None = None self._running = False - #: Set by :meth:`reset` and consumed by the flow stage, so that resetting sensor state - #: never races the stage that owns that state. - self._reset_pending = threading.Event() - #: How many tracker resets the flow stage has actually applied. Published on - #: every snapshot so a consumer can tell a frame measured before a queued - #: reset from one measured after it. + #: Request allocation and flow-stage claiming share a lock. A request + #: already claimed by flow cannot acknowledge another request made later. + self._reset_lock = threading.Lock() + self._requested_reset_version = 0 + self._pending_reset_version: int | None = None + #: How many image-side resets the flow stage has actually applied. A frame + #: carries the value in effect when its flow was computed (through the flow + #: queue); consumers read :attr:`FrameSnapshot.tracker_reset_seq`, not this. self._tracker_reset_seq = 0 + #: The reference version whose force-side reset the HHD stage has applied. + #: The flow stage bumps the version; the HHD stage owns the force state and + #: applies its half on the first frame that carries a newer version. + self._force_reset_applied_version = 0 #: Pause state. ``_pause_level`` is what a pause would hold when armed; #: ``_currently_capturing`` records whether the producer is still #: reading, which decides whether a resume has to rebase the tracker. @@ -330,24 +338,46 @@ class StreamPipeline: def tracker_reset_seq(self) -> int: """Tracker resets the flow stage has applied so far (queued, not immediate). - A consumer that asked for a reset through :meth:`reset` uses this (and - :attr:`FrameSnapshot.tracker_reset_seq`) to identify the first frame whose - measurement was produced after the reset. + This is the live diagnostic count, not a frame's identity: a reset queued + now is only applied before a later frame is measured. Read + :attr:`FrameSnapshot.tracker_reset_seq` for the version a frame carries. """ return int(self._tracker_reset_seq) def reset(self) -> None: - """Reset the sensor's tracker and force-filter state, safely. + """Request a reference reset, applied at frame boundaries by its owning stage. Call this from the consumer thread (after a key press, on a drift alarm). The reset is - queued and applied by the flow stage before its next frame, never inline: ``Sensor.reset`` - rebases ``initial_img`` on the current ``Sensor.img``, and the flow stage is reading that - buffer concurrently, so resetting from the caller's thread would capture a half-written - reference frame. + queued, never inline: ``Sensor.reset_image_state`` rebases ``initial_img`` on the current + ``Sensor.img``, and the flow stage is reading that buffer concurrently, so resetting from + the caller's thread would capture a half-written reference frame. - Idempotent: a second call before the next frame does not queue twice. + The two halves are applied by the threads that own their state. The flow stage resets the + image/tracker state and bumps the reference version before its next frame; the HHD stage + resets the force/contact state on the first frame that carries the new version, before + decomposing it. A sensor that implements only ``reset()`` keeps the historical whole reset + on the flow stage, and then no force-side reset is applied by the HHD stage. + + Requests coalesce until the flow stage claims them. Use :meth:`request_reset` + when the caller needs a target version to confirm completion. """ - self._reset_pending.set() + self.request_reset() + + def request_reset(self) -> int: + """Queue a reset and return the reference version that will confirm it. + + Requests coalesce only until the flow stage claims them. A request made + while an earlier reset is being applied gets a new target, even though + :attr:`tracker_reset_seq` still reports the previous applied version. + Wait for a snapshot with ``tracker_reset_seq >= target``; neither the + current snapshot nor a separately read live counter identifies a request. + Sensor state is always reset by its owning stages, outside this lock. + """ + with self._reset_lock: + if self._pending_reset_version is None: + self._requested_reset_version += 1 + self._pending_reset_version = self._requested_reset_version + return self._pending_reset_version @property def paused(self) -> bool: @@ -399,7 +429,7 @@ class StreamPipeline: # Frames read during the pause were never computed, so the sensor's # reference frame is stale by however long the hold lasted. Ask for a # rebase; the flow stage applies it before its next frame. - self._reset_pending.set() + self.reset() # Arm capture *before* lifting the hold. The producer stops as soon as # the pause flag is set and only returns once it is clear, so a level-1 # resume must have `_currently_capturing` true by then or the producer @@ -585,14 +615,23 @@ class StreamPipeline: self._drain(self._q1) self._drain(self._q2) continue - # Applied here, on the thread that owns the sensor's flow state, before this frame is - # computed: see reset() for why this must not run on the caller's thread. - if self._reset_pending.is_set(): - self._reset_pending.clear() - self.sensor.reset() - # Published on this frame's snapshot: this is the first frame whose - # measurement (flow) was produced after the reset was applied. - self._tracker_reset_seq += 1 + # Applied here, on the thread that owns the sensor's image/flow state, before this + # frame is computed: see reset() for why this must not run on the caller's thread. + with self._reset_lock: + reset_version = self._pending_reset_version + self._pending_reset_version = None + if reset_version is not None: + reset_image_state = getattr(self.sensor, "reset_image_state", None) + if reset_image_state is not None: + reset_image_state() + else: + # A sensor without the split seam keeps the historical whole reset on this + # thread; its force state is then not reset by the HHD stage. + self.sensor.reset() + self._tracker_reset_seq = reset_version + # The version travels with the frame: the HHD stage publishes the snapshot later and + # must not read the live counter at that point. + reference_version = self._tracker_reset_seq frame, t_cap, produce_s, source_metadata = item t0 = time.perf_counter() # The frame is passed explicitly: the SDK's self.img slot is shared @@ -605,7 +644,8 @@ class StreamPipeline: flow_copy = flow.copy() if isinstance(flow, np.ndarray) else None self._put( self._q2, - (frame, flow_copy, t_cap, produce_s, flow_s, source_metadata), + (frame, flow_copy, t_cap, produce_s, flow_s, source_metadata, + reference_version), ) self._put(self._q2, None) @@ -621,7 +661,15 @@ class StreamPipeline: if not self._wait_out_pause(): break continue - frame, flow, t_cap, produce_s, flow_s, source_metadata = item + frame, flow, t_cap, produce_s, flow_s, source_metadata, reference_version = item + # The force/contact state belongs to this stage, so its half of a + # queued reset is applied here, on the first frame of the new version, + # before the decomposition that would consume the stale model state. + if reference_version > self._force_reset_applied_version: + self._force_reset_applied_version = reference_version + reset_force_state = getattr(self.sensor, "reset_force_state", None) + if reset_force_state is not None: + reset_force_state() t0 = time.perf_counter() # Explicit flow: compute_hhd reads self.flow, which the flow stage # overwrites for the next frame, so it cannot be the transport. @@ -630,7 +678,8 @@ class StreamPipeline: self.sensor.compute_contact() hhd_s = time.perf_counter() - t0 snap = self._build( - frame, flow, produce_s, flow_s, hhd_s, t_cap, source_metadata + frame, flow, produce_s, flow_s, hhd_s, t_cap, source_metadata, + reference_version, ) self._publish(snap) if self._on_frame is not None: @@ -652,7 +701,8 @@ class StreamPipeline: def _build(self, frame: np.ndarray, flow: np.ndarray | None, produce_s: float, flow_s: float, hhd_s: float, t_cap: float, - source_metadata: Any = None) -> FrameSnapshot: + source_metadata: Any = None, + reference_version: int = 0) -> FrameSnapshot: sensor = self.sensor info = sensor.info include = self._include @@ -697,7 +747,9 @@ class StreamPipeline: hhd_ms=hhd_s * 1000.0, image_size_raw=image_size_raw, source_metadata=source_metadata, - tracker_reset_seq=int(self._tracker_reset_seq), + # This frame's own measurement version, carried through the flow queue: + # never the live counter, which a later frame may already have bumped. + tracker_reset_seq=int(reference_version), ) def _fps(self) -> float: diff --git a/src/orisys/sensor.py b/src/orisys/sensor.py index c223e2e..02d6af1 100644 --- a/src/orisys/sensor.py +++ b/src/orisys/sensor.py @@ -1138,15 +1138,53 @@ class Sensor: return tuple(mapping.get(arg, lambda: None)() for arg in args if isinstance(arg, self.info)) - def reset(self): - """Reset optical flow tracker and force-filter state.""" - self.reset_tracker() - # The unfiltered pair belongs to the frame it was computed for: a new - # session must not report the previous one's reading. - self._reset_calibration_session() + def reset_image_state(self) -> None: + """Reset the image/flow half of the pipeline state (flow thread only). + + Rebases the optical-flow tracker on the current ``img`` and drops the flow + EMA. :class:`StreamPipeline` calls this on its flow thread before computing + the first frame of a new reference version, because that thread owns the + image buffers. The force half belongs to the HHD thread and is reset + separately by :meth:`reset_force_state`. + """ + self.__tracker.reset() + self._force_readout.reset_flow_ema() if self.img is not None: self.initial_img = self.img.copy() + def reset_force_state(self) -> None: + """Reset the force/contact half of the pipeline state (HHD thread only). + + Drops the readout filters and the last reading, the contact flags and the + calibration session (model zero/reference), so the first frame of a new + reference version is measured from a clean force state. The image half + belongs to the flow thread and is reset by :meth:`reset_image_state`. + """ + self._force_readout.reset_readout() + # The unfiltered pair belongs to the frame it was computed for: a new + # session must not report the previous one's reading. + self._force_unfiltered = None + self._force_raw_unfiltered = None + self._force_filtered_raw = None + self.compute_deformation_valid = False + self.compute_contact_valid = False + self.centroid_found = False + self._reset_calibration_session() + + def reset(self): + """Reset the image and force state together (single-threaded use). + + Kept for callers that own the whole sensor without the streaming + pipeline, and for a third-party sensor/stage that has no split seam: the + final state is unchanged from before the split. :class:`StreamPipeline` + applies the two halves on the threads that own them instead - + :meth:`reset_image_state` on the flow stage and + :meth:`reset_force_state` on the HHD stage, before the first frame of the + new reference version. + """ + self.reset_image_state() + self.reset_force_state() + def disconnect(self): """Release resources and stop async threads.""" diff --git a/tests/test_force_session.py b/tests/test_force_session.py index d541fce..22f89bb 100644 --- a/tests/test_force_session.py +++ b/tests/test_force_session.py @@ -155,6 +155,33 @@ def collect_baseline(session, source, *, start=1, count=30, reference_id="ref-1" ] +def test_external_zero_waits_for_the_requested_target_reference_version(): + source = FakeMeasurementSource(raw_for=lambda _: (10.0, 20.0, 30.0)) + session = make_session(source) + session.request_zero_external_reset( + reference_id="second-reset", declared_unloaded=True, target_reference_version=2 + ) + outputs = [] + for frame_id, version in enumerate([0, 1, 1, 2, 2]): + row = source.read_measurement_frame(None, frame_id=frame_id) + outputs.append(session.process_measurement(row, stamp(frame_id), tracker_reset_seq=version)) + assert [o.measurement.zero.phase for o in outputs] == [ + "resetting", "resetting", "resetting", "collecting", "collecting" + ] + assert outputs[-1].measurement.zero.collected_frames == 2 + assert outputs[-1].measurement.zero.first_frame_id == 3 + + +@pytest.mark.parametrize("target", [0, -1, True]) +def test_external_zero_rejects_invalid_target_reference_version(target): + session = make_session(FakeMeasurementSource()) + with pytest.raises(ValueError, match="target_reference_version"): + session.request_zero_external_reset( + reference_id="invalid-target", declared_unloaded=True, + target_reference_version=target, + ) + + # --------------------------------------------------------------------------- # T1: executable fake chain, one model call, cached reads, missing is not zero # --------------------------------------------------------------------------- @@ -1210,3 +1237,39 @@ def test_old_readout_entry_keeps_filter_then_convert_order() -> None: # is exactly the recorded historical behaviour. assert outputs[-1] == pytest.approx(4.0) assert readout.raw_unfiltered == (250.0, 0.0, 0.0) + + +def test_readout_reset_halves_clear_only_their_own_state() -> None: + """The flow EMA is image-side state; the readout history is force-side state.""" + from orisys.force_readout import ForceReadout + + flow = np.zeros((2, 2, 2), dtype=np.float32) + + readout = ForceReadout({"flow_ema_alpha": 0.5, "force_median_window": 3}, + config_fps=30.0) + readout.apply_flow_ema(flow) + readout.apply((10.0, 1.0, 2.0)) + filtered = readout.filtered_raw + raw_unfiltered = readout.raw_unfiltered + unfiltered_newton = readout.unfiltered_newton + + readout.reset_flow_ema() + assert readout._ema_flow is None + # The force-side history survives a reset that only rebases the image stage. + assert readout.filtered_raw == filtered + assert readout.raw_unfiltered == raw_unfiltered + assert readout.unfiltered_newton == unfiltered_newton + + readout = ForceReadout({"flow_ema_alpha": 0.5, "force_median_window": 3}, + config_fps=30.0) + readout.apply_flow_ema(flow) + readout.apply((10.0, 1.0, 2.0)) + readout.reset_readout() + assert readout._ema_flow is not None + assert readout.filtered_raw is None + assert readout.raw_unfiltered is None + assert readout.unfiltered_newton is None + + readout.reset() + assert readout._ema_flow is None + assert readout.filtered_raw is None diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 3c82a59..5c1cd35 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -429,3 +429,225 @@ def test_snapshot_raw_capture_size_is_none_when_unreported(): time.sleep(0.001) pipe.stop() assert frames and frames[0].image_size_raw is None + + +class SplitResetFakeSensor(FakeSensor): + """A fake with the split reset seam and event-gated flow/HHD stages. + + The two halves of a reset belong to different pipeline threads, and the + version a frame is stamped with depends on which half has already run. The + events pin both stages to a known frame so the regression orders the race + instead of hoping for it: the flow stage parks at the end of its second + frame and the HHD stage parks inside its first decomposition. + """ + + FLOW_GATE_CALL = 2 + + def __init__(self) -> None: + super().__init__() + self.flow_gate_reached = threading.Event() + self.release_flow_gate = threading.Event() + self.hhd_gate_reached = threading.Event() + self.release_hhd_gate = threading.Event() + self.hhd_calls = 0 + self.image_reset_threads: list[str] = [] + self.force_reset_threads: list[str] = [] + self.force_reset_at_hhd_calls: list[int] = [] + + def compute_flow(self, img=None, **kwargs) -> None: + super().compute_flow(img, **kwargs) + # Park after frame 2's flow. The next frame that reaches the flow stage + # is then provably the first one that can observe a reset queued now. + if self._flow_calls == self.FLOW_GATE_CALL: + self.flow_gate_reached.set() + self.release_flow_gate.wait(timeout=5.0) + + def compute_hhd(self, flow=None, **kwargs) -> None: + self.hhd_calls += 1 + if self.hhd_calls == 1: + self.hhd_gate_reached.set() + self.release_hhd_gate.wait(timeout=5.0) + super().compute_hhd(flow, **kwargs) + + def reset_image_state(self) -> None: + self.image_reset_threads.append(threading.current_thread().name) + + def reset_force_state(self) -> None: + self.force_reset_threads.append(threading.current_thread().name) + self.force_reset_at_hhd_calls.append(self.hhd_calls) + + def reset(self) -> None: + # Mirrors Sensor.reset(): both halves, immediately, on the caller's thread. + self.reset_calls += 1 + self.reset_at_frame.append(self._flow_calls) + self.reset_image_state() + self.reset_force_state() + + +class LegacyResetFakeSensor(FakeSensor): + """A stand-in with only the historical ``reset()``, like a third-party sensor.""" + + def __init__(self) -> None: + super().__init__() + self.reset_threads: list[str] = [] + + def reset(self) -> None: + self.reset_calls += 1 + self.reset_at_frame.append(self._flow_calls) + self.reset_threads.append(threading.current_thread().name) + + +def test_reset_request_target_is_shared_until_flow_claims_it(): + sensor = FakeSensor() + frames = [] + pipe = StreamPipeline(sensor, make_producer(sensor, 2), on_frame=frames.append) + assert pipe.request_reset() == 1 + assert pipe.reset() is None # The published entry keeps its return contract. + assert pipe.request_reset() == 1 + assert pipe.tracker_reset_seq == 0 + pipe.start() + try: + assert _wait_until(lambda: len(frames) == 2) + assert [snap.tracker_reset_seq for snap in frames] == [1, 1] + assert sensor.reset_calls == 1 + assert pipe.request_reset() == 2 + finally: + pipe.stop() + + +def test_reset_requested_while_previous_reset_is_running_gets_a_new_target(): + entered = threading.Event() + release = threading.Event() + + class Sensor(LegacyResetFakeSensor): + def reset(self): + super().reset() + if self.reset_calls == 1: + entered.set() + assert release.wait(timeout=5.0) + + sensor = Sensor() + frames = [] + pipe = StreamPipeline(sensor, make_producer(sensor, 2), on_frame=frames.append) + assert pipe.request_reset() == 1 + pipe.start() + try: + assert entered.wait(timeout=5.0) + # Flow has claimed request 1, but has not finished applying it. The live + # counter is still 0; request 2 must not be merged into request 1. + assert pipe.tracker_reset_seq == 0 + assert pipe.request_reset() == 2 + assert pipe.request_reset() == 2 + release.set() + assert _wait_until(lambda: len(frames) == 2) + assert [snap.tracker_reset_seq for snap in frames] == [1, 2] + assert sensor.reset_calls == 2 + finally: + release.set() + pipe.stop() + + +def _wait_until(predicate, timeout: float = 5.0) -> bool: + """Poll a predicate; the events, not this poll, order the race.""" + deadline = time.perf_counter() + timeout + while time.perf_counter() < deadline: + if predicate(): + return True + time.sleep(0.001) + return predicate() + + +def _run_gated_reset(sensor: SplitResetFakeSensor, frames: list, count: int = 3) -> None: + """Drive two pre-reset frames and one post-reset frame through the real loops.""" + pipe = StreamPipeline(sensor, make_producer(sensor, count), + on_frame=frames.append, depth=2) + pipe.start() + try: + assert sensor.flow_gate_reached.wait(timeout=5.0), "flow stage never reached its gate" + assert sensor.hhd_gate_reached.wait(timeout=5.0), "HHD stage never reached its gate" + # Both stages are provably parked: no frame has been published, and the + # frame whose flow would observe this reset has not been computed yet. + pipe.reset() + sensor.release_flow_gate.set() + assert _wait_until(lambda: sensor._flow_calls >= 3), ( + "flow stage did not compute the first post-reset frame" + ) + sensor.release_hhd_gate.set() + deadline = time.perf_counter() + 5.0 + while len(frames) < count and time.perf_counter() < deadline: + time.sleep(0.001) + finally: + sensor.release_flow_gate.set() + sensor.release_hhd_gate.set() + pipe.stop() + assert len(frames) == count, f"expected {count} published frames, got {len(frames)}" + + +def test_frames_keep_the_reference_version_they_were_measured_with(): + """A reset queued after a frame's flow must not relabel that frame. + + Frames 0 and 1 are measured while the HHD stage is held on the first + snapshot. The caller then queues a reset, and frame 2 is measured with the + new version. Each snapshot must carry the version its own flow was computed + with; reading the live counter at publish time would stamp the two older + frames with the newer version. + """ + sensor = SplitResetFakeSensor() + frames: list[FrameSnapshot] = [] + _run_gated_reset(sensor, frames) + versions = [snap.tracker_reset_seq for snap in frames] + assert versions == [0, 0, 1], f"expected per-frame versions [0, 0, 1], got {versions}" + + +def test_reset_halves_run_on_the_threads_that_own_their_state(): + """Image state resets on the flow stage, force state on the HHD stage. + + The force half must land on the HHD stage before it decomposes the first + frame of the new version. Resetting it from the flow thread - the old + behaviour - mutates model/zero state that the HHD stage is using. + """ + sensor = SplitResetFakeSensor() + frames: list[FrameSnapshot] = [] + _run_gated_reset(sensor, frames) + assert sensor.image_reset_threads == ["orisys-pipeline-flow"] + assert sensor.force_reset_threads == ["orisys-pipeline-hhd"] + # Two decompositions (frames 0 and 1) had completed when the force reset ran, + # so it preceded the third one - the first frame of the new version. + assert sensor.force_reset_at_hhd_calls == [2] + assert sensor.hhd_calls == 3 + assert [snap.tracker_reset_seq for snap in frames] == [0, 0, 1] + + +def test_legacy_sensor_with_only_reset_is_reset_on_the_flow_stage(): + """A sensor without the split methods keeps the historical fallback.""" + sensor = LegacyResetFakeSensor() + ready = threading.Event() + release = threading.Event() + state = {"produced": 0} + + def gated_produce(): + ready.set() + release.wait(timeout=5.0) + if state["produced"]: + return None + state["produced"] = 1 + return sensor.make_frame(0) + + frames: list = [] + pipe = StreamPipeline(sensor, gated_produce, on_frame=frames.append) + pipe.start() + try: + assert ready.wait(timeout=5.0) + pipe.reset() + assert sensor.reset_calls == 0, "reset must not run on the caller's thread" + release.set() + deadline = time.perf_counter() + 5.0 + while not frames and time.perf_counter() < deadline: + time.sleep(0.001) + finally: + release.set() + pipe.stop() + assert len(frames) == 1 + assert sensor.reset_calls == 1 + assert sensor.reset_threads == ["orisys-pipeline-flow"] + assert sensor.reset_at_frame == [0], "the fallback reset must precede compute_flow"