fixed commetn from songxiao

This commit is contained in:
ykcheungab
2026-09-25 17:46:37 +08:00
parent 6effe74172
commit 12781854ca
56 changed files with 4935 additions and 419 deletions
@@ -27,6 +27,7 @@ from orisys_app_shared.acquisition.sdk_worker import (
FILE_FIRST_FRAME_TIMEOUT_S,
FPS_WINDOW_S,
HANDOVER_TIMEOUT_S,
PAUSE_POLL_S,
STOP_JOIN_TIMEOUT_S,
VIDEO_EOF_TIMEOUT_S,
_capture_sequence,
@@ -88,6 +89,9 @@ class FingerAcquisitionWorker(RecordSinkMixin):
self._on_reset_applied = on_reset_applied
self._thread: Optional[threading.Thread] = None
self._stop = threading.Event()
#: Set while the operator has paused: the camera stays open and the source keeps
#: being read, but no frame is processed and no packet is emitted.
self._pause = threading.Event()
self._lock = threading.Lock()
self._latest: Optional[SdkPacket] = None
self._session_id = 0
@@ -219,6 +223,9 @@ class FingerAcquisitionWorker(RecordSinkMixin):
"""
self._set_end_reason("stopped")
self._stop.set()
# A stopped worker is not paused: leaving the flag set would make the next session
# start silent.
self._pause.clear()
thread = self._thread
if wait and thread is not None and thread.is_alive():
thread.join(timeout=STOP_JOIN_TIMEOUT_S)
@@ -236,6 +243,24 @@ class FingerAcquisitionWorker(RecordSinkMixin):
self._latest = None
self._phase = "idle"
def pause(self) -> None:
"""Stop computing and publishing; the camera stays open.
Same contract as the box worker's: the source keeps being read so the device
is not released and the disconnect watchdog stays quiet, but no frame is
processed (no flow, no depth, no contact) and no packet is emitted, so the
display freezes on its last frame and the record channel sees a gap.
"""
self._pause.set()
def resume(self) -> None:
"""Start computing and publishing again, from the *newest* frame."""
self._pause.clear()
@property
def paused(self) -> bool:
return self._pause.is_set()
def reset(self) -> None:
"""Request a tracker reset, applied by the acquisition thread.
@@ -328,6 +353,9 @@ class FingerAcquisitionWorker(RecordSinkMixin):
self._on_status(message)
def _run(self, session: int, previous: Optional[threading.Thread] = None) -> None:
# A new session always starts unpaused: pausing is a statement about the session the
# operator is watching, not a sticky property of the device.
self._pause.clear()
try:
self._run_session(session, previous)
finally:
@@ -461,6 +489,17 @@ class FingerAcquisitionWorker(RecordSinkMixin):
time.sleep(0.005)
continue
if self._pause.is_set():
# Paused: the source read above proves the camera is alive, so refresh the
# liveness stamps the disconnect watchdog reads and take no frame. Skipping
# ``process_frame`` is what "pause the computation" means; publishing nothing
# is what "pause the display" means. ``read_source`` hands back the newest
# frame, so a resume starts from live data, not a backlog.
last_frame_mono = t0
saw_frame = True
time.sleep(PAUSE_POLL_S)
continue
unrolled = runtime.process_frame(source_frame, color=False)
if unrolled is None:
if saw_frame and is_file and none_streak_start is None:
@@ -52,6 +52,12 @@ FPS_WINDOW_S = 1.0
#: ``_high_resolution_timer``.
CAPTURE_REPEAT_POLL_S = 0.002
#: How long a paused loop waits before looking at the capture source again. Paused is not
#: stopped: the frame is still read (that is what keeps the device open and the disconnect
#: watchdog quiet) but nothing is computed, so this only has to be short enough that a resume
#: feels immediate.
PAUSE_POLL_S = 0.005
@contextlib.contextmanager
def _high_resolution_timer():
@@ -500,6 +506,9 @@ class SdkAcquisitionWorker(RecordSinkMixin):
self._on_reset_applied = on_reset_applied
self._thread: Optional[threading.Thread] = None
self._stop = threading.Event()
#: Set while the operator has paused: the device stays open and the capture keeps
#: publishing, but nothing is computed and no packet is emitted.
self._pause = threading.Event()
self._lock = threading.Lock()
self._latest: Optional[SdkPacket] = None
self._session_id = 0
@@ -651,6 +660,9 @@ class SdkAcquisitionWorker(RecordSinkMixin):
"""
self._set_end_reason("stopped")
self._stop.set()
# A stopped worker is not paused: leaving the flag set would make the next
# session start silent.
self._pause.clear()
thread = self._thread
if wait and thread is not None and thread.is_alive():
thread.join(timeout=STOP_JOIN_TIMEOUT_S)
@@ -669,6 +681,27 @@ class SdkAcquisitionWorker(RecordSinkMixin):
self._latest = None
self._phase = "idle"
def pause(self) -> None:
"""Stop computing and publishing; the device stays open.
This is what makes pause different from :meth:`stop`: the camera is not
released, so resuming is immediate and the device does not have to be
re-opened (and, on some boards, re-enumerated). While paused the loop keeps
draining the capture source — the disconnect watchdog must not fire on a
healthy camera just because an operator stopped looking at it — but it runs
no algorithm and emits no packet: no flow, no depth, no contact, and the
record channel sees a gap for the paused interval.
"""
self._pause.set()
def resume(self) -> None:
"""Start computing and publishing again, from the *newest* frame."""
self._pause.clear()
@property
def paused(self) -> bool:
return self._pause.is_set()
def reset(self) -> None:
"""Request a tracker reset, applied by the acquisition thread.
@@ -764,6 +797,9 @@ class SdkAcquisitionWorker(RecordSinkMixin):
self._on_status(message)
def _run(self, session: int, previous: Optional[threading.Thread] = None) -> None:
# A new session always starts unpaused: pausing is a statement about the
# session the operator is watching, not a sticky property of the device.
self._pause.clear()
try:
# A fine timer for the whole session: the loop waits for the next
# frame between frames, and on Windows a sub-tick wait otherwise
@@ -927,6 +963,18 @@ class SdkAcquisitionWorker(RecordSinkMixin):
time.sleep(0)
continue
last_delivered = delivered
if self._pause.is_set():
# Paused: the capture above already refreshed ``last_source_seq``, so the
# device is demonstrably alive — refresh the liveness stamps the disconnect
# watchdog reads, then take no frame. ``get_img()`` is the algorithm (flow,
# depth, contact), so skipping it is what "pause the computation" means; no
# packet is published either, which is what "pause the display" means. The
# capture source hands back only the newest frame, so resuming picks up live
# data rather than working through a backlog.
last_frame_mono = t0
saw_frame = True
time.sleep(PAUSE_POLL_S)
continue
processed = sensor.get_img()
if processed is None:
if saw_frame and is_file and none_streak_start is None:
@@ -1,7 +1,34 @@
"""Display-only force presentation: baseline, deadband, and formatting.
Never mutates recorded samples. Curves, statistics and CSV keep the measurement
values; value cards (and optional display copies) go through this layer.
Never mutates recorded samples. The session store, ``forces.csv`` and the metric
history keep the measurement; the value cards *and the curves* are drawn from
this layer, which removes the idle bias and holds a resting channel at exactly
zero.
The estimate is kept **per display unit**. A raw count and a newton reading of
the same axis are different scales, so one deadband in the wrong unit would
either hide a real contact or leave the noise outside it. Values arrive here
already expressed in the unit the operator is looking at (``force_plots.push``
passes the live triplet in its own unit), and the unit is part of the key.
The estimate is published in two stages, because waiting for the whole startup
window leaves the operator staring at the device's raw bias:
* **provisional** — as soon as a few quiet samples exist (see
:data:`DEFAULT_PROVISIONAL_SAMPLES`) an estimate is fitted from them and
*applied*; it is re-fitted on every further quiet sample. The cards and the
curves therefore read zero from the first tenths of a second of a session
instead of after ``baseline_seconds``.
* **frozen** — when the quiet window closes with quality ``ok`` the estimate is
frozen exactly as before and stops chasing the data.
A window that closes *without* a trustworthy estimate (contaminated, unstable,
insufficient samples) keeps the provisional estimate applied rather than
reverting to the raw bias: `quality` on :class:`ZeroEstimate` and
:attr:`ForcePresentation.display_quality` say where the numbers came from. A
window in which *no* quiet sample was ever seen has nothing to estimate from, so
nothing is invented — the measurement is displayed unchanged, which is how a
contact present at session start stays visible instead of being zeroed away.
"""
from __future__ import annotations
@@ -11,6 +38,27 @@ from typing import Optional, Sequence
import numpy as np
#: Floor under the estimated deadband, per display unit. 50 raw counts is a
#: meaningful noise floor for the uncalibrated device; 50 N would be a contact,
#: so newton mode defaults to no floor at all and relies on the estimated
#: percentile (see :func:`ForcePresentation.fit_zero`).
DEFAULT_MIN_DEADBAND_RAW = 50.0
DEFAULT_MIN_DEADBAND_N = 0.0
#: How many quiet samples are enough to publish a *provisional* estimate.
#: Three is the smallest window ``fit_zero`` can take a median and a residual
#: percentile from, and at the ~30 Hz force rate it is about a tenth of a
#: second: the operator sees the display policy working almost immediately
#: rather than watching the device's raw bias for the whole startup window.
#: Every further quiet sample re-fits the estimate, so three is a starting
#: point, not a decision — the frozen window still has the final say.
DEFAULT_PROVISIONAL_SAMPLES = 3
def normalise_unit(unit: Optional[str]) -> str:
"""``"N"`` or ``"raw"`` — the two units a force reading can be shown in."""
return "N" if str(unit or "").upper().startswith("N") else "raw"
@dataclass(frozen=True)
class ZeroEstimate:
@@ -34,72 +82,236 @@ class DisplayForce:
@dataclass
class ForcePresentation:
"""Stateful display policy for force value cards.
class _UnitEstimate:
"""What was learned about one display unit's idle state."""
Startup auto-baseline collects quiet samples, freezes offsets and deadbands,
then applies a symmetric hysteretic deadband only to the display copy.
collecting: bool = False
samples: list[tuple[float, float, float]] = field(default_factory=list)
sample_times: list[float] = field(default_factory=list)
t0: Optional[float] = None
#: Frozen estimate: set once the quiet window closes with quality ``ok``.
offsets: Optional[tuple[float, float, float]] = None
deadbands: Optional[tuple[float, float, float]] = None
estimate: Optional[ZeroEstimate] = None
#: Provisional estimate, published from the first few quiet samples and
#: re-fitted on each further one. It is what the display uses while the
#: window is still open, and what it keeps if the window closes badly.
provisional_offsets: Optional[tuple[float, float, float]] = None
provisional_deadbands: Optional[tuple[float, float, float]] = None
provisional_samples: int = 0
#: A quiet sample was refused because a contact was present or the flow was
#: too high: this window is not a trustworthy idle window any more.
contaminated: bool = False
@property
def ready(self) -> bool:
"""A *frozen* estimate: the startup window closed with quality ``ok``."""
return self.offsets is not None and self.deadbands is not None
@property
def display_offsets(self) -> Optional[tuple[float, float, float]]:
"""What the display is centred on: frozen if there is one, else provisional."""
return self.offsets if self.offsets is not None else self.provisional_offsets
@property
def display_deadbands(self) -> Optional[tuple[float, float, float]]:
"""The deadband the display is using (frozen if there is one)."""
return (
self.deadbands if self.deadbands is not None else self.provisional_deadbands
)
@property
def display_ready(self) -> bool:
"""Whether *some* estimate — provisional or frozen — is being applied."""
return (
self.display_offsets is not None and self.display_deadbands is not None
)
@dataclass
class ForcePresentation:
"""Stateful display policy for the force numbers *and* the force curves.
Startup auto-baseline collects quiet samples per display unit, freezes
offsets and deadbands, then applies a symmetric hysteretic deadband to the
display copy. It is enabled for every sensor type and both units: an idle
channel that reads a raw count or a newton offset at rest is a display
problem whichever device produced it.
"""
enabled: bool = True
unit: str = "raw"
baseline_seconds: float = 3.0
min_samples: int = 30
#: Quiet samples needed before a provisional estimate is applied (see
#: :data:`DEFAULT_PROVISIONAL_SAMPLES`).
provisional_samples: int = DEFAULT_PROVISIONAL_SAMPLES
max_flow_p95: float = 0.4
require_no_contact: bool = True
#: Residual percentile used as the per-axis deadband after offset removal.
deadband_percentile: float = 99.0
#: Extra margin on top of the percentile residual.
deadband_margin: float = 1.05
#: Floor so tiny MAD windows do not produce a zero deadband.
min_deadband: float = 50.0
#: Floor so tiny MAD windows do not produce a zero deadband (see above).
min_deadband_raw: float = DEFAULT_MIN_DEADBAND_RAW
min_deadband_n: float = DEFAULT_MIN_DEADBAND_N
hysteresis_ratio: float = 0.8
raw_significant_digits: int = 4
newton_decimals: int = 3
_collecting: bool = field(default=True, init=False, repr=False)
_samples: list[tuple[float, float, float]] = field(
default_factory=list, init=False, repr=False
)
_sample_times: list[float] = field(default_factory=list, init=False, repr=False)
_offsets: Optional[tuple[float, float, float]] = field(
default=None, init=False, repr=False
)
_deadbands: Optional[tuple[float, float, float]] = field(
default=None, init=False, repr=False
#: One idle estimate per display unit ("raw" / "N"), because the floors and
#: the values are in different scales.
_states: dict[str, _UnitEstimate] = field(
default_factory=dict, init=False, repr=False
)
_suppressed: tuple[bool, bool, bool] = field(
default=(False, False, False), init=False, repr=False
)
_last_estimate: Optional[ZeroEstimate] = field(
default=None, init=False, repr=False
)
_t0: Optional[float] = field(default=None, init=False, repr=False)
# ------------------------------------------------------------------ policy
def reset_session(self, *, unit: str = "raw") -> None:
"""Start a new idle estimate for a session."""
self.unit = "N" if str(unit).upper().startswith("N") else "raw"
self._collecting = self.enabled and self.unit == "raw"
self._samples = []
self._sample_times = []
self._offsets = None
self._deadbands = None
"""Start a new idle estimate for one display unit."""
self.unit = normalise_unit(unit)
state = self._state(self.unit)
state.collecting = bool(self.enabled)
state.samples = []
state.sample_times = []
state.t0 = None
state.offsets = None
state.deadbands = None
state.estimate = None
state.provisional_offsets = None
state.provisional_deadbands = None
state.provisional_samples = 0
state.contaminated = False
self._suppressed = (False, False, False)
self._last_estimate = None
self._t0 = None
def request_reestimate(self) -> None:
"""Operator-triggered restart of the idle window."""
self.reset_session(unit=self.unit)
def set_enabled(self, enabled: bool, *, unit: Optional[str] = None) -> None:
"""The master switch; switching it on restarts the quiet window."""
self.enabled = bool(enabled)
self.reset_session(unit=self.unit if unit is None else unit)
def min_deadband_for(self, unit: Optional[str] = None) -> float:
"""The floor under the estimated deadband, for one display unit."""
chosen = self.unit if unit is None else unit
return (
float(self.min_deadband_n)
if normalise_unit(chosen) == "N"
else float(self.min_deadband_raw)
)
def set_min_deadband(self, unit: str, floor: float) -> None:
"""Move one unit's floor, and re-floor the estimate already applied.
The floor is a lower bound, never a replacement: the samples collected so
far are re-fitted, so setting it back to 0 ("auto") restores the
estimated band instead of leaving the raised one behind. Both the frozen
estimate and the provisional one in use are re-fitted, because either can
be what the cards and the curves are reading.
"""
floor = max(0.0, float(floor))
if normalise_unit(unit) == "N":
self.min_deadband_n = floor
else:
self.min_deadband_raw = floor
state = self._states.get(normalise_unit(unit))
if state is None or not state.samples:
return # a window still being collected will use the new floor
estimate = self.fit_zero(state.samples, unit=unit)
if estimate.quality != "ok":
return
if state.collecting or not state.ready:
# Nothing frozen yet: this is the provisional estimate on screen.
state.provisional_offsets = estimate.offsets
state.provisional_deadbands = estimate.deadbands
state.provisional_samples = estimate.sample_count
return
state.offsets = estimate.offsets
state.deadbands = estimate.deadbands
state.estimate = estimate
@property
def ready(self) -> bool:
return self._offsets is not None and self._deadbands is not None
"""A frozen estimate for the active unit (the startup window closed)."""
state = self._states.get(self.unit)
return bool(state is not None and state.ready)
@property
def display_ready(self) -> bool:
"""Whether the display policy is being applied to the active unit.
True from the first few quiet samples (provisional) and after the frozen
window as well — this is what the cards and the curves ask, because both
are drawn from the same applied estimate.
"""
state = self._states.get(self.unit)
return bool(state is not None and state.display_ready)
@property
def display_quality(self) -> str:
"""Where the numbers on screen come from, for diagnostics and reporting.
``"off"`` — the aid is switched off; the measurement is shown.
``"provisional"`` — a provisional estimate from the first quiet samples
is applied; the startup window is still open.
``"ok"`` — the full quiet window closed and was frozen.
``"contaminated"`` / ``"unstable"`` / ``"insufficient"`` — the window
closed without a trustworthy estimate: the
provisional estimate stays applied (this is the
state that used to fall back to the raw bias).
``"none"`` — no quiet sample was ever seen, so nothing is
zeroed; a contact at session start is shown as it
is measured rather than hidden.
"""
if not self.enabled:
return "off"
state = self._states.get(self.unit)
if state is None or not state.display_ready:
return "none"
if state.ready:
return "ok"
if not state.collecting and state.estimate is not None:
return state.estimate.quality
return "provisional"
@property
def last_estimate(self) -> Optional[ZeroEstimate]:
return self._last_estimate
"""The window's outcome: the frozen estimate, the failed one, or the
provisional estimate while the window is still open."""
state = self._states.get(self.unit)
if state is None:
return None
if state.estimate is not None:
return state.estimate
if state.provisional_offsets is None or state.provisional_deadbands is None:
return None
return ZeroEstimate(
offsets=state.provisional_offsets,
deadbands=state.provisional_deadbands,
sample_count=state.provisional_samples,
quality="provisional",
message=f"provisional: {state.provisional_samples} quiet samples so far",
)
def _state(self, unit: str) -> _UnitEstimate:
"""The estimate for one unit, created on first use.
A unit that was never estimated — the operator switched mid-session —
starts its own quiet window instead of borrowing the other unit's, which
would be an offset in the wrong scale.
"""
key = normalise_unit(unit)
state = self._states.get(key)
if state is None:
state = _UnitEstimate(collecting=bool(self.enabled))
self._states[key] = state
return state
# --------------------------------------------------------------- estimator
def observe(
self,
values: Sequence[float],
@@ -108,51 +320,96 @@ class ForcePresentation:
valid: bool = True,
flow_p95: Optional[float] = None,
contact_found: bool = False,
unit: Optional[str] = None,
) -> Optional[ZeroEstimate]:
"""Feed one live sample into the startup estimator.
``values`` must already be expressed in ``unit`` (the displayed one).
Returns a :class:`ZeroEstimate` when the window closes (ok or failed).
"""
if not self._collecting or self.unit != "raw":
key = normalise_unit(self.unit if unit is None else unit)
state = self._state(key)
if not state.collecting:
return None
if self._t0 is None:
self._t0 = float(t_rel_s)
if state.t0 is None:
state.t0 = float(t_rel_s)
if not valid:
return None
if self.require_no_contact and contact_found:
# Contaminated window — keep collecting but mark later if needed.
# Contaminated window — the sample is not a quiet one, so it must not
# enter the estimate (that is how a press at session start stays
# visible). The window is marked so a close can report it honestly.
state.contaminated = True
return None
if flow_p95 is not None and float(flow_p95) > self.max_flow_p95:
state.contaminated = True
return None
sample = tuple(float(v) for v in values[:3])
if len(sample) < 3 or not all(np.isfinite(sample)):
return None
self._samples.append(sample) # type: ignore[arg-type]
self._sample_times.append(float(t_rel_s))
elapsed = float(t_rel_s) - float(self._t0)
if elapsed < self.baseline_seconds and len(self._samples) < self.min_samples * 2:
state.samples.append(sample) # type: ignore[arg-type]
state.sample_times.append(float(t_rel_s))
# Publish or refine the provisional estimate *before* the window is
# allowed to close, so the display policy is effective from the first
# samples and a later contamination cannot drop it back to the raw bias.
self._update_provisional(state, key)
elapsed = float(t_rel_s) - float(state.t0)
if elapsed < self.baseline_seconds and len(state.samples) < self.min_samples * 2:
return None
if len(self._samples) < self.min_samples:
if len(state.samples) < self.min_samples:
# The window closed before it was ever a full quiet window. It is a
# *failure* of the startup estimate, and the honest thing to do is
# say so — but the display keeps whatever the provisional estimate
# already worked out, because falling back to the raw bias is the
# defect this class exists to remove.
quality = "contaminated" if state.contaminated else "insufficient"
detail = (
"window contaminated (contact or flow) after"
if state.contaminated
else "only"
)
estimate = ZeroEstimate(
offsets=(0.0, 0.0, 0.0),
deadbands=(0.0, 0.0, 0.0),
sample_count=len(self._samples),
quality="insufficient",
message=f"only {len(self._samples)} quiet samples in {elapsed:.2f}s",
sample_count=len(state.samples),
quality=quality,
message=f"{detail} {len(state.samples)} quiet samples in {elapsed:.2f}s",
)
self._last_estimate = estimate
self._collecting = False
state.estimate = estimate
state.collecting = False
return estimate
estimate = self.fit_zero(self._samples)
self._last_estimate = estimate
self._collecting = False
estimate = self.fit_zero(state.samples, unit=key)
state.estimate = estimate
state.collecting = False
if estimate.quality == "ok":
self._offsets = estimate.offsets
self._deadbands = estimate.deadbands
state.offsets = estimate.offsets
state.deadbands = estimate.deadbands
return estimate
def fit_zero(self, samples: Sequence[Sequence[float]]) -> ZeroEstimate:
def _update_provisional(self, state: _UnitEstimate, unit: str) -> None:
"""Fit and apply the display estimate from the quiet samples so far.
Called on every accepted quiet sample. The estimate is only published
once there are enough samples to take a median (:data:`DEFAULT_PROVISIONAL_SAMPLES`);
from then on each sample sharpens it until the frozen window closes. A
sample set the fitter rejects (not enough finite rows, wildly unstable)
leaves the previous provisional estimate in place rather than dropping
the display back to the raw bias.
"""
if len(state.samples) < max(3, int(self.provisional_samples)):
return
estimate = self.fit_zero(state.samples, unit=unit)
if estimate.quality != "ok":
return
state.provisional_offsets = estimate.offsets
state.provisional_deadbands = estimate.deadbands
state.provisional_samples = estimate.sample_count
def fit_zero(
self, samples: Sequence[Sequence[float]], *, unit: Optional[str] = None
) -> ZeroEstimate:
"""Estimate offsets and deadbands from quiet samples (does not apply them)."""
floor = self.min_deadband_for(unit)
block = np.asarray(samples, dtype=np.float64)
if block.ndim != 2 or block.shape[1] < 3 or block.shape[0] < 3:
return ZeroEstimate(
@@ -178,7 +435,7 @@ class ForcePresentation:
med = float(np.median(finite))
residual = np.abs(finite - med)
p = float(np.percentile(residual, self.deadband_percentile))
db = max(self.min_deadband, p * self.deadband_margin)
db = max(floor, p * self.deadband_margin)
# Reject wildly unstable windows (p99 >> median magnitude scale).
mad = float(np.median(residual))
if mad > 0 and p > 20.0 * mad and p > 2000.0:
@@ -198,15 +455,19 @@ class ForcePresentation:
quality="ok",
)
def apply_estimate(self, estimate: ZeroEstimate) -> None:
def apply_estimate(
self, estimate: ZeroEstimate, *, unit: Optional[str] = None
) -> None:
"""Explicitly freeze an estimate (e.g. after user confirmation)."""
if estimate.quality != "ok":
return
self._offsets = estimate.offsets
self._deadbands = estimate.deadbands
self._collecting = False
self._last_estimate = estimate
state = self._state(self.unit if unit is None else unit)
state.offsets = estimate.offsets
state.deadbands = estimate.deadbands
state.collecting = False
state.estimate = estimate
# ------------------------------------------------------------------ display
def accept(
self,
values: Sequence[float],
@@ -214,19 +475,23 @@ class ForcePresentation:
unit: Optional[str] = None,
available: tuple[bool, bool, bool] = (True, True, True),
) -> DisplayForce:
"""Return a display copy; never mutates ``values``."""
unit = self.unit if unit is None else (
"N" if str(unit).upper().startswith("N") else "raw"
)
"""Return a display copy of one live sample; never mutates ``values``.
The deadband is hysteretic: an axis already held at zero stays there
until it leaves the full band, so a reading on the edge does not flicker
between a number and zero.
The offset it subtracts is the frozen estimate once the startup window
has closed cleanly, and the provisional one before (or instead of) that
— never "nothing", which is what used to leave the raw bias on the cards
for the first seconds of a session.
"""
unit = normalise_unit(self.unit if unit is None else unit)
raw = [float(v) for v in values[:3]]
while len(raw) < 3:
raw.append(float("nan"))
if (
not self.enabled
or unit != "raw"
or self._offsets is None
or self._deadbands is None
):
state = self._states.get(unit)
if not self.enabled or state is None or not state.display_ready:
return DisplayForce(
values=(raw[0], raw[1], raw[2]),
suppressed=(False, False, False),
@@ -242,8 +507,8 @@ class ForcePresentation:
out.append(raw[i])
suppressed.append(False)
continue
centered = raw[i] - self._offsets[i]
on = self._deadbands[i]
centered = raw[i] - state.display_offsets[i] # type: ignore[index]
on = state.display_deadbands[i] # type: ignore[index]
off = on * self.hysteresis_ratio
was = prev[i]
if was:
@@ -264,11 +529,48 @@ class ForcePresentation:
zero_zone=any(self._suppressed),
)
def present(
self, values, *, unit: Optional[str] = None
) -> tuple[np.ndarray, np.ndarray]:
"""Present a whole block of readings: ``(display copy, suppressed mask)``.
One row per sample, for the curves — the live buffer draws a rolling
window, not one sample at a time. Vectorised and *stateless*: the
hysteresis latch in :meth:`accept` belongs to the live sample stream, and
a curve redrawn from a window must not depend on which frame drew it.
Non-finite readings stay non-finite — an axis with no value in this unit,
or a failed readout, is a gap and never a zero.
It reads the same applied estimate as :meth:`accept` (provisional while
the startup window is open), so a card and its own curve can never
disagree about what the same sample is worth.
"""
unit = normalise_unit(self.unit if unit is None else unit)
block = np.array(values, dtype=np.float64, copy=True, ndmin=2)[:, :3]
if block.shape[1] < 3:
pad = np.full((block.shape[0], 3 - block.shape[1]), np.nan)
block = np.hstack([block, pad])
suppressed = np.zeros(block.shape, dtype=bool)
state = self._states.get(unit)
if not self.enabled or state is None or not state.display_ready:
return block, suppressed
offsets = state.display_offsets
deadbands = state.display_deadbands
for axis in range(3):
column = block[:, axis]
finite = np.isfinite(column)
if not finite.any():
continue
centered = column[finite] - offsets[axis] # type: ignore[index]
quiet = np.abs(centered) <= deadbands[axis] # type: ignore[index]
column[finite] = np.where(quiet, 0.0, centered)
suppressed[finite, axis] = quiet
return block, suppressed
def format_force(self, value: float, unit: Optional[str] = None) -> str:
"""Stable digit formatting for value cards."""
unit = self.unit if unit is None else (
"N" if str(unit).upper().startswith("N") else "raw"
)
unit = normalise_unit(self.unit if unit is None else unit)
if not np.isfinite(value):
return "—"
# Avoid signed zero after deadband.
+14 -1
View File
@@ -23,7 +23,11 @@ from orisys_app_shared.paths import preference_path, preferences_default_path
_SCENE3D_DISPLAY: dict[str, Any] = {
"deform_enabled": False,
"display_gain": 0.0,
"housing_visible": True,
#: Matches ``DisplaySettings.housing_visible`` and the shipped
#: ``config/preferences.default.json``: the frame starts hidden and the operator
#: turns it on for a mechanical check. A mismatch here would re-show it for a
#: user whose preferences file had been reset from these defaults.
"housing_visible": False,
"max_fps": 30,
}
@@ -40,13 +44,22 @@ _DEFAULTS: dict[str, Any] = {
"config_name": "",
"units": "newton",
"plot_window_sec": 4.0,
# FN is asymmetric (its lower bound may be negative); FX/FY take one absolute
# magnitude each, applied symmetrically. 0 on a raw bound means "auto range".
"plot_fn_min": 0.0,
"plot_fn_max": 20.0,
"plot_shear_max": 20.0,
# Raw (uncalibrated) HHD units. Sized so the idle jitter is sub-pixel on
# both raw-mode products while never clipping: measured peaks are ~447k
# (mini, 365x365) and ~41k (finger, 280x300); idle sigma is ~80 (finger).
"plot_fn_raw_min": 0.0,
"plot_fn_raw_max": 1000000.0,
"plot_shear_raw_max": 500000.0,
# Display deadband floors, per display unit. 0 = auto: the noise estimated
# from the quiet startup window (raw keeps its 50-count floor under it).
# Display only — the session store and forces.csv keep the measurement.
"plot_deadband_n": 0.0,
"plot_deadband_raw": 0.0,
"image_view_mode": "arrows_pure",
"arrow_rotation_deg": 90,
"show_fps": False,
@@ -48,6 +48,12 @@ LENGTH_UNIT_M = "m"
#: Weight sum tolerance for a :class:`SurfaceBinding` row.
WEIGHT_SUM_TOL = 1e-4
#: The dihedral angle above which a part's faces are a real edge and its vertices
#: must not share one averaged normal (the housing's 90° corners). The compiler
#: splits at this angle (``legacy.HOUSING_CREASE_ANGLE_DEG``); ``validate_assets``
#: uses it to warn about a package whose stored normals predate that split.
SHADING_CREASE_ANGLE_DEG = 30.0
#: The standard failure codes of §8.3, and how loud each one is.
DIAGNOSTIC_SEVERITY: Mapping[str, Severity] = {
"asset_auth_failed": "error",
@@ -65,6 +71,10 @@ DIAGNOSTIC_SEVERITY: Mapping[str, Severity] = {
#: predates the record, or the device cannot state it). Informational: the pane may
#: show the geometry, but it must not claim the measurement was checked against it.
"geometry_unverified": "info",
#: The geometry is right but its shading was compiled before crease-aware normals:
#: a housing's 90° corners carry one averaged normal each and band at the rim. A
#: warning, because the picture is usable and the repair is a recompile.
"geometry_shading_stale": "warning",
"stale_frame": "info",
#: The operator asked for one quantity and the frame could only be coloured with the
#: other one. Informational: the picture is still a measurement and names itself; what
@@ -590,7 +600,11 @@ class DisplaySettings:
display_gain: float = 0.0
deform_enabled: bool = False
color_scale: Optional[ColorScale] = None
housing_visible: bool = True
#: Non-sensing parts (the housing/frame) start hidden: the measurement is what the pane is
#: for, and on the finger sensor the frame is ~33% wider than the sensing surface, so it
#: dominated the picture. Operators who want it for a mechanical check turn it on in the 3D
#: display settings; a saved preference wins over this default.
housing_visible: bool = False
max_fps: int = 30
#: Which scalar field colours the surface: the chain's deformation response
#: (``"depth"``) or the tracker's optical-flow magnitude (``"flow"``). This is a
@@ -932,6 +946,68 @@ def validate_mesh_part(part: MeshPart) -> None:
raise error("geometry_mismatch", f"{part.part_id}: normals are not unit length")
def stale_housing_shading(assets: SceneAssets) -> tuple[Diagnostic, ...]:
"""Warnings about a package whose housing shading predates crease-aware normals.
A housing with 90° corners needs more than one vertex normal there: a single
averaged normal agrees with *none* of the faces meeting at the corner, which is
exactly the banding and streaking at the rim (measured on the shipped frame: up
to 80.5° off, and 130 of 726 vertices affected). A package compiled by an older
build carries one such normal per vertex.
This is deliberately **not** part of :func:`validate_assets`: the geometry is
correct, the picture is usable, and "every fresh package must produce no warnings"
is a property the package tests rely on. The check is offered here so a tool, a
release gate or a test can ask the question explicitly, and the repair is a
recompile (``compile_scene.py`` derives the crease-aware normals).
"""
out: list[Diagnostic] = []
for part in assets.parts:
if part.role != "housing":
continue
positions = np.asarray(part.positions_sensor_m, dtype=np.float64)
triangles = np.asarray(part.triangles, dtype=np.int64)
normals = np.asarray(part.normals_sensor, dtype=np.float64)
if normals.shape != positions.shape or not len(triangles):
continue
if not np.allclose(np.linalg.norm(normals, axis=1), 1.0, atol=1e-3):
out.append(
diagnostic(
"geometry_shading_stale",
f"{part.part_id}: the stored normals are not unit length",
part=part.part_id,
)
)
continue
a = positions[triangles[:, 0]]
b = positions[triangles[:, 1]]
c = positions[triangles[:, 2]]
face = np.cross(b - a, c - a)
unit = face / np.maximum(np.linalg.norm(face, axis=1, keepdims=True), 1e-30)
best = np.full(len(positions), -1.0)
for corner in range(3):
indices = triangles[:, corner]
np.maximum.at(best, indices, np.einsum("ij,ij->i", unit, normals[indices]))
used = best > -1.0
if not used.any():
continue
threshold = float(np.cos(np.radians(SHADING_CREASE_ANGLE_DEG)))
if bool(np.any(best[used] < threshold)):
# The worst agreement any vertex has with all of its faces: exactly the
# number that says how far the shading is from the geometry.
worst = float(np.degrees(np.arccos(np.clip(best[used].min(), -1.0, 1.0))))
out.append(
diagnostic(
"geometry_shading_stale",
f"{part.part_id}: a stored normal is {worst:.1f}° from every face "
"that touches its vertex, so its corners are smoothed into banding; "
"recompile the package with compile_scene.py for crease-aware normals",
part=part.part_id,
)
)
return tuple(out)
def validate_assets(
assets: SceneAssets, expected: Optional[ExpectedGeometry] = None
) -> tuple[Diagnostic, ...]:
+401 -48
View File
@@ -61,6 +61,23 @@ MESH_KEYS = ("vertices", "faces", "uv_data", "loop_to_vertex")
#: The mesh's material: the sensing surface is colour-mapped, never shaded into
#: looking like a different value.
SENSING_MATERIAL = Material((0.85, 0.85, 0.88, 1.0), "unlit_scalar", True)
#: A non-sensing part is never coloured by the measurement, so it is a plain diffuse grey.
#: The GL backend also keys its own base colours by part id (``"housing"`` → graphite), so
#: the id matters as much as the colour here.
HOUSING_MATERIAL = Material((0.32, 0.33, 0.35, 1.0), "diffuse")
HOUSING_PART_ID = "housing"
HOUSING_FILE = "housing.npz"
_HOUSING_KEYS = ("vertices", "faces")
#: The dihedral angle above which two faces are treated as a real edge and their
#: shared vertices are split. The Blender export writes no normals and does not
#: split vertices either (the housing is 726 vertices, 726 unique, a closed
#: 2-manifold), so its genuine 90° corners have exactly one vertex each; averaging
#: every incident face unconditionally turns each of those into a ~45° normal and
#: smears the rim over ~100 px. 30° is the usual shading-crease value: it keeps the
#: curved rim and the countersink smooth (their dihedral angles are a few degrees)
#: and splits only the corners (measured: 134 of 2172 edges exceed it, max 90°).
HOUSING_CREASE_ANGLE_DEG = 30.0
@dataclass(frozen=True)
@@ -261,23 +278,102 @@ def split_at_uv_seams(
)
def vertex_normals(positions: np.ndarray, triangles: np.ndarray) -> np.ndarray:
def _face_normals_and_areas(
positions: np.ndarray, triangles: np.ndarray
) -> tuple[np.ndarray, np.ndarray]:
"""Unnormalised face normals (length = twice the area) and their lengths."""
a = positions[triangles[:, 0]]
b = positions[triangles[:, 1]]
c = positions[triangles[:, 2]]
# Not normalised: the cross product's length is twice the triangle area.
face_normals = np.cross(b - a, c - a)
return face_normals, np.linalg.norm(face_normals, axis=1)
def smooth_groups_by_crease(
positions: np.ndarray,
triangles: np.ndarray,
*,
crease_angle_deg: float = HOUSING_CREASE_ANGLE_DEG,
) -> np.ndarray:
"""Which smoothing group every triangle belongs to, splitting at ``crease_angle_deg``.
Two triangles that share an edge are in the same group unless the angle between
their normals exceeds the threshold; groups are the connected components of that
relation. Every triangle is in exactly one group (index ``0..k-1``), so
:func:`vertex_normals` can average only the faces that are actually smooth with
each other.
"""
positions = np.asarray(positions, dtype=np.float64)
triangles = np.asarray(triangles, dtype=np.int64)
count = len(triangles)
if count == 0:
return np.zeros(0, dtype=np.int64)
face_normals, areas = _face_normals_and_areas(positions, triangles)
unit = face_normals / np.maximum(areas[:, None], 1e-30)
# Below cos(threshold) the two faces are creased and must not share a normal.
threshold = float(np.cos(np.radians(float(crease_angle_deg))))
parent = list(range(count))
def find(item: int) -> int:
root = item
while parent[root] != root:
root = parent[root]
while parent[item] != root: # path compression
parent[item], item = root, parent[item]
return root
def union(left: int, right: int) -> None:
left_root, right_root = find(left), find(right)
if left_root != right_root:
parent[max(left_root, right_root)] = min(left_root, right_root)
# Edge (smaller index, larger index) -> the triangle that first used it. A
# manifold mesh gives each edge exactly two triangles; a non-manifold edge
# (more than two) simply joins all of them, which is the conservative choice.
first_use: dict[tuple[int, int], int] = {}
for index in range(count):
tri = triangles[index]
for corner in range(3):
low, high = int(tri[corner]), int(tri[(corner + 1) % 3])
if low > high:
low, high = high, low
other = first_use.get((low, high))
if other is None:
first_use[(low, high)] = index
continue
if float(unit[other] @ unit[index]) >= threshold:
union(other, index)
roots = np.array([find(index) for index in range(count)], dtype=np.int64)
_, groups = np.unique(roots, return_inverse=True)
return groups.astype(np.int64)
def vertex_normals(
positions: np.ndarray,
triangles: np.ndarray,
*,
crease_angle_deg: Optional[float] = None,
) -> np.ndarray:
"""Area-weighted vertex normals: unit length, computed from the geometry.
Derived rather than stored, so a normal is never a second, disagreeing copy of
the shape. A vertex used by no triangle (or by a degenerate one only) gets a
finite fallback instead of a NaN.
Without a crease angle every incident face is averaged, which is right for a
smooth surface and wrong for a box: on a 90° corner it produces a ~45° normal
that no incident face agrees with. For a mesh with creases use
:func:`split_at_crease_edges` first, so that each side of the crease owns its own
vertex — one vertex cannot hold two normals.
"""
positions = np.asarray(positions, dtype=np.float64)
triangles = np.asarray(triangles, dtype=np.int64)
normals = np.zeros_like(positions)
if len(triangles):
a = positions[triangles[:, 0]]
b = positions[triangles[:, 1]]
c = positions[triangles[:, 2]]
# Not normalised: the cross product's length is twice the triangle area,
# which is exactly the weight a vertex normal wants.
face_normals = np.cross(b - a, c - a)
face_normals, _ = _face_normals_and_areas(positions, triangles)
for corner in range(3):
np.add.at(normals, triangles[:, corner], face_normals)
lengths = np.linalg.norm(normals, axis=1, keepdims=True)
@@ -290,6 +386,112 @@ def vertex_normals(positions: np.ndarray, triangles: np.ndarray) -> np.ndarray:
return normals / np.maximum(lengths, fallback)
def split_at_crease_edges(
positions: np.ndarray,
triangles: np.ndarray,
*,
crease_angle_deg: float = HOUSING_CREASE_ANGLE_DEG,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
"""Duplicate the vertices that sit on a crease, one copy per smoothing group.
Returns ``(positions, triangles, source_vertex)``: a mesh whose triangles are
unchanged in *number* but whose vertex indices address the copies, plus the map
from every new vertex back to the geometry vertex it came from (the analogue of
``source_vertex_of_render`` for the sensing surface). A vertex that is smooth all
the way around keeps exactly one copy, so the duplicated count is precisely the
number of extra normals the creases need — not a blanket unweld.
This is what makes a crease-aware normal *possible*: the averaging in
:func:`vertex_normals` is per index, so two sides of a 90° edge have to be two
indices. Every triangle is then smooth with every face it shares a vertex with,
and the averaged normal agrees with each of them.
"""
source_positions = np.asarray(positions, dtype=np.float64)
source_triangles = np.asarray(triangles, dtype=np.int64)
if len(source_triangles) == 0:
return (
source_positions.copy(),
source_triangles.astype(np.int64).copy(),
np.arange(len(source_positions), dtype=np.int64),
)
groups = smooth_groups_by_crease(
source_positions, source_triangles, crease_angle_deg=crease_angle_deg
)
# One render vertex per (geometry vertex, smoothing group) actually used.
keys = np.concatenate(
[
source_triangles[:, corner] * (int(groups.max()) + 1) + groups
for corner in range(3)
]
)
unique_keys, inverse = np.unique(keys, return_inverse=True)
group_count = int(groups.max()) + 1
source_vertex = (unique_keys // group_count).astype(np.int64)
new_triangles = inverse.reshape(3, len(source_triangles)).T.astype(np.int64)
return source_positions[source_vertex].copy(), new_triangles, source_vertex
def crease_aware_mesh(
positions: np.ndarray,
triangles: np.ndarray,
*,
crease_angle_deg: float = HOUSING_CREASE_ANGLE_DEG,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
"""A housing part's geometry and normals, split at its creases.
Returns ``(positions, triangles, normals)`` ready for a :class:`MeshPart`: the
positions may carry more vertices than the source mesh (one per side of each
crease), the triangle list is identical in count, and every normal is the
area-weighted average of the faces that are actually smooth with its vertex.
"""
split_positions, split_triangles, _ = split_at_crease_edges(
positions, triangles, crease_angle_deg=crease_angle_deg
)
normals = vertex_normals(split_positions, split_triangles)
return split_positions, split_triangles, normals
def normals_disagree_with_faces(
positions: np.ndarray,
triangles: np.ndarray,
normals: np.ndarray,
*,
crease_angle_deg: float = HOUSING_CREASE_ANGLE_DEG,
) -> bool:
"""Whether a mesh's stored normals are inconsistent with its own faces.
True when any vertex normal is more than ``crease_angle_deg`` from every face it
touches: at a crease, a single averaged normal cannot agree with both sides, and
that disagreement is exactly what bands in the render. Used as a *detector*, so a
loader repairs a package built by an older compiler without touching one that was
built correctly — a distinction that keeps a package round trip exact.
"""
positions = np.asarray(positions, dtype=np.float64)
triangles = np.asarray(triangles, dtype=np.int64)
normals = np.asarray(normals, dtype=np.float64)
if normals.shape != positions.shape or not len(triangles):
return True
lengths = np.linalg.norm(normals, axis=1)
if not np.allclose(lengths, 1.0, atol=1e-3):
return True
face_normals, areas = _face_normals_and_areas(positions, triangles)
unit = face_normals / np.maximum(areas[:, None], 1e-30)
threshold = float(np.cos(np.radians(float(crease_angle_deg))))
# The best agreement each vertex has with any incident face.
best = np.full(len(positions), -1.0)
for corner in range(3):
indices = triangles[:, corner]
agreement = np.einsum("ij,ij->i", unit, normals[indices])
np.maximum.at(best, indices, agreement)
used = best > -1.0
if not used.any():
return True
return bool(np.any(best[used] < threshold))
def _apply_rigid(
positions: np.ndarray, transform_source_from_sensor: np.ndarray
) -> np.ndarray:
@@ -471,6 +673,7 @@ class LegacyFingerImporter:
asset_version: str = "1.0.0",
lod_id: str = "lod0",
camera_presets: Optional[Iterable[CameraPreset]] = None,
housing_path: Optional[Any] = None,
) -> SceneAssets:
export_dir = Path(export_dir)
map_path = Path(map_path)
@@ -539,16 +742,41 @@ class LegacyFingerImporter:
).astype(np.float32),
dtype=np.float32,
)
housing_source = (
Path(housing_path) if housing_path is not None else export_dir / HOUSING_FILE
)
if housing_path is not None and not housing_source.is_file():
# An explicitly requested housing that is not there is a mistake worth stopping
# for: silently shipping a package without it looks exactly like success.
raise error(
"geometry_mismatch",
f"a housing was requested but {housing_source} does not exist",
)
housing = self._load_housing(
housing_source,
units=units,
transform=transform,
lod_id=lod_id,
)
# The housing is deliberately *not* part of ``geometry_id`` or
# ``source_geometry_id``: those are the identity of the sensing surface, and the
# acquisition side has to be able to reproduce them from the device's own files.
# Folding a display-only frame into them would make every device look unverified.
# It does belong in the asset hash, which names the package as a whole.
hash_arrays: dict[str, Any] = {
"positions": sensing.positions_sensor_m,
"triangles": sensing.triangles,
"indices": binding.sample_indices,
"weights": binding.sample_weights,
}
if housing is not None:
hash_arrays["housing_positions"] = housing.positions_sensor_m
hash_arrays["housing_triangles"] = housing.triangles
assets = SceneAssets(
asset_id=str(asset_id),
asset_version=str(asset_version),
asset_hash=content_digest(
{
"positions": sensing.positions_sensor_m,
"triangles": sensing.triangles,
"indices": binding.sample_indices,
"weights": binding.sample_weights,
},
hash_arrays,
{
"geometry_id": geometry_value,
"binding_id": binding_id(binding),
@@ -557,9 +785,16 @@ class LegacyFingerImporter:
},
),
geometry_id=geometry_value,
parts=(sensing,),
parts=(sensing,) if housing is None else (sensing, housing),
binding=binding,
camera_presets=tuple(camera_presets or default_finger_presets(bounds)),
camera_presets=tuple(
camera_presets
or default_finger_presets(
bounds,
positions_sensor_m=positions_sensor,
triangles=triangles,
)
),
bounds_sensor_m=bounds,
field_domain_id=domain.domain_id,
shape_hw=domain.shape_hw,
@@ -587,6 +822,71 @@ class LegacyFingerImporter:
)
return assets
@staticmethod
def _load_housing(
path: Path,
*,
units: float,
transform: np.ndarray,
lod_id: str,
) -> Optional[MeshPart]:
"""The optional non-sensing part (the housing / mounting frame), or ``None``.
The file is expected in the **same source frame and units as** ``mesh.npz``: an
export that lives in another frame is aligned once, by
``tools/scene_assets/align_housing.py``, rather than guessed here — the design
export of the finger sensor is in a camera/world frame whose relationship to the
sensor frame is a fixed similarity, and getting that wrong would put the frame in
the wrong place while still looking plausible.
A housing carries no UV mapping (it is never coloured by the measurement), so only
vertices and faces are read, and it gets no surface binding.
"""
if not path.is_file():
return None
data = _open_npz(path, what="housing")
_require(data, _HOUSING_KEYS, what=str(path))
vertices = np.asarray(data["vertices"])
faces = np.asarray(data["faces"])
if vertices.ndim != 2 or vertices.shape[1] != 3 or not np.isfinite(vertices).all():
raise error(
"geometry_mismatch",
f"{path.name}: vertices must be a finite [N,3] array, got {vertices.shape}",
)
if faces.ndim != 2 or faces.shape[1] != 3 or faces.size == 0:
raise error(
"geometry_mismatch",
f"{path.name}: faces must be a non-empty [M,3] array, got {faces.shape}",
)
if int(faces.max()) >= len(vertices) or int(faces.min()) < 0:
raise error(
"geometry_mismatch",
f"{path.name}: a triangle index is out of range for {len(vertices)} vertices",
)
# Same order as the sensing surface: units first, then the rigid transform.
positions_m = vertices.astype(np.float64) * units
triangles = faces.astype(np.int64)
# Crease-aware, not a plain average: the housing is a box, and averaging its
# 90° corners produces ~45° normals that band and streak at the rim. The
# creases are split into separate vertices first, because one vertex cannot
# hold the two normals a 90° corner needs. The triangle *count* is unchanged.
positions_m, triangles, normals = crease_aware_mesh(positions_m, triangles)
triangles = triangles.astype(np.uint32)
positions_sensor = _apply_rigid(positions_m, transform)
if not np.isfinite(positions_sensor).all():
raise error(
"geometry_mismatch", f"{path.name}: the transform produced non-finite positions"
)
return MeshPart(
part_id=HOUSING_PART_ID,
role="housing",
positions_sensor_m=freeze(positions_sensor, dtype=np.float32),
normals_sensor=freeze(normals @ transform[:3, :3].T, dtype=np.float32),
triangles=freeze(triangles, dtype=np.uint32),
material=HOUSING_MATERIAL,
lod_id=lod_id,
)
def validate(
self, assets: SceneAssets, expected: ExpectedGeometry
) -> tuple[Any, ...]:
@@ -595,46 +895,99 @@ class LegacyFingerImporter:
return validate_assets(assets, expected)
def default_finger_presets(bounds_sensor_m: np.ndarray) -> tuple[CameraPreset, ...]:
def sensing_surface_normal(
positions_sensor_m: np.ndarray, triangles: np.ndarray
) -> np.ndarray:
"""The sensing surface's outward mean face normal, in the sensor frame.
Area-weighted over the triangles (so a dense region does not outvote a large
face) and oriented away from the mesh's own centroid. This is the direction a
camera has to sit on: the sensor's *own* geometry says which side is the sensing
side, and a preset that assumes an axis instead is wrong the moment the export's
frame rotates (measured: the shipped ``default`` preset sits 122.7° from this
normal, i.e. behind the surface).
"""
positions = np.asarray(positions_sensor_m, dtype=np.float64)
triangles = np.asarray(triangles, dtype=np.int64)
face_normals, areas = _face_normals_and_areas(positions, triangles)
mean = face_normals.sum(axis=0)
norm = float(np.linalg.norm(mean))
if norm <= 1e-15:
return np.array([0.0, 0.0, 1.0])
normal = mean / norm
# The exports wind outward (verified on the shipped housing and mesh), but a
# reversed export must not silently aim the camera at the back of the surface.
centroid = positions[triangles.reshape(-1)].mean(axis=0) if len(triangles) else positions.mean(axis=0)
if float(normal @ (centroid - positions.mean(axis=0))) < 0.0:
normal = -normal
return normal
def default_finger_presets(
bounds_sensor_m: np.ndarray,
positions_sensor_m: Optional[np.ndarray] = None,
triangles: Optional[np.ndarray] = None,
*,
surface_normal_sensor: Optional[np.ndarray] = None,
) -> tuple[CameraPreset, ...]:
"""The default views for a flat sensing surface, in sensor metres.
Derived from the bounds rather than hard-coded: the surface faces +Z after the
transform, and a three-quarter view is the one that shows both its shape and
its relief.
Derived from the bounds and from the surface's **own** normal. The absolute axes
are not the sensor's: the design export's frame is rotated relative to the SDK
asset's, so a preset hard-coded to look down ``+Z`` ends up behind the surface
(~123° from its normal), where 96.5% of it is hidden by the frame when the
housing is shown. The three-quarter view therefore comes from the mesh: the eye
is placed along the outward normal, raised off-axis so the relief is visible.
"""
low, high = np.asarray(bounds_sensor_m, dtype=np.float64)
center = (low + high) * 0.5
centre = (low + high) * 0.5
span = float(np.max(high - low))
distance = max(span, 1e-4)
normal = surface_normal_sensor
if normal is None:
if positions_sensor_m is None or triangles is None:
normal = np.array([0.0, 0.0, 1.0])
else:
normal = sensing_surface_normal(positions_sensor_m, triangles)
normal = np.asarray(normal, dtype=np.float64)
normal = normal / max(float(np.linalg.norm(normal)), 1e-15)
# In-plane basis for this surface: the principal axes of the vertex cloud with
# the normal removed. Used for the ``up`` of the off-axis views so the picture is
# not rolled, and for the off-axis direction of the three-quarter view.
reference_cloud = (
np.asarray(positions_sensor_m, dtype=np.float64)
if positions_sensor_m is not None
else np.array([low, high], dtype=np.float64)
)
spread = reference_cloud - reference_cloud.mean(axis=0)
spread = spread - np.outer(spread @ normal, normal)
if float(np.linalg.norm(spread)) > 1e-15:
_, _, vectors = np.linalg.svd(spread, full_matrices=False)
tangent = vectors[0] - normal * float(vectors[0] @ normal)
tangent = tangent / max(float(np.linalg.norm(tangent)), 1e-15)
else: # a degenerate cloud (a single point): any perpendicular will do
fallback = np.array([0.0, 0.0, 1.0])
if abs(float(normal @ fallback)) > 0.9:
fallback = np.array([0.0, 1.0, 0.0])
tangent = np.cross(normal, fallback)
tangent = tangent / max(float(np.linalg.norm(tangent)), 1e-15)
bitangent = np.cross(normal, tangent)
def eye(offset: np.ndarray) -> tuple[float, float, float]:
return tuple(centre + offset * distance)
return (
CameraPreset(
"default",
tuple(center + np.array([-0.6, -0.6, 1.1]) * distance),
tuple(center),
(0.0, 0.0, 1.0),
0.6,
),
CameraPreset(
"front",
tuple(center + np.array([0.0, 0.0, 1.4]) * distance),
tuple(center),
(0.0, 1.0, 0.0),
0.6,
),
CameraPreset(
"side",
tuple(center + np.array([1.4, 0.0, 0.0]) * distance),
tuple(center),
(0.0, 0.0, 1.0),
0.6,
),
CameraPreset(
"top",
tuple(center + np.array([0.0, 1.4, 0.0]) * distance),
tuple(center),
(0.0, 0.0, 1.0),
0.6,
),
# Three-quarter: 31.0° off the normal, so the surface is seen almost head-on
# with enough tilt to show its relief. (The old fixed vector put this 122.7°
# away — behind the surface — for the design frame.)
CameraPreset("default", eye((1.0 * normal + 0.6 * tangent) / np.sqrt(1.36)),
tuple(centre), tuple(normal), 0.6),
# Straight down the outward normal.
CameraPreset("front", eye(1.4 * normal), tuple(centre),
tuple(tangent), 0.6),
# Grazing, from each in-plane direction: the profile of the surface.
CameraPreset("side", eye(1.4 * tangent), tuple(centre), tuple(normal), 0.6),
CameraPreset("top", eye(1.4 * bitangent), tuple(centre), tuple(normal), 0.6),
)
+83 -17
View File
@@ -64,6 +64,12 @@ from orisys_app_shared.scene3d.digest import (
content_digest,
geometry_id as compute_geometry_id,
)
from orisys_app_shared.scene3d.legacy import (
crease_aware_mesh,
default_finger_presets,
normals_disagree_with_faces,
sensing_surface_normal,
)
from orisys_app_shared.scene3d import protect as _protect
#: Re-exported so a caller can ask "can this build seal a package at all?" without
@@ -652,13 +658,26 @@ def scene_from_contents(
positions = freeze(arrays[positions_key], dtype=np.float32)
if not np.isfinite(positions).all():
raise error("geometry_mismatch", f"{part_id} has non-finite positions")
triangles = freeze(arrays[f"{prefix}/triangles.npy"], dtype=np.uint32)
if prefix == "sensing":
# The sensing surface is genuinely smooth, so its stored normals stand.
normals = freeze(arrays["sensing/normals.npy"], dtype=np.float32)
else:
# A housing's normals are compiled crease-aware (a box has 90° corners, and
# a plain per-vertex average rounds them into ~45° normals that band at the
# rim). Nothing is rewritten here: rebuilding a part's vertices on load
# would break the loader's contract that a package round-trips into the
# scene it was written from. The staleness of an older package is instead a
# warning from ``validate_assets``, and the repair is a recompile
# (``compile_scene.py`` derives the crease-aware normals).
normals = freeze(arrays["housing/normals.npy"], dtype=np.float32)
parts.append(
MeshPart(
part_id=part_id,
role=role, # type: ignore[arg-type]
positions_sensor_m=positions,
normals_sensor=freeze(arrays[f"{prefix}/normals.npy"], dtype=np.float32),
triangles=freeze(arrays[f"{prefix}/triangles.npy"], dtype=np.uint32),
normals_sensor=normals,
triangles=triangles,
material=(
Material((0.85, 0.85, 0.88, 1.0), "unlit_scalar", True)
if prefix == "sensing"
@@ -715,6 +734,48 @@ def scene_from_contents(
)
compatible = dict(manifest.get("compatible") or {})
source_geometry = dict(manifest.get("source_geometry") or {})
bounds = freeze(
np.asarray(manifest.get("bounds_sensor_m") or np.zeros((2, 3)), dtype=np.float32),
dtype=np.float32,
)
presets = tuple(
CameraPreset(
name=str(item.get("name")),
eye_sensor_m=tuple(float(v) for v in item.get("eye", ())),
target_sensor_m=tuple(float(v) for v in item.get("target", ())),
up_sensor=tuple(float(v) for v in item.get("up", ())),
vertical_fov_rad=float(item.get("fov", 0.6)),
)
for item in manifest.get("camera_presets", [])
if isinstance(item, dict) and item.get("name")
)
# A preset baked by an older compiler points down a fixed axis, which is only
# correct if the export's frame happens to match the sensor frame. The design
# export does not: its shipped ``default`` sits ~123° from the sensing surface's
# own normal, where the housing hides ~100% of the surface. If the package's
# default preset is on the far side of the surface, replace the set with one
# derived from the mesh rather than presenting a view of the back of the sensor.
sensing_only_for_presets = tuple(part for part in parts if part.role == "sensing")
if sensing_only_for_presets:
sensing_part = sensing_only_for_presets[0]
surface_normal = sensing_surface_normal(
np.asarray(sensing_part.positions_sensor_m, dtype=np.float64),
np.asarray(sensing_part.triangles, dtype=np.int64),
)
baked_default = next(
(preset for preset in presets if preset.name == "default"), None
)
if baked_default is None or _preset_is_behind_the_surface(
baked_default, surface_normal
):
presets = default_finger_presets(
bounds,
positions_sensor_m=np.asarray(
sensing_part.positions_sensor_m, dtype=np.float64
),
triangles=np.asarray(sensing_part.triangles, dtype=np.int64),
surface_normal_sensor=surface_normal,
)
assets = SceneAssets(
asset_id=str(manifest.get("asset_id") or ""),
asset_version=str(manifest.get("asset_version") or ""),
@@ -722,21 +783,8 @@ def scene_from_contents(
geometry_id=str(manifest.get("geometry_id") or ""),
parts=tuple(parts),
binding=binding,
camera_presets=tuple(
CameraPreset(
name=str(item.get("name")),
eye_sensor_m=tuple(float(v) for v in item.get("eye", ())),
target_sensor_m=tuple(float(v) for v in item.get("target", ())),
up_sensor=tuple(float(v) for v in item.get("up", ())),
vertical_fov_rad=float(item.get("fov", 0.6)),
)
for item in manifest.get("camera_presets", [])
if isinstance(item, dict) and item.get("name")
),
bounds_sensor_m=freeze(
np.asarray(manifest.get("bounds_sensor_m") or np.zeros((2, 3)), dtype=np.float32),
dtype=np.float32,
),
camera_presets=presets,
bounds_sensor_m=bounds,
field_domain_id=str(domain.get("id") or ""),
shape_hw=(shape[0], shape[1]),
source_vertex_of_render=(
@@ -754,6 +802,24 @@ def scene_from_contents(
return assets
def _preset_is_behind_the_surface(preset: CameraPreset, surface_normal: np.ndarray) -> bool:
"""Whether a preset's eye sits on the non-sensing side of the surface.
The test is the angle between the eye direction (from the target) and the
surface's outward normal: more than 90° means the camera is behind the plate and
is therefore looking at its back, through whatever frame surrounds it.
"""
eye = np.asarray(preset.eye_sensor_m, dtype=np.float64)
target = np.asarray(preset.target_sensor_m, dtype=np.float64)
offset = eye - target
norm = float(np.linalg.norm(offset))
if norm <= 0.0:
return True
normal = np.asarray(surface_normal, dtype=np.float64)
normal = normal / max(float(np.linalg.norm(normal)), 1e-15)
return float((offset / norm) @ normal) < 0.0
class PackagedAssetLoader:
"""The loader: plain, signed and encrypted packages through one entry point."""
@@ -83,6 +83,9 @@ def _housing() -> MeshPart:
(0, 3, 7), (0, 7, 4), # left
(1, 5, 6), (1, 6, 2), # right
]
# Averaged from the incident faces, winding and all. Deliberately unchanged: the
# POC is a stand-in for a *smooth* part, and its pixels are pinned by the GL smoke
# tests. The real housing's crease handling is tested against the real asset.
normals = np.zeros_like(corners)
for triangle in faces:
a, b, c = (corners[index] for index in triangle)
@@ -6,7 +6,19 @@ from pathlib import Path
from orisys_app_shared.paths import sensors_dir
PRODUCT_TYPES = ("box", "mini", "finger")
#: Product types the UI offers.
#:
#: ``finger-mems`` is a deprecated product line that is deliberately **not** part of the SDK
#: (there is no MEMS read path there), so it is carried studio-side only: its bundle copies
#: ``box``'s ``sensor.json`` without a ``force_calibration.json``, which means N units are
#: unavailable and the app reports raw for it — the honest answer for an uncalibrated device.
#: It runs on the box/mini pipeline, because the worker routing is a binary ``== "finger"``
#: test (see ``session_controller.worker_for``).
#:
#: NOTE: the copied ``camera``/``stitcher`` numbers belong to the box prototype. If the real
#: MEMS-finger device stitches differently, live frames are refused with
#: ``field_domain_mismatch`` — correct them from the device, not by guessing.
PRODUCT_TYPES = ("box", "mini", "finger", "finger-mems")
def list_sensor_types() -> list[str]:
+311 -3
View File
@@ -1,4 +1,4 @@
"""Display-only force baseline / deadband (value cards)."""
"""Display-only force baseline / deadband (value cards and curves)."""
from __future__ import annotations
@@ -6,6 +6,11 @@ import numpy as np
from orisys_app_shared.force.presentation import ForcePresentation, ZeroEstimate
#: The idle triplet from the customer's screenshot at run start: the numbers the
#: cards were showing because the estimate was not applied yet.
REPORTED_RAW_BIAS = (1118.0, 34.16, -186.1)
REPORTED_NEWTON_BIAS = (0.4, -0.25, 0.1)
def test_fit_zero_rejects_insufficient_samples() -> None:
presentation = ForcePresentation(enabled=True, min_samples=30)
@@ -22,7 +27,7 @@ def test_fit_zero_and_accept_deadband() -> None:
enabled=True,
min_samples=30,
baseline_seconds=0.0,
min_deadband=50.0,
min_deadband_raw=50.0,
deadband_percentile=99.0,
)
estimate = presentation.fit_zero(samples)
@@ -60,7 +65,7 @@ def test_observe_freezes_after_quiet_window() -> None:
enabled=True,
min_samples=10,
baseline_seconds=0.5,
min_deadband=10.0,
min_deadband_raw=10.0,
)
estimate: ZeroEstimate | None = None
for i in range(40):
@@ -80,8 +85,311 @@ def test_observe_freezes_after_quiet_window() -> None:
assert later is None
def test_the_estimate_is_kept_per_display_unit() -> None:
"""A newton baseline and a raw baseline are different scales, not one number."""
presentation = ForcePresentation(enabled=True, min_samples=5, baseline_seconds=1.0)
for i in range(12):
presentation.observe(
(1000.0 + i * 0.1, 0.0, 0.0), t_rel_s=i * 0.02, unit="raw"
)
assert presentation.unit == "raw"
assert presentation.ready
assert presentation.accept((1000.55, 0.0, 0.0), unit="raw").values[0] == 0.0
# The other unit has no estimate of its own and must show what was measured.
assert presentation.accept((1000.55, 0.0, 0.0), unit="N").values == (
1000.55,
0.0,
0.0,
)
presentation.unit = "N"
assert not presentation.ready, "the raw estimate made newtons ready"
presentation.unit = "raw"
# ...and it collects its own window, in its own scale.
for i in range(12):
presentation.observe((0.5 + i * 0.01, 0.0, 0.0), t_rel_s=i * 0.02, unit="N")
presentation.unit = "N"
assert presentation.ready
assert presentation.accept((0.555, 0.0, 0.0), unit="N").values[0] == 0.0
def test_the_newton_floor_is_not_the_raw_floor() -> None:
"""50 raw counts is a noise floor; 50 N would be a contact."""
presentation = ForcePresentation(enabled=True)
assert presentation.min_deadband_for("raw") == 50.0
assert presentation.min_deadband_for("N") == 0.0
rng = np.random.default_rng(3)
quiet = (0.5 + rng.normal(0.0, 0.002, size=(60, 3))).tolist()
estimate = presentation.fit_zero(quiet, unit="N")
assert estimate.quality == "ok"
assert max(estimate.deadbands) < 0.01, estimate.deadbands
def test_present_holds_a_resting_block_at_zero_and_keeps_an_excursion() -> None:
rng = np.random.default_rng(7)
bias = np.array([9555.0, -13368.0, -353.0])
quiet = (bias + rng.normal(0.0, 20.0, size=(80, 3))).tolist()
presentation = ForcePresentation(
enabled=True, min_samples=30, baseline_seconds=0.0, deadband_percentile=99.0
)
estimate = presentation.fit_zero(quiet)
assert estimate.quality == "ok"
presentation.apply_estimate(estimate)
block = np.asarray(
[bias, bias + 5.0, bias + 4000.0, [np.nan, np.nan, np.nan]], dtype=float
)
displayed, suppressed = presentation.present(block, unit="raw")
assert displayed[0].tolist() == [0.0, 0.0, 0.0], "a resting sample drifted"
assert suppressed[0].all()
assert displayed[1].tolist() == [0.0, 0.0, 0.0], "noise outside the deadband?"
# A real excursion survives, centred on the estimated offset.
assert 3900.0 < displayed[2][0] < 4100.0, displayed[2]
assert not suppressed[2].any()
# A gap is a gap: an unavailable axis must never become a zero.
assert np.isnan(displayed[3]).all()
assert not suppressed[3].any()
def test_present_is_a_display_copy_and_never_touches_the_input() -> None:
presentation = ForcePresentation(enabled=True, min_samples=5, baseline_seconds=1.0)
for i in range(12):
presentation.observe((100.0 + i * 0.01, 0.0, 0.0), t_rel_s=i * 0.02)
source = np.array([[100.055, 0.0, 0.0]])
displayed, _suppressed = presentation.present(source, unit="raw")
assert source.tolist() == [[100.055, 0.0, 0.0]], "present mutated the measurement"
assert displayed[0][0] == 0.0
def test_the_operator_floor_moves_the_frozen_deadband() -> None:
presentation = ForcePresentation(enabled=True, min_samples=5, baseline_seconds=1.0)
for i in range(12):
presentation.observe((100.0 + i * 0.01, 0.0, 0.0), t_rel_s=i * 0.02, unit="raw")
assert presentation.ready
presentation.set_min_deadband("raw", 500.0)
assert presentation.accept((300.0, 0.0, 0.0), unit="raw").values[0] == 0.0, (
"200 raw counts above the device's own band but inside the operator's floor"
)
# Back to auto: the floor is a floor, so the estimated band returns.
presentation.set_min_deadband("raw", 0.0)
assert presentation.min_deadband_for("raw") == 0.0
assert presentation.accept((300.0, 0.0, 0.0), unit="raw").values[0] != 0.0
def test_format_force_stable_digits() -> None:
presentation = ForcePresentation(enabled=True)
assert presentation.format_force(0.0, unit="raw") == "0"
assert presentation.format_force(1262.4, unit="raw") == "1262"
assert presentation.format_force(1.2345, unit="N") == "1.234"
# ------------------------------------------------------- the reported defect
def _quiet_until_zeroed(
presentation: ForcePresentation,
bias: tuple[float, float, float],
*,
unit: str,
limit: int = 29,
) -> int:
"""Feed quiet samples until ``accept`` reads zero; return the sample count.
Returns ``limit + 1`` (i.e. "never within the limit") if it did not zero.
"""
for index in range(limit):
presentation.observe(bias, t_rel_s=index / 30.0, flow_p95=0.1, unit=unit)
if presentation.accept(bias, unit=unit).values == (0.0, 0.0, 0.0):
return index + 1
return limit + 1
def test_a_resting_raw_channel_reads_zero_from_the_first_samples() -> None:
"""FN 1118 / FX 34.16 / FY -186.1 must be zero long before 30 samples / 3 s.
This is the reported defect: the cards kept the device's raw bias for the
whole startup window, so a session that started at rest looked like it was
under load.
"""
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
zeroed_at = _quiet_until_zeroed(presentation, REPORTED_RAW_BIAS, unit="raw")
assert zeroed_at <= 3, f"the raw bias was still on the cards after {zeroed_at} samples"
assert not presentation.ready, "zeroing should not have needed the whole quiet window"
assert presentation.display_ready, "the display policy was not applied while collecting"
assert presentation.display_quality == "provisional"
# ...and no invented zero: the offset is the one the quiet samples measured.
offsets = presentation.last_estimate.offsets
assert offsets[0] == REPORTED_RAW_BIAS[0]
assert offsets[1] == REPORTED_RAW_BIAS[1]
assert offsets[2] == REPORTED_RAW_BIAS[2]
def test_a_resting_newton_channel_reads_zero_from_the_first_samples() -> None:
"""The same aid in newtons — the estimate is per display unit."""
presentation = ForcePresentation(
enabled=True, unit="N", min_samples=30, baseline_seconds=3.0
)
zeroed_at = _quiet_until_zeroed(presentation, REPORTED_NEWTON_BIAS, unit="N")
assert zeroed_at <= 3, f"the newton bias survived {zeroed_at} samples"
assert not presentation.ready
assert presentation.display_quality == "provisional"
def test_a_noisy_channel_is_zeroed_from_the_first_samples_too() -> None:
"""The provisional band covers the samples it was fitted from.
With a handful of samples the p99 residual is essentially the largest one, so
the newest sample of the window is inside the band: the first tenths of a
second are zeroed, and the *first* sample is not claimed to be zero because
nothing has been estimated from it yet.
"""
rng = np.random.default_rng(11)
bias = np.array(REPORTED_RAW_BIAS)
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
for index in range(8):
sample = bias + rng.normal(0.0, 30.0, size=3)
presentation.observe(tuple(sample), t_rel_s=index / 30.0, flow_p95=0.1)
if index < 2:
# Nothing has been estimated from two samples: those frames honestly
# show the measurement (and are asserted to do so elsewhere).
continue
displayed = presentation.accept(tuple(sample), unit="raw")
assert displayed.values == (0.0, 0.0, 0.0), (
f"sample {index} of a resting noisy channel was drawn"
)
def test_a_contact_present_at_session_start_is_never_zeroed() -> None:
"""A press at run start must stay visible: no estimate may be invented for it.
The contact guard decides what feeds the estimate, so a window that has only
ever seen a contact has nothing to subtract and displays the measurement.
"""
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
pressed = (6000.0, 40.0, -180.0)
for index in range(20):
assert (
presentation.observe(
pressed, t_rel_s=index / 30.0, contact_found=True, flow_p95=0.1
)
is None
)
displayed = presentation.accept(pressed, unit="raw")
assert displayed.values[0] == pressed[0], "a contact at startup was zeroed away"
assert not displayed.suppressed[0]
assert not presentation.display_ready
assert presentation.display_quality == "none"
def test_a_contaminated_window_never_falls_back_to_the_raw_bias() -> None:
"""A window that closes badly keeps the estimate it already worked out.
Falling back to the raw offset is the reported defect. Both ways a startup
window can be spoiled — a contact, or too much flow — must leave the display
zeroed and say so through ``display_quality``.
"""
bias = REPORTED_RAW_BIAS
pressed = (6000.0, 40.0, -180.0)
for spoiled_by in ("contact", "flow"):
presentation = ForcePresentation(
enabled=True, min_samples=30, baseline_seconds=3.0
)
for index in range(3):
presentation.observe(bias, t_rel_s=index / 30.0, flow_p95=0.1)
assert presentation.display_ready and not presentation.ready
for index in range(90): # the window is spoiled before it can close
presentation.observe(
pressed,
t_rel_s=0.5 + index / 30.0,
contact_found=(spoiled_by == "contact"),
flow_p95=0.9 if spoiled_by == "flow" else 0.1,
)
estimate = presentation.observe(bias, t_rel_s=4.0, flow_p95=0.1)
assert estimate is not None, spoiled_by
assert estimate.quality == "contaminated", (spoiled_by, estimate)
assert not presentation.ready, spoiled_by
assert presentation.display_quality == "contaminated", spoiled_by
assert presentation.accept(bias, unit="raw").values == (0.0, 0.0, 0.0), spoiled_by
def test_a_window_that_closed_short_still_applies_what_it_learned() -> None:
"""A session shorter than the quiet window is not a reason to show the bias."""
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=0.5)
for index in range(3):
presentation.observe(REPORTED_RAW_BIAS, t_rel_s=index / 30.0, flow_p95=0.1)
estimate = presentation.observe(REPORTED_RAW_BIAS, t_rel_s=0.6, flow_p95=0.1)
assert estimate is not None and estimate.quality == "insufficient"
assert presentation.display_quality == "insufficient"
assert presentation.accept(REPORTED_RAW_BIAS, unit="raw").values == (0.0, 0.0, 0.0)
def test_a_frozen_window_stops_chasing_the_data() -> None:
"""The provisional phase is a start, not a licence to re-zero a press."""
presentation = ForcePresentation(
enabled=True, min_samples=5, baseline_seconds=3.0, min_deadband_raw=50.0
)
for index in range(10):
presentation.observe(REPORTED_RAW_BIAS, t_rel_s=index / 30.0, flow_p95=0.1)
assert presentation.ready
assert presentation.display_quality == "ok"
# A held press after the window is a reading, not a new zero.
assert presentation.accept((5000.0, 34.16, -186.1), unit="raw").values[0] > 3000.0
def test_the_cards_and_the_curves_are_the_same_presentation() -> None:
"""What ``accept`` puts on a card is the row ``present`` draws on its curve.
This is what the extra "the estimate is ready" gate in the value cards broke:
the card kept the raw measurement while the curve was already drawn from the
applied estimate, so the two disagreed about the same sample.
"""
bias = REPORTED_RAW_BIAS
#: The samples the acquisition would flag as a contact: they must not feed
#: the idle estimate, or the display would zero the press away.
presses = {
3: (bias[0] + 4000.0, bias[1] + 10.0, bias[2] - 10.0),
5: (bias[0] - 4000.0, bias[1], bias[2]),
}
stream = [
bias,
bias,
bias,
presses[3],
bias,
presses[5],
bias,
]
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
for index, sample in enumerate(stream):
presentation.observe(
sample,
t_rel_s=index / 30.0,
flow_p95=0.1,
contact_found=index in presses,
)
card = presentation.accept(sample, unit="raw")
curve, suppressed = presentation.present([sample], unit="raw")
assert card.values == tuple(float(value) for value in curve[0]), index
assert card.suppressed == tuple(bool(flag) for flag in suppressed[0]), index
# The assertion is not vacuous: the excursion really is on both.
assert presentation.accept(stream[3], unit="raw").values[0] > 3000.0
def test_present_uses_the_provisional_estimate_before_the_window_closes() -> None:
"""The curves get the same early zero as the cards, not only after 3 s."""
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
block = np.asarray([REPORTED_RAW_BIAS] * 4, dtype=float)
raw, _suppressed = presentation.present(block, unit="raw")
assert raw[0].tolist() == list(REPORTED_RAW_BIAS), "nothing was learned yet"
for index in range(3):
presentation.observe(REPORTED_RAW_BIAS, t_rel_s=index / 30.0, flow_p95=0.1)
displayed, suppressed = presentation.present(block, unit="raw")
assert displayed.tolist() == [[0.0, 0.0, 0.0]] * 4
assert suppressed.all()
assert not presentation.ready, "the whole quiet window was not needed"
+224 -2
View File
@@ -9,6 +9,7 @@ stage owes, and a synthetic one can only prove the code agrees with itself.
from __future__ import annotations
import json
import shutil
import struct
import zipfile
from pathlib import Path
@@ -42,8 +43,11 @@ from orisys_app_shared.scene3d.glb import (
from orisys_app_shared.scene3d.legacy import (
LegacyFingerImporter,
build_binding,
crease_aware_mesh,
estimate_meters_per_source_unit,
load_legacy_domain,
normals_disagree_with_faces,
split_at_crease_edges,
split_at_uv_seams,
uv_correspondence_residual,
validate_rigid_transform,
@@ -72,9 +76,11 @@ def _identity() -> np.ndarray:
def _import_run(**kwargs) -> SceneAssets:
export_dir = kwargs.pop("export_dir", RUN_DIR)
map_path = kwargs.pop("map_path", RUN_MAP)
return LegacyFingerImporter().convert(
RUN_DIR,
RUN_MAP,
export_dir,
map_path,
transform_sensor_from_source=kwargs.pop("transform", _identity()),
meters_per_source_unit=kwargs.pop("units", 1e-3),
**kwargs,
@@ -97,6 +103,79 @@ def test_the_real_export_imports_and_validates():
assert assets.source_vertex_of_render.shape == (1032,)
def _write_housing(path: Path) -> Path:
"""A tiny tetrahedron standing in for a mounting frame."""
vertices = np.array(
[[0.0, 0.0, 0.0], [10.0, 0.0, 0.0], [0.0, 10.0, 0.0], [0.0, 0.0, 10.0]],
dtype=np.float32,
)
faces = np.array([[0, 1, 2], [0, 1, 3], [0, 2, 3], [1, 2, 3]], dtype=np.int32)
np.savez(path, vertices=vertices, faces=faces)
return path
@requires_run
def test_a_housing_becomes_its_own_part_without_touching_the_geometry_ids(tmp_path: Path):
"""The frame is display-only: it must not enter the identities the device verifies."""
plain = _import_run()
housing_path = _write_housing(tmp_path / "housing.npz")
with_housing = _import_run(housing_path=housing_path)
assert [part.part_id for part in plain.parts] == ["sensing_surface"]
assert [(part.part_id, part.role) for part in with_housing.parts] == [
("sensing_surface", "sensing"),
("housing", "housing"),
]
assert with_housing.geometry_id == plain.geometry_id, (
"adding a housing changed the sensing geometry id, which the device verifies"
)
assert with_housing.source_geometry_id == plain.source_geometry_id
assert with_housing.asset_hash != plain.asset_hash, (
"the package content changed, so the asset hash has to say so"
)
assert validate_assets(with_housing) == ()
# Units and the rigid transform are applied to the housing exactly as to the surface.
frame = with_housing.part("housing")
assert frame is not None
assert np.allclose(np.ptp(np.asarray(frame.positions_sensor_m), axis=0), 0.01, atol=1e-6)
@requires_run
def test_a_housing_in_the_export_directory_is_used_automatically(tmp_path: Path, monkeypatch):
"""``<legacy-dir>/housing.npz`` is picked up, so a re-export needs no new flag."""
export = tmp_path / "export"
export.mkdir()
shutil.copy(RUN_DIR / "mesh.npz", export / "mesh.npz")
assert len(_import_run(export_dir=export).parts) == 1
_write_housing(export / "housing.npz")
assert [part.role for part in _import_run(export_dir=export).parts] == [
"sensing",
"housing",
]
@requires_run
def test_a_requested_housing_that_is_missing_is_refused(tmp_path: Path):
"""A missing frame must not look like a package that simply has none."""
with pytest.raises(ThreeDError) as caught:
_import_run(housing_path=tmp_path / "not-there.npz")
assert caught.value.diagnostic.code == "geometry_mismatch"
@requires_run
def test_a_housing_with_an_out_of_range_triangle_is_refused(tmp_path: Path):
path = tmp_path / "housing.npz"
np.savez(
path,
vertices=np.zeros((3, 3), dtype=np.float32),
faces=np.array([[0, 1, 9]], dtype=np.int32),
)
with pytest.raises(ThreeDError) as caught:
_import_run(housing_path=path)
assert caught.value.diagnostic.code == "geometry_mismatch"
@requires_run
def test_the_uv_convention_is_measured_against_the_asset():
"""The one fact the importer must not guess: which pixel a UV addresses."""
@@ -235,6 +314,149 @@ def test_vertex_normals_are_unit_and_area_weighted():
assert np.isfinite(lonely[4]).all()
def _box(half: float = 1.0) -> tuple[np.ndarray, np.ndarray]:
"""A closed cube: every corner is a 90° crease, every edge a smooth pair."""
positions = np.array(
[
[-half, -half, -half], [half, -half, -half], [half, half, -half],
[-half, half, -half], [-half, -half, half], [half, -half, half],
[half, half, half], [-half, half, half],
],
dtype=np.float64,
)
triangles = np.array(
[
[0, 2, 1], [0, 3, 2], # -z
[4, 5, 6], [4, 6, 7], # +z
[0, 1, 5], [0, 5, 4], # -y
[2, 3, 7], [2, 7, 6], # +y
[1, 2, 6], [1, 6, 5], # +x
[3, 0, 4], [3, 4, 7], # -x
],
dtype=np.int64,
)
return positions, triangles
def test_crease_aware_normals_agree_with_every_face_of_a_box():
"""The defect this fixes: an averaged corner normal agrees with *no* face."""
positions, triangles = _box()
averaged = vertex_normals(positions, triangles)
crease = crease_aware_mesh(positions, triangles)
split_positions, split_triangles, split_normals = crease
assert len(split_triangles) == len(triangles), "splitting must not change the faces"
assert len(split_positions) == 24, "each of the 8 corners needs 3 normals"
def worst(normals, faces, verts):
a, b, c = verts[faces[:, 0]], verts[faces[:, 1]], verts[faces[:, 2]]
face = np.cross(b - a, c - a)
unit = face / np.linalg.norm(face, axis=1, keepdims=True)
incident = [[] for _ in range(len(verts))]
for index, triangle in enumerate(faces):
for k in range(3):
incident[int(triangle[k])].append(index)
return max(
float(np.degrees(np.arccos(np.clip(unit[incident[v]] @ normals[v], -1, 1))).min())
for v in range(len(verts))
if incident[v]
)
assert worst(averaged, triangles, positions) > 30.0, (
"the fixture must actually exhibit the defect"
)
assert worst(split_normals, split_triangles, split_positions) < 1e-6, (
"a split corner normal must be exactly a face normal on a cube"
)
def test_crease_splitting_keeps_the_smooth_parts_welded():
"""A cylinder's curved side must survive whole; only its rim split."""
ring = 24
angles = np.linspace(0.0, 2.0 * np.pi, ring, endpoint=False)
positions = np.vstack(
[
np.column_stack([np.cos(angles), np.sin(angles), np.zeros(ring)]),
np.column_stack([np.cos(angles), np.sin(angles), np.ones(ring)]),
]
)
triangles = []
for i in range(ring):
j = (i + 1) % ring
triangles.append([i, j, ring + i])
triangles.append([j, ring + j, ring + i])
triangles = np.asarray(triangles, dtype=np.int64)
split_positions, split_triangles, _ = split_at_crease_edges(positions, triangles)
# The side is a smooth ring (3.7° per step), so nothing on it may be duplicated;
# the two rims are boundaries, so the split only separates the *caps'* edges.
assert len(split_triangles) == len(triangles)
assert len(split_positions) == len(positions), (
"a smoothly curved side must stay welded"
)
def test_normals_disagree_with_faces_detects_the_defect():
positions, triangles = _box()
assert normals_disagree_with_faces(
positions, triangles, vertex_normals(positions, triangles)
)
_, _, crease = crease_aware_mesh(positions, triangles)
split_positions, split_triangles, _ = split_at_crease_edges(positions, triangles)
assert not normals_disagree_with_faces(split_positions, split_triangles, crease)
# A flat sheet has no crease at all.
flat = np.array([[0.0, 0.0, 0.0], [1.0, 0.0, 0.0], [0.0, 1.0, 0.0]])
assert not normals_disagree_with_faces(
flat, np.array([[0, 1, 2]]), np.array([[0.0, 0.0, 1.0]] * 3)
)
@requires_run
def test_the_shipped_housing_has_no_stale_shading():
"""The real housing: 134 of 2172 edges are creases, so plain averaging bands."""
import io
import zipfile
package = (
Path(__file__).resolve().parents[2]
/ "apps" / "studio" / "config" / "sensors" / "finger" / "scene"
/ "finger-display-1.0.0.oryscene"
)
if not package.is_file():
pytest.skip("the compiled finger package is not present")
with zipfile.ZipFile(package) as archive:
positions = np.load(io.BytesIO(archive.read("housing/positions.npy"))).astype(np.float64)
triangles = np.load(io.BytesIO(archive.read("housing/triangles.npy"))).astype(np.int64)
normals = np.load(io.BytesIO(archive.read("housing/normals.npy"))).astype(np.float64)
# Every stored normal agrees with an incident face, which is only possible
# because the source mesh's 726 vertices became 861 (one per side of a crease).
assert not normals_disagree_with_faces(positions, triangles, normals), (
"the shipped housing still carries normals averaged across its 90° corners"
)
# ...and the split is not a blanket unweld: the smooth vertices stay shared.
unique_positions = len(np.unique(positions, axis=0))
assert unique_positions == 726, (
f"{unique_positions} unique positions: the smooth surface was unwelded too"
)
assert len(positions) == 861, (
f"{len(positions)} vertices: the creases were not split into their own normals"
)
# The source mesh this was compiled from is the shape that used to be shipped,
# and plain averaging on it is exactly the defect.
source = Path(
r"D:\orisys_ws\orisys-sensor-design\build\export\finger\r01\compute\housing.npz"
)
if source.is_file():
with np.load(source, allow_pickle=False) as handle:
source_positions = np.asarray(handle["vertices"], dtype=np.float64)
source_triangles = np.asarray(handle["faces"], dtype=np.int64)
assert len(source_positions) == 726, len(source_positions)
assert normals_disagree_with_faces(
source_positions,
source_triangles,
vertex_normals(source_positions, source_triangles),
), "the detector does not fire on the geometry that actually banded"
# ------------------------------------------------------------------- binding
def _domain(shape_hw=(4, 5), flip_v=False, world=None):
valid = np.ones(shape_hw, dtype=bool)
+277 -3
View File
@@ -13,8 +13,10 @@ that only have the SDK.
from __future__ import annotations
import io
import json
import shutil
import zipfile
from dataclasses import replace
from pathlib import Path
@@ -38,6 +40,106 @@ BUNDLE = (
)
PACKAGE = BUNDLE / "scene" / "finger-display-1.0.0.oryscene"
#: The **runtime** SDK asset: the mesh the device actually runs, in source units.
SDK_ASSET_DIR = Path(r"D:\orisys_ws\orisys-sdk\sensors\finger\assets\30fps-0703")
SDK_MESH = SDK_ASSET_DIR / "mesh.npz"
#: The physical statement about that asset: one source unit is one millimetre.
#: Documented in ``tools/scene_assets/README.md`` and stated by the compiler's
#: ``--meters-per-source-unit``; it is the *design* export that is in other units.
SDK_METERS_PER_SOURCE_UNIT = 0.001
#: What the SDK asset measures when the file above is not on this machine (metres):
#: the same 15.81 x 15.0 x 7.5 mm envelope, kept as a documented constant so the
#: guard still runs — and can still fail — on a checkout without the SDK.
SENSOR_ENVELOPE_FALLBACK_M = np.array([0.01580961, 0.015, 0.0075])
#: How far the shipped package may be from the SDK asset's own envelope. The defect
#: this guards against is a 54.38x error, so one percent is three orders of magnitude
#: tighter and still leaves room for float32 storage and the export's own rounding.
ENVELOPE_RELATIVE_TOLERANCE = 0.01
def sensor_frame_envelope_m() -> np.ndarray:
"""The sensing surface's true envelope, in the sensor frame, in metres.
Derived from the **SDK asset mesh** — the file the device runs — and *not* from
the package: a guard that reads its expectation out of the thing it is checking
cannot fail. ``manifest.length_unit`` is ``m`` and ``coordinate_frame`` is
``sensor``, so the package's arrays are directly comparable with this.
"""
if not SDK_MESH.is_file():
return SENSOR_ENVELOPE_FALLBACK_M.copy()
with np.load(SDK_MESH, allow_pickle=False) as mesh:
extent = np.ptp(np.asarray(mesh["vertices"], dtype=np.float64), axis=0)
return extent * SDK_METERS_PER_SOURCE_UNIT
def _package_arrays(package: Path) -> tuple[dict, dict[str, np.ndarray]]:
"""The raw manifest and entries, read straight out of the zip.
Deliberately not through the loader: this guard is about the *written file*,
and a check that goes through code which might normalise or repair would not
see what actually shipped.
"""
with zipfile.ZipFile(package) as archive:
manifest = json.loads(archive.read("manifest.json"))
arrays = {
name: np.load(io.BytesIO(archive.read(name)))
for name in archive.namelist()
if name.endswith(".npy")
}
return manifest, arrays
def sensing_envelope_m(package: Path) -> np.ndarray:
_, arrays = _package_arrays(package)
return np.ptp(np.asarray(arrays["sensing/positions.npy"], dtype=np.float64), axis=0)
def assert_package_geometry_is_sensor_scale(package: Path) -> None:
"""The scale check: the shipped envelope must be the SDK asset's envelope.
Two independent statements inside the package have to be right, and the second
is the one that reaches the renderer's own framing:
* ``sensing/positions.npy``, in the metres ``manifest.length_unit`` claims;
* ``manifest.bounds_sensor_m``, which the loader trusts for the camera.
"""
expected = sensor_frame_envelope_m()
manifest, arrays = _package_arrays(package)
length_unit = str(manifest.get("length_unit") or "")
assert length_unit == "m", (
f"{package.name} declares length_unit={length_unit!r}; this guard compares "
"metres, so a different unit statement has to be converted first"
)
extent = np.ptp(np.asarray(arrays["sensing/positions.npy"], dtype=np.float64), axis=0)
np.testing.assert_allclose(
extent,
expected,
rtol=ENVELOPE_RELATIVE_TOLERANCE,
err_msg=(
f"{package.name}: the sensing envelope in the units the manifest claims "
"(m) does not match the SDK asset's own sensor-frame envelope. Measured "
f"{np.sort(extent)} m, the SDK asset states {np.sort(expected)} m "
f"(a factor of {float(np.sort(extent)[-1] / np.sort(expected)[-1]):.5f} on "
"the largest axis). This is the unit statement, not a formatting choice: "
"the design export is not in millimetres, and calling its units "
"millimetres compiles cleanly and is 54.38x too large."
),
)
bounds = np.asarray(manifest.get("bounds_sensor_m"), dtype=np.float64)
assert bounds.shape == (2, 3), f"{package.name}: bounds_sensor_m is {bounds.shape}"
np.testing.assert_allclose(
bounds[1] - bounds[0],
expected,
rtol=ENVELOPE_RELATIVE_TOLERANCE,
err_msg=(
f"{package.name}: manifest.bounds_sensor_m spans {bounds[1] - bounds[0]} m, "
f"but the SDK asset states {expected} m: the camera would frame the wrong box"
),
)
def _require_bundle() -> None:
if not (BUNDLE / "sensor.json").is_file():
@@ -50,6 +152,178 @@ def _require_package() -> None:
pytest.skip("the compiled finger package is not present")
def _bundle_map_name() -> str:
"""The map filename this bundle's own config resolves to — derived, never hard-coded.
The filename follows ``camera.source_scale`` (``scaled_map_cache_path``), and the bundle moved
from ``0.4`` to ``0.42`` when the package was recompiled from the authoritative design export.
Deriving it here keeps these tests about the behaviour they name — the device description comes
from the files the runtime loads — instead of about one filename that any scale change renames.
"""
_require_bundle()
config = json.loads((BUNDLE / "sensor.json").read_text(encoding="utf-8"))
scale = config.get("camera", {}).get("source_scale")
return map_path_for(BUNDLE / "assets" / "30fps-0703", scale).name
def test_the_shipped_package_looks_at_the_sensing_side_of_the_surface():
"""The default camera must be in front of the plate, not behind it.
The shipped ``default`` preset used to be a fixed vector, which is only correct
if the export's frame happens to equal the sensor frame. The design export's does
not: that preset ended up 122.7° from the sensing surface's own normal — behind
the surface — where the frame hides essentially all of it. Both the package's
baked preset and the presets the loader derives are checked here.
"""
_require_package()
from orisys_app_shared.scene3d.contracts import DisplaySettings
from orisys_app_shared.scene3d.legacy import (
default_finger_presets,
sensing_surface_normal,
)
from orisys_app_shared.scene3d.package import read_package, scene_from_contents
scene = scene_from_contents(read_package(PACKAGE))
sensing = scene.sensing_part()
normal = sensing_surface_normal(
np.asarray(sensing.positions_sensor_m, dtype=np.float64),
np.asarray(sensing.triangles, dtype=np.int64),
)
frame = _package_arrays(PACKAGE)[0]
baked = next(item for item in frame["camera_presets"] if item["name"] == "default")
centre = np.asarray(scene.bounds_sensor_m, dtype=np.float64).mean(axis=0)
def angle(preset) -> float:
eye = np.asarray(preset.eye_sensor_m, dtype=np.float64)
offset = eye - centre
offset = offset / np.linalg.norm(offset)
return float(np.degrees(np.arccos(np.clip(offset @ normal, -1.0, 1.0))))
baked_angle = angle(
type(scene.camera_presets[0])(
name="default",
eye_sensor_m=tuple(baked["eye"]),
target_sensor_m=tuple(baked["target"]),
up_sensor=tuple(baked["up"]),
vertical_fov_rad=float(baked["fov"]),
)
)
assert baked_angle < 60.0, (
f"the baked default preset is {baked_angle:.1f}° from the sensing surface's "
"normal: it points at the back of the plate"
)
loaded_angle = angle(next(p for p in scene.camera_presets if p.name == "default"))
assert loaded_angle < 60.0, f"the loaded default preset is {loaded_angle:.1f}° off"
derived = default_finger_presets(
scene.bounds_sensor_m,
positions_sensor_m=np.asarray(sensing.positions_sensor_m, dtype=np.float64),
triangles=np.asarray(sensing.triangles, dtype=np.int64),
)
assert angle(derived[0]) < 40.0
assert all(abs(angle(preset)) <= 90.0 + 1e-6 for preset in derived)
# The housing stays hidden by default: the operator turns it on for inspection.
assert DisplaySettings().housing_visible is False
def test_a_preset_behind_the_surface_is_replaced_on_load(tmp_path):
"""The repair path, driven with a deliberately wrong preset.
A package whose ``default`` preset points at the back of the plate is loaded
through the real container, and the loader replaces the whole set with one
derived from the mesh. The point is that the guard is on the *consuming* side:
a package baked by an older compiler is corrected without being rebuilt.
"""
_require_package()
from dataclasses import replace
from orisys_app_shared.scene3d.legacy import sensing_surface_normal
from orisys_app_shared.scene3d.package import (
PackagedAssetLoader,
read_package,
scene_from_contents,
write_package,
)
scene = scene_from_contents(read_package(PACKAGE))
normal = sensing_surface_normal(
np.asarray(scene.sensing_part().positions_sensor_m, dtype=np.float64),
np.asarray(scene.sensing_part().triangles, dtype=np.int64),
)
centre = np.asarray(scene.bounds_sensor_m, dtype=np.float64).mean(axis=0)
backwards = tuple(centre - normal * 0.02)
broken = replace(
scene.camera_presets[0],
eye_sensor_m=backwards,
target_sensor_m=tuple(centre),
)
assert float(np.dot(np.asarray(backwards) - centre, normal)) < 0.0, "fixture is not behind"
path = tmp_path / "backwards.oryscene"
write_package(replace(scene, camera_presets=(broken,) + tuple(scene.camera_presets[1:])), path)
# The file keeps what was written...
assert _package_arrays(path)[0]["camera_presets"][0]["eye"] == list(backwards)
# ...and the loader does not present a view of the back of the sensor.
from orisys_app_shared.scene3d.contracts import ExpectedGeometry
expected = ExpectedGeometry(
"finger", "*", scene.geometry_id, scene.field_domain_id, scene.shape_hw,
scene.source_geometry_id,
)
loaded = PackagedAssetLoader().load(path, expected)
fixed = next(preset for preset in loaded.camera_presets if preset.name == "default")
offset = np.asarray(fixed.eye_sensor_m, dtype=np.float64) - centre
offset = offset / np.linalg.norm(offset)
angle = float(np.degrees(np.arccos(np.clip(offset @ normal, -1.0, 1.0))))
assert angle < 60.0, f"the loader kept a preset {angle:.1f}° off the surface normal"
def test_the_shipped_package_is_at_sensor_scale_not_design_scale():
"""The scale guard that was missing.
Every other check in this file is about *identity*: the geometry id, the domain
id and the source-geometry id are digests of unit-free data, so a package whose
geometry is 54.38x too large passes all of them — the ids are correct and only
the scale is wrong (measured: ``sensing/positions.npy`` at 0.8157 x 0.4079 x
0.8598 m in a package that claims millimetres-as-metres). This asserts the
package agrees with the SDK asset mesh's own physical envelope.
"""
_require_package()
assert_package_geometry_is_sensor_scale(PACKAGE)
def test_the_guard_itself_discriminates():
"""A guard that cannot fail is not a guard: 54.38x must be rejected."""
_require_package()
expected = sensor_frame_envelope_m()
too_large = expected * 54.381958120648946
with pytest.raises(AssertionError):
np.testing.assert_allclose(
too_large, expected, rtol=ENVELOPE_RELATIVE_TOLERANCE
)
assert float(np.max(too_large / expected)) == pytest.approx(54.38196, rel=1e-4)
def test_the_expected_envelope_comes_from_the_sdk_asset_not_the_package():
"""The expectation is external, so a wrong package cannot define it as right."""
if not SDK_MESH.is_file():
pytest.skip("the SDK asset mesh is not present on this machine")
with np.load(SDK_MESH, allow_pickle=False) as mesh:
extent_source_units = np.ptp(
np.asarray(mesh["vertices"], dtype=np.float64), axis=0
)
assert np.allclose(np.sort(extent_source_units), [7.5, 15.0, 15.81], atol=0.02), (
extent_source_units
)
assert np.allclose(
np.sort(sensor_frame_envelope_m()), [0.0075, 0.015, 0.01581], atol=2e-5
)
if PACKAGE.is_file():
# And it really is a different statement from the package's own numbers.
assert not np.allclose(
sensing_envelope_m(PACKAGE), extent_source_units
), "the SDK asset and the package are in the same units: this guard would be vacuous"
def test_the_map_name_follows_the_scale_the_sdk_would_use(tmp_path):
assert map_path_for(tmp_path, 0.4).name == "unroll_map_uv_s0.4.npz"
assert map_path_for(tmp_path, 1.0).name == "unroll_map_uv.npz"
@@ -67,7 +341,7 @@ def test_the_device_description_comes_from_the_files_the_runtime_loads():
assert device.expected.field_domain_id
assert device.expected.source_geometry_id
assert tuple(device.expected.shape_hw) == (300, 280)
assert Path(device.detail["map"]).name == "unroll_map_uv_s0.4.npz"
assert Path(device.detail["map"]).name == _bundle_map_name()
assert device.detail["shape_hw"] == "300x280"
@@ -84,7 +358,7 @@ def test_a_same_size_different_mapping_is_refused(tmp_path):
_require_package()
fake = tmp_path / "finger"
shutil.copytree(BUNDLE, fake)
map_path = fake / "assets" / "30fps-0703" / "unroll_map_uv_s0.4.npz"
map_path = fake / "assets" / "30fps-0703" / _bundle_map_name()
with np.load(map_path, allow_pickle=False) as handle:
data = {key: handle[key] for key in handle.files}
# The same grid, the same valid area — a different mapping.
@@ -149,7 +423,7 @@ def test_a_missing_map_names_the_file_it_looked_for(tmp_path):
with pytest.raises(ThreeDError) as excinfo:
describe_sensor_geometry("finger", fake / "sensor.json")
assert excinfo.value.diagnostic.code == "asset_rejected"
assert "unroll_map_uv_s0.4.npz" in excinfo.value.diagnostic.message
assert _bundle_map_name() in excinfo.value.diagnostic.message
def test_a_bundle_without_an_assets_dir_is_unverified(tmp_path):
+17 -5
View File
@@ -527,18 +527,30 @@ def _tool(name: str):
def _shipped_asset_dirs() -> tuple[Path, Path] | None:
"""The Blender export and its unroll map, when the repository carries them."""
base = (
"""The Blender export and its unroll map, when the repository carries them.
The map's name follows ``camera.source_scale`` (the SDK's ``scaled_map_cache_path``), so it is
derived from the bundle's own ``sensor.json`` rather than pinned: a scale change renames the
file, and a hard-coded name turns this acceptance test into a silent skip.
"""
bundle = (
Path(__file__).resolve().parents[2]
/ "apps"
/ "studio"
/ "config"
/ "sensors"
/ "finger"
/ "assets"
/ "30fps-0703"
)
mapping = base / "unroll_map_uv_s0.4.npz"
base = bundle / "assets" / "30fps-0703"
config_path = bundle / "sensor.json"
if not config_path.is_file():
return None
from orisys_app_shared.acquisition.geometry import map_path_for
scale = (json.loads(config_path.read_text(encoding="utf-8")).get("camera") or {}).get(
"source_scale"
)
mapping = map_path_for(base, scale)
if (base / "mesh.npz").is_file() and mapping.is_file():
return base, mapping
return None
+80
View File
@@ -410,6 +410,86 @@ def test_a_file_that_yields_no_frames_ends_the_session_with_an_error(monkeypatch
_restore(saved)
def test_a_paused_worker_stops_producing_but_keeps_the_device_open():
"""Pause is not stop: compute stops, the capture keeps running.
It lives in this file because the fake device that can answer the question (does the
source keep publishing while the loop takes no frame?) is the harness above.
"""
from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker
CREATED.clear()
saved = _install()
worker = SdkAcquisitionWorker()
packets: list = []
lock = threading.Lock()
def sink(packet) -> None:
with lock:
packets.append(packet)
try:
worker.add_sink(sink)
worker.start(0, config_ref=BOX_CONFIG)
time.sleep(0.4)
before = len(packets)
assert before > 0, "the fake device produced nothing to pause"
assert not worker.paused
worker.pause()
assert worker.paused
time.sleep(0.4)
during = len(packets)
# The frames the source published across the pause: the device is demonstrably
# alive, which is what makes this different from ``stop``.
source_seq_at_pause = CREATED[-1].capture_source.status.last_seq
assert during == before, (
f"a paused worker emitted {during - before} packet(s): the computation was "
"supposed to stop"
)
worker.resume()
assert not worker.paused
time.sleep(0.4)
after = len(packets)
assert after > during, "the worker did not resume producing packets"
assert CREATED[-1].capture_source.status.last_seq > source_seq_at_pause, (
"the capture source stopped publishing while paused — the device was released "
"or the reader was starved, and neither is what pause means"
)
# Resuming continues from the newest frame rather than replaying the pause: the
# packet that follows the pause is dated after it, not before.
assert float(packets[after - 1].mono_t) > 0.0
stop_requests = getattr(worker, "_stop").is_set()
assert not stop_requests, "pausing must not set the stop flag"
worker.stop(wait=True)
finally:
_restore(saved)
def test_a_new_session_starts_unpaused():
"""Pausing is a statement about one session, not a sticky property of the device."""
from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker
CREATED.clear()
saved = _install()
worker = SdkAcquisitionWorker()
try:
worker.add_sink(lambda _packet: None)
worker.start(0, config_ref=BOX_CONFIG)
time.sleep(0.3)
worker.pause()
assert worker.paused
worker.stop(wait=True)
assert not worker.paused, "stop cleared the pause flag"
worker.start(0, config_ref=BOX_CONFIG)
assert not worker.paused, "the next session started paused"
worker.stop(wait=True)
finally:
_restore(saved)
if __name__ == "__main__": # pragma: no cover - manual runner
failures = 0
for name, function in sorted(globals().items()):