diff --git a/CHANGELOG.md b/CHANGELOG.md index 1761c5b..0deabb6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,9 @@ All notable user-facing changes to DerZug are documented here. - Aggregate's phase-weighted stack no longer fails headlessly when no stack dimension is set; it defaults to `distance` (else the first patch dimension), the same default the widget applies, and the widget's transform-dimension chooser now excludes that effective stack dimension. - The Coords widget now builds and validates its task exclusively through the node layer, so the task that runs on the canvas and the task exported into a saved workflow are always identical. Two behavior fixes come with it: headless callers filling only the `set_coords` draft fields (`set_coords_dim`/`start`/`stop`/`step`) now get a real coordinate update instead of a silent no-op, and data-flipping a non-dimension coordinate now reports through the "Invalid flip selection" banner instead of a generic operation failure. - The Spool widget's preview/output pipeline now runs the same node-layer select and chunk stages as a headless workflow. A chunk value that parses to `None` (e.g. the text `None`) now disables chunking headlessly, matching the canvas chunk controls. +- Filter `mode` parameters are now validated `Literal`s on the params models, so an unsupported boundary mode (e.g. `SavgolFilterParams(mode="reflect")`, which SciPy rejects) fails at validation with a clear message instead of surviving until the SciPy call. +- `SelectionState.apply_select_params` now accepts `patch=None` for display-only hosts, replacing the Select widget's hand-built state. +- PlayAudio's signal processing (patch validation, rate inference, PCM normalization, time-scaling and resampling) moved to the Qt-free `derzug.nodes.playaudio`, and the node's `time_scale`/`volume_percent` parameters are now actually consumed there: the new `render_audition(patch, params)` returns device-ready PCM headlessly. ### Changed (breaking) diff --git a/src/derzug/models/selection_state.py b/src/derzug/models/selection_state.py index 5d60931..fda3490 100644 --- a/src/derzug/models/selection_state.py +++ b/src/derzug/models/selection_state.py @@ -576,13 +576,32 @@ def to_select_params(self) -> SelectParams: samples=flags["samples"], ) - def apply_select_params(self, params: SelectParams, patch) -> None: - """Seed patch-mode selection state from public patch.select parameters.""" + def apply_select_params(self, params: SelectParams, patch=None) -> None: + """Seed patch-mode selection state from public patch.select parameters. + + With a patch the ranges are clamped against its dimensions; without + one (a display-only host) each parameter range doubles as its own + extent so the state can still be shown. + """ basis = PatchSelectionBasis.ABSOLUTE if params.relative: basis = PatchSelectionBasis.RELATIVE elif params.samples: basis = PatchSelectionBasis.SAMPLES + if patch is None: + ranges: dict[str, tuple[Any, Any]] = {} + for dim, value_range in params.kwargs.items(): + try: + low, high = value_range + except (TypeError, ValueError): + continue + ranges[dim] = (low, high) + self.mode = SelectionMode.PATCH + self.patch.basis = basis + self.patch.extents = dict(ranges) + self.patch.ranges = ranges + self.patch.enabled = {dim: True for dim in ranges} + return self.set_patch_source(patch) self.set_patch_basis(basis) for dim, value_range in params.kwargs.items(): diff --git a/src/derzug/nodes/filter.py b/src/derzug/nodes/filter.py index 9584b03..431bdb2 100644 --- a/src/derzug/nodes/filter.py +++ b/src/derzug/nodes/filter.py @@ -14,7 +14,7 @@ from __future__ import annotations -from typing import Annotated, ClassVar, Literal +from typing import Annotated, ClassVar, Literal, get_args import dascore as dc from pydantic import BaseModel, Field, TypeAdapter @@ -36,6 +36,14 @@ "wiener_filter", ) +#: Boundary modes the general SciPy-backed filters accept. Savitzky-Golay +#: rejects ``reflect``, so it gets its own narrower set; typing the params +#: fields with these keeps a headless ``mode`` typo from surviving until SciPy. +FilterMode = Literal["reflect", "constant", "nearest", "wrap", "mirror", "interp"] +SavgolFilterMode = Literal["mirror", "constant", "nearest", "wrap", "interp"] +MODE_OPTIONS: tuple[str, ...] = get_args(FilterMode) +SAVGOL_MODE_OPTIONS: tuple[str, ...] = get_args(SavgolFilterMode) + # -- Parameter models --------------------------------------------------------- @@ -72,7 +80,7 @@ class MedianFilterParams(_SharedFilter): kind: Literal["median_filter"] = "median_filter" window: str = "0.01" # stored as filter_window samples: bool = False - mode: str = "reflect" + mode: FilterMode = "reflect" cval: float = 0.0 @@ -93,7 +101,7 @@ class SavgolFilterParams(_SharedFilter): window: str = "0.01" # stored as filter_window polyorder: int = 3 samples: bool = False - mode: str = "interp" # savgol modes differ from the other filters + mode: SavgolFilterMode = "interp" # savgol rejects reflect cval: float = 0.0 @@ -122,7 +130,7 @@ class GaussianFilterParams(_SharedFilter): default_factory=lambda: [GaussianWindow(window="")] ) samples: bool = False - mode: str = "reflect" + mode: FilterMode = "reflect" cval: float = 0.0 truncate: float = 4.0 @@ -131,7 +139,7 @@ class SobelFilterParams(_SharedFilter): """Sobel edge filter.""" kind: Literal["sobel_filter"] = "sobel_filter" - mode: str = "reflect" + mode: FilterMode = "reflect" cval: float = 0.0 diff --git a/src/derzug/nodes/playaudio.py b/src/derzug/nodes/playaudio.py index cf29eda..00cfa5b 100644 --- a/src/derzug/nodes/playaudio.py +++ b/src/derzug/nodes/playaudio.py @@ -1,17 +1,40 @@ """The PlayAudio node: audition a 1D time patch, passing it through unchanged. -Playback is a display concern, so the parameters here describe the audition -only; the compiled workflow just forwards the patch. +Playback is a display concern, so the compiled workflow just forwards the +patch — but everything between a patch and device-ready PCM (validation, rate +inference, normalization, time-scaling, resampling) is plain signal +processing and lives here. The widget only adds the Qt sink and controls. """ from __future__ import annotations +from dataclasses import dataclass +from math import isfinite + import dascore as dc +import numpy as np from pydantic import BaseModel from derzug.nodes.spec import NodeSpec, PortSpec +from derzug.utils.sampling import strided_step from derzug.workflow.widget_tasks import PatchPassThroughTask +AUDIBLE_MIN_HZ = 20.0 +AUDIBLE_MAX_HZ = 20_000.0 +DEFAULT_TARGET_HZ = 4_000.0 +MIN_PLAYBACK_DURATION_S = 2.0 +MIN_OUTPUT_SAMPLE_RATE_HZ = 8_000.0 +MAX_OUTPUT_SAMPLE_RATE_HZ = 48_000.0 +MIN_TIME_SCALE = 1e-6 +MAX_TIME_SCALE = 1e6 +PCM_HEADROOM = 0.95 +PCM_NORMALIZE_PERCENTILE = 95.0 +DEFAULT_OUTPUT_GAIN_DB = 0.0 +#: Auto-gain calibration reads at most this many strided samples. +PCM_CALIBRATION_SAMPLES = 1_000_000 +#: Resampling and rate validation work in bounded blocks of this size. +RESAMPLE_BLOCK_SIZE = 500_000 + class PlayAudioParams(BaseModel): """Parameters for the PlayAudio node.""" @@ -20,6 +43,235 @@ class PlayAudioParams(BaseModel): volume_percent: int = 100 +@dataclass(frozen=True) +class PreparedAudio: + """Prepared patch playback metadata and PCM payload.""" + + native_rate_hz: float + pcm_bytes: bytes + sample_count: int + + +def coerce_playable_patch(patch: dc.Patch) -> dc.Patch: + """Return the squeezed patch shape used for playback validation/rendering.""" + return patch.squeeze() + + +def validate_patch_shape(patch: dc.Patch) -> None: + """Validate that the patch is a 1D time series.""" + data = np.asarray(patch.data) + if data.ndim != 1: + raise ValueError(f"expected a 1D patch, got shape {data.shape}") + if tuple(patch.dims) != ("time",): + raise ValueError(f"expected patch dims ('time',), got {patch.dims}") + + +def coord_to_seconds(coord: np.ndarray) -> np.ndarray: + """Convert time coordinates to seconds for rate inference.""" + arr = np.asarray(coord) + if np.issubdtype(arr.dtype, np.datetime64): + ns = arr.astype("datetime64[ns]").astype(np.int64) + return ns.astype(np.float64) / 1e9 + if np.issubdtype(arr.dtype, np.timedelta64): + ns = arr.astype("timedelta64[ns]").astype(np.int64) + return ns.astype(np.float64) / 1e9 + if np.issubdtype(arr.dtype, np.number): + return arr.astype(np.float64, copy=False) + raise ValueError("time coordinate must be numeric or datetime-like") + + +def infer_native_rate_hz(coord: np.ndarray) -> float: + """Infer the source sample rate from the patch time coordinate.""" + seconds = coord_to_seconds(coord) + if seconds.size < 2: + raise ValueError("time coordinate must contain at least two samples") + first = float(seconds[1] - seconds[0]) + if not isfinite(first): + raise ValueError("time coordinate must contain finite sample spacing") + if first <= 0: + raise ValueError("time coordinate must be strictly increasing") + tolerance = max(abs(first) * 1e-6, 1e-12) + # Check bounded slices so validating a long recording does not allocate + # a second full-length float64 difference array. + for start in range(1, seconds.size, RESAMPLE_BLOCK_SIZE): + stop = min(start + RESAMPLE_BLOCK_SIZE, seconds.size) + diffs = seconds[start:stop] - seconds[start - 1 : stop - 1] + if not np.all(np.isfinite(diffs)): + raise ValueError("time coordinate must contain finite sample spacing") + if np.any(diffs <= 0): + raise ValueError("time coordinate must be strictly increasing") + if not np.allclose(diffs, first, rtol=1e-6, atol=tolerance): + raise ValueError("time coordinate must have uniform sample spacing") + rate_hz = 1.0 / first + if not isfinite(rate_hz) or rate_hz <= 0: + raise ValueError("time coordinate must define a positive sample rate") + return rate_hz + + +def prepare_pcm_audio( + data: np.ndarray, + *, + output_gain_db: float = DEFAULT_OUTPUT_GAIN_DB, +) -> tuple[bytes, int]: + """Normalize mono samples with robust auto-gain and convert to PCM.""" + samples = np.array(data, dtype=np.float32, copy=True).reshape(-1) + if samples.size == 0: + raise ValueError("patch data is empty") + finite_mask = np.isfinite(samples) + if not np.any(finite_mask): + raise ValueError("patch data must contain at least one finite sample") + step = strided_step(samples.size, PCM_CALIBRATION_SAMPLES) + calibration = samples[::step] + finite_calibration = calibration[np.isfinite(calibration)] + if finite_calibration.size == 0: + finite_calibration = samples[np.argmax(finite_mask) :][:1] + nonzero = np.abs(finite_calibration) + ref = float(np.percentile(nonzero, PCM_NORMALIZE_PERCENTILE)) + if ref <= 0: + # The strided calibration subset can miss all signal energy + # (e.g. sparse spikes between stride points); fall back to the + # full-array peak so only truly silent data skips normalization. + ref = float(np.max(np.abs(samples[finite_mask]))) + if ref > 0: + samples *= PCM_HEADROOM / ref + linear_gain = float(10 ** (float(output_gain_db) / 20.0)) + samples[~finite_mask] = 0.0 + samples *= linear_gain + np.clip(samples, -PCM_HEADROOM, PCM_HEADROOM, out=samples) + samples *= np.iinfo(np.int16).max + np.rint(samples, out=samples) + pcm = samples.astype(" PreparedAudio: + """Validate the patch and prepare normalized PCM audio bytes.""" + validate_patch_shape(patch) + native_rate_hz = infer_native_rate_hz(np.asarray(patch.get_array("time"))) + pcm_bytes, sample_count = prepare_pcm_audio( + np.asarray(patch.data), + output_gain_db=output_gain_db, + ) + return PreparedAudio( + native_rate_hz=native_rate_hz, + pcm_bytes=pcm_bytes, + sample_count=sample_count, + ) + + +def default_time_scale(native_rate_hz: float, sample_count: int) -> float: + """Choose a default scale that moves the source into audible range.""" + if AUDIBLE_MIN_HZ <= native_rate_hz <= AUDIBLE_MAX_HZ: + scale = 1.0 + elif native_rate_hz <= 0: + scale = 1.0 + else: + scale = float(DEFAULT_TARGET_HZ / native_rate_hz) + if native_rate_hz > 0 and sample_count > 0: + duration_seconds = float(sample_count) / float(native_rate_hz) + min_duration_scale = duration_seconds / MIN_PLAYBACK_DURATION_S + if duration_seconds < MIN_PLAYBACK_DURATION_S: + scale = min(scale, min_duration_scale) + return float(np.clip(scale, MIN_TIME_SCALE, MAX_TIME_SCALE)) + + +def output_gain_db_from_volume_percent(volume_percent: int) -> float: + """Map a volume percentage onto a linear-gain dB value.""" + if volume_percent <= 0: + return -120.0 + return float(20.0 * np.log10(float(volume_percent) / 100.0)) + + +def effective_sample_rate_hz(native_rate_hz: float, time_scale: float) -> float: + """Return the playback rate after applying the configured time scale.""" + return float(native_rate_hz) * float(time_scale) + + +def playback_output_rate_hz(effective_rate_hz: float) -> int: + """Clamp the audio-device output rate to a broadly supported range.""" + if not isfinite(effective_rate_hz) or effective_rate_hz <= 0: + raise ValueError("effective sample rate must be positive") + return round( + float( + np.clip( + effective_rate_hz, + MIN_OUTPUT_SAMPLE_RATE_HZ, + MAX_OUTPUT_SAMPLE_RATE_HZ, + ) + ) + ) + + +def render_playback_pcm( + prepared: PreparedAudio, + *, + effective_rate_hz: float, + output_rate_hz: int, +) -> bytes: + """Render PCM, stretching or resampling when the sink rate is clamped.""" + if prepared.sample_count <= 0: + return b"" + if not isfinite(effective_rate_hz) or effective_rate_hz <= 0: + raise ValueError("effective sample rate must be positive") + if output_rate_hz <= 0: + raise ValueError("output sample rate must be positive") + if round(effective_rate_hz) == output_rate_hz: + return prepared.pcm_bytes + + source = np.frombuffer(prepared.pcm_bytes, dtype=" tuple[PreparedAudio, int, bytes]: + """Return ``(prepared, output_rate_hz, pcm)`` for one patch and parameters. + + The one-call headless form of what the widget does interactively: the + params' volume sets the normalization gain and the time scale sets the + playback rate, clamped to a device-supported output rate. + """ + params = PlayAudioParams() if params is None else params + playable = coerce_playable_patch(patch) + prepared = prepare_patch_audio( + playable, + output_gain_db=output_gain_db_from_volume_percent(int(params.volume_percent)), + ) + effective = effective_sample_rate_hz( + prepared.native_rate_hz, float(params.time_scale) + ) + output_rate = playback_output_rate_hz(effective) + pcm = render_playback_pcm( + prepared, effective_rate_hz=effective, output_rate_hz=output_rate + ) + return prepared, output_rate, pcm + + def playaudio_task_from_params( params: PlayAudioParams | None = None, ) -> PatchPassThroughTask: diff --git a/src/derzug/widgets/filter.py b/src/derzug/widgets/filter.py index 312e396..807536b 100644 --- a/src/derzug/widgets/filter.py +++ b/src/derzug/widgets/filter.py @@ -26,7 +26,9 @@ _FILTER_MODELS, _FILTER_NAMES, _PARAM_FIELDS, + MODE_OPTIONS, NODE_SPEC, + SAVGOL_MODE_OPTIONS, FilterParams, FilterTask, filter_settings_from_params, @@ -37,21 +39,8 @@ from derzug.utils.parsing import parse_patch_text_value from derzug.workflow import Task -_MODE_OPTIONS: tuple[str, ...] = ( - "reflect", - "constant", - "nearest", - "wrap", - "mirror", - "interp", -) -_SAVGOL_MODE_OPTIONS: tuple[str, ...] = ( - "mirror", - "constant", - "nearest", - "wrap", - "interp", -) +_MODE_OPTIONS = MODE_OPTIONS +_SAVGOL_MODE_OPTIONS = SAVGOL_MODE_OPTIONS class Filter(PatchDimWidget): diff --git a/src/derzug/widgets/playaudio.py b/src/derzug/widgets/playaudio.py index c0cdcd7..4a3b798 100644 --- a/src/derzug/widgets/playaudio.py +++ b/src/derzug/widgets/playaudio.py @@ -2,10 +2,8 @@ from __future__ import annotations -from dataclasses import dataclass from enum import Enum from importlib import import_module -from math import isfinite import dascore as dc import numpy as np @@ -24,7 +22,20 @@ from Orange.widgets.widget import Msg from derzug.core.zugwidget import ZugWidget -from derzug.nodes.playaudio import NODE_SPEC +from derzug.nodes.playaudio import ( + MAX_TIME_SCALE, + MIN_TIME_SCALE, + NODE_SPEC, + PreparedAudio, + coerce_playable_patch, + coord_to_seconds, + default_time_scale, + effective_sample_rate_hz, + output_gain_db_from_volume_percent, + playback_output_rate_hz, + prepare_patch_audio, + render_playback_pcm, +) from derzug.utils.display import format_display from derzug.utils.sampling import strided_step from derzug.workflow import Task @@ -106,32 +117,10 @@ def __init__(self, *_args, **_kwargs) -> None: raise RuntimeError("QtMultimedia is not available") -_AUDIBLE_MIN_HZ = 20.0 -_AUDIBLE_MAX_HZ = 20_000.0 -_DEFAULT_TARGET_HZ = 4_000.0 -_MIN_PLAYBACK_DURATION_S = 2.0 -_MIN_OUTPUT_SAMPLE_RATE_HZ = 8_000.0 -_MAX_OUTPUT_SAMPLE_RATE_HZ = 48_000.0 -_MIN_TIME_SCALE = 1e-6 -_MAX_TIME_SCALE = 1e6 -_PCM_HEADROOM = 0.95 -_PCM_NORMALIZE_PERCENTILE = 95.0 -_DEFAULT_OUTPUT_GAIN_DB = 0.0 _DEFAULT_VOLUME_PERCENT = 100 _MIN_VOLUME_PERCENT = 0 _MAX_VOLUME_PERCENT = 200 _MAX_WAVEFORM_PLOT_SAMPLES = 200_000 -_PCM_CALIBRATION_SAMPLES = 1_000_000 -_RESAMPLE_BLOCK_SIZE = 500_000 - - -@dataclass(frozen=True) -class _PreparedAudio: - """Prepared patch playback metadata and PCM payload.""" - - native_rate_hz: float - pcm_bytes: bytes - sample_count: int class PlayAudio(ZugWidget): @@ -169,7 +158,7 @@ def get_task(self) -> Task: def __init__(self) -> None: super().__init__() self._patch: dc.Patch | None = None - self._prepared_audio: _PreparedAudio | None = None + self._prepared_audio: PreparedAudio | None = None self._validation_error: str | None = None self._status_text = "No patch loaded" self._native_rate_hz: float | None = None @@ -189,7 +178,7 @@ def __init__(self) -> None: gui.widgetLabel(box, "Time scale:") self._time_scale_spin = QDoubleSpinBox(box) self._time_scale_spin.setDecimals(6) - self._time_scale_spin.setRange(_MIN_TIME_SCALE, _MAX_TIME_SCALE) + self._time_scale_spin.setRange(MIN_TIME_SCALE, MAX_TIME_SCALE) self._time_scale_spin.setStepType( QDoubleSpinBox.StepType.AdaptiveDecimalStepType ) @@ -276,8 +265,8 @@ def set_patch(self, patch: dc.Patch | None) -> None: self._playback_sample_index = None else: try: - playable_patch = self._coerce_playable_patch(patch) - self._prepared_audio = self._prepare_patch_audio( + playable_patch = coerce_playable_patch(patch) + self._prepared_audio = prepare_patch_audio( playable_patch, output_gain_db=self._current_output_gain_db(), ) @@ -359,7 +348,7 @@ def _apply_default_time_scale( """Set the default time scale for a newly received valid patch.""" self._syncing_time_scale = True try: - self.time_scale = self._default_time_scale(native_rate_hz, sample_count) + self.time_scale = default_time_scale(native_rate_hz, sample_count) self._time_scale_spin.setValue(self.time_scale) finally: self._syncing_time_scale = False @@ -383,8 +372,8 @@ def _on_volume_changed(self, *_args) -> None: self._update_volume_label() if self._patch is not None and self._validation_error is None: try: - playable_patch = self._coerce_playable_patch(self._patch) - self._prepared_audio = self._prepare_patch_audio( + playable_patch = coerce_playable_patch(self._patch) + self._prepared_audio = prepare_patch_audio( playable_patch, output_gain_db=self._current_output_gain_db(), ) @@ -404,14 +393,7 @@ def _update_volume_label(self) -> None: def _current_output_gain_db(self) -> float: """Return the current slider volume mapped to decibels.""" - return self._output_gain_db_from_volume_percent(int(self.volume_percent)) - - @staticmethod - def _output_gain_db_from_volume_percent(volume_percent: int) -> float: - """Map a UI volume percentage onto a linear-gain dB value.""" - if volume_percent <= 0: - return -120.0 - return float(20.0 * np.log10(float(volume_percent) / 100.0)) + return output_gain_db_from_volume_percent(int(self.volume_percent)) def _start_playback(self) -> None: """Start audio playback for the current prepared patch.""" @@ -421,10 +403,10 @@ def _start_playback(self) -> None: return self._stop_playback() effective_rate_hz = self._effective_sample_rate_hz() - sample_rate = self._playback_output_rate_hz(effective_rate_hz) + sample_rate = playback_output_rate_hz(effective_rate_hz) audio_format = self._build_audio_format(sample_rate) payload = QByteArray( - self._render_playback_pcm( + render_playback_pcm( prepared, effective_rate_hz=effective_rate_hz, output_rate_hz=sample_rate, @@ -519,7 +501,7 @@ def _effective_sample_rate_hz(self) -> float: """Return the current effective playback rate in Hz.""" if self._native_rate_hz is None: return float("nan") - return self._native_rate_hz * float(self.time_scale) + return effective_sample_rate_hz(self._native_rate_hz, float(self.time_scale)) def _set_waveform_data(self, patch: dc.Patch) -> None: """Cache waveform data for the plot using seconds relative to the start.""" @@ -527,7 +509,7 @@ def _set_waveform_data(self, patch: dc.Patch) -> None: step = strided_step(source_samples.size, _MAX_WAVEFORM_PLOT_SAMPLES) samples = source_samples[::step] time_coord = np.asarray(patch.get_array("time")) - time_seconds = self._coord_to_seconds(time_coord[::step]) + time_seconds = coord_to_seconds(time_coord[::step]) time_seconds = np.asarray(time_seconds, dtype=np.float64) if time_seconds.size: time_seconds = time_seconds - float(time_seconds[0]) @@ -599,78 +581,6 @@ def _update_playback_marker(self) -> None: sample_index = prepared.sample_count - 1 self._set_playback_marker(sample_index) - @staticmethod - def _default_time_scale(native_rate_hz: float, sample_count: int) -> float: - """Choose a default scale that moves the source into audible range.""" - if _AUDIBLE_MIN_HZ <= native_rate_hz <= _AUDIBLE_MAX_HZ: - scale = 1.0 - elif native_rate_hz <= 0: - scale = 1.0 - else: - scale = float(_DEFAULT_TARGET_HZ / native_rate_hz) - if native_rate_hz > 0 and sample_count > 0: - duration_seconds = float(sample_count) / float(native_rate_hz) - min_duration_scale = duration_seconds / _MIN_PLAYBACK_DURATION_S - if duration_seconds < _MIN_PLAYBACK_DURATION_S: - scale = min(scale, min_duration_scale) - return float(np.clip(scale, _MIN_TIME_SCALE, _MAX_TIME_SCALE)) - - @staticmethod - def _playback_output_rate_hz(effective_rate_hz: float) -> int: - """Clamp the audio-device output rate to a broadly supported range.""" - if not isfinite(effective_rate_hz) or effective_rate_hz <= 0: - raise ValueError("effective sample rate must be positive") - return round( - float( - np.clip( - effective_rate_hz, - _MIN_OUTPUT_SAMPLE_RATE_HZ, - _MAX_OUTPUT_SAMPLE_RATE_HZ, - ) - ) - ) - - @staticmethod - def _render_playback_pcm( - prepared: _PreparedAudio, - *, - effective_rate_hz: float, - output_rate_hz: int, - ) -> bytes: - """Render PCM, stretching or resampling when sink rate is clamped.""" - if prepared.sample_count <= 0: - return b"" - if not isfinite(effective_rate_hz) or effective_rate_hz <= 0: - raise ValueError("effective sample rate must be positive") - if output_rate_hz <= 0: - raise ValueError("output sample rate must be positive") - if round(effective_rate_hz) == output_rate_hz: - return prepared.pcm_bytes - - source = np.frombuffer(prepared.pcm_bytes, dtype=" QAudioFormat: """Build the mono PCM format used for playback.""" @@ -683,115 +593,3 @@ def _build_audio_format(sample_rate_hz: int) -> QAudioFormat: def _create_audio_sink(self, audio_format: QAudioFormat) -> QAudioSink: """Create a Qt audio sink for the requested output format.""" return QAudioSink(audio_format, self) - - @classmethod - def _prepare_patch_audio( - cls, - patch: dc.Patch, - *, - output_gain_db: float = _DEFAULT_OUTPUT_GAIN_DB, - ) -> _PreparedAudio: - """Validate the patch and prepare normalized PCM audio bytes.""" - cls._validate_patch_shape(patch) - native_rate_hz = cls._infer_native_rate_hz(np.asarray(patch.get_array("time"))) - pcm_bytes, sample_count = cls._prepare_pcm_audio( - np.asarray(patch.data), - output_gain_db=output_gain_db, - ) - return _PreparedAudio( - native_rate_hz=native_rate_hz, - pcm_bytes=pcm_bytes, - sample_count=sample_count, - ) - - @staticmethod - def _coerce_playable_patch(patch: dc.Patch) -> dc.Patch: - """Return the squeezed patch shape used for playback validation/rendering.""" - return patch.squeeze() - - @staticmethod - def _validate_patch_shape(patch: dc.Patch) -> None: - """Validate that the patch is a 1D time series.""" - data = np.asarray(patch.data) - if data.ndim != 1: - raise ValueError(f"expected a 1D patch, got shape {data.shape}") - if tuple(patch.dims) != ("time",): - raise ValueError(f"expected patch dims ('time',), got {patch.dims}") - - @staticmethod - def _infer_native_rate_hz(coord: np.ndarray) -> float: - """Infer the source sample rate from the patch time coordinate.""" - seconds = PlayAudio._coord_to_seconds(coord) - if seconds.size < 2: - raise ValueError("time coordinate must contain at least two samples") - first = float(seconds[1] - seconds[0]) - if not isfinite(first): - raise ValueError("time coordinate must contain finite sample spacing") - if first <= 0: - raise ValueError("time coordinate must be strictly increasing") - tolerance = max(abs(first) * 1e-6, 1e-12) - # Check bounded slices so validating a long recording does not allocate - # a second full-length float64 difference array. - for start in range(1, seconds.size, _RESAMPLE_BLOCK_SIZE): - stop = min(start + _RESAMPLE_BLOCK_SIZE, seconds.size) - diffs = seconds[start:stop] - seconds[start - 1 : stop - 1] - if not np.all(np.isfinite(diffs)): - raise ValueError("time coordinate must contain finite sample spacing") - if np.any(diffs <= 0): - raise ValueError("time coordinate must be strictly increasing") - if not np.allclose(diffs, first, rtol=1e-6, atol=tolerance): - raise ValueError("time coordinate must have uniform sample spacing") - rate_hz = 1.0 / first - if not isfinite(rate_hz) or rate_hz <= 0: - raise ValueError("time coordinate must define a positive sample rate") - return rate_hz - - @staticmethod - def _coord_to_seconds(coord: np.ndarray) -> np.ndarray: - """Convert time coordinates to seconds for rate inference.""" - arr = np.asarray(coord) - if np.issubdtype(arr.dtype, np.datetime64): - ns = arr.astype("datetime64[ns]").astype(np.int64) - return ns.astype(np.float64) / 1e9 - if np.issubdtype(arr.dtype, np.timedelta64): - ns = arr.astype("timedelta64[ns]").astype(np.int64) - return ns.astype(np.float64) / 1e9 - if np.issubdtype(arr.dtype, np.number): - return arr.astype(np.float64, copy=False) - raise ValueError("time coordinate must be numeric or datetime-like") - - @staticmethod - def _prepare_pcm_audio( - data: np.ndarray, - *, - output_gain_db: float = _DEFAULT_OUTPUT_GAIN_DB, - ) -> tuple[bytes, int]: - """Normalize mono samples with robust auto-gain and convert to PCM.""" - samples = np.array(data, dtype=np.float32, copy=True).reshape(-1) - if samples.size == 0: - raise ValueError("patch data is empty") - finite_mask = np.isfinite(samples) - if not np.any(finite_mask): - raise ValueError("patch data must contain at least one finite sample") - step = strided_step(samples.size, _PCM_CALIBRATION_SAMPLES) - calibration = samples[::step] - finite_calibration = calibration[np.isfinite(calibration)] - if finite_calibration.size == 0: - finite_calibration = samples[np.argmax(finite_mask) :][:1] - nonzero = np.abs(finite_calibration) - ref = float(np.percentile(nonzero, _PCM_NORMALIZE_PERCENTILE)) - if ref <= 0: - # The strided calibration subset can miss all signal energy - # (e.g. sparse spikes between stride points); fall back to the - # full-array peak so only truly silent data skips normalization. - ref = float(np.max(np.abs(samples[finite_mask]))) - if ref > 0: - samples *= _PCM_HEADROOM / ref - linear_gain = float(10 ** (float(output_gain_db) / 20.0)) - samples[~finite_mask] = 0.0 - samples *= linear_gain - np.clip(samples, -_PCM_HEADROOM, _PCM_HEADROOM, out=samples) - samples *= np.iinfo(np.int16).max - np.rint(samples, out=samples) - pcm = samples.astype(" dc.Patch | None: diff --git a/tests/test_nodes/test_playaudio.py b/tests/test_nodes/test_playaudio.py new file mode 100644 index 0000000..ac38f7e --- /dev/null +++ b/tests/test_nodes/test_playaudio.py @@ -0,0 +1,60 @@ +"""Tests for the Qt-free PlayAudio node DSP.""" + +from __future__ import annotations + +import dascore as dc +import numpy as np +import pytest +from derzug.nodes.playaudio import ( + PlayAudioParams, + playback_output_rate_hz, + render_audition, +) + + +def _one_d_time_patch() -> dc.Patch: + """Return a 1D time patch suitable for audition.""" + return dc.get_example_patch("example_event_2").mean("distance").squeeze() + + +class TestRenderAudition: + """The params-driven audition renderer works without any widget.""" + + def test_default_params_produce_pcm(self): + """A playable patch renders non-empty PCM at a supported rate.""" + patch = _one_d_time_patch() + + prepared, output_rate, pcm = render_audition(patch) + + assert prepared.sample_count == np.asarray(patch.data).size + assert output_rate == playback_output_rate_hz(prepared.native_rate_hz * 1.0) + assert len(pcm) > 0 + assert len(pcm) % 2 == 0 # 16-bit mono PCM + + def test_time_scale_changes_output_rate(self): + """The params' time scale drives the playback rate.""" + patch = _one_d_time_patch() + + _, slow_rate, _ = render_audition(patch, PlayAudioParams(time_scale=1.0)) + prepared, fast_rate, _ = render_audition( + patch, PlayAudioParams(time_scale=100.0) + ) + + assert fast_rate == playback_output_rate_hz(prepared.native_rate_hz * 100.0) + assert fast_rate >= slow_rate + + def test_volume_scales_amplitude(self): + """A lower volume percent renders quieter PCM.""" + patch = _one_d_time_patch() + + _, _, loud = render_audition(patch, PlayAudioParams(volume_percent=100)) + _, _, quiet = render_audition(patch, PlayAudioParams(volume_percent=10)) + + loud_peak = np.abs(np.frombuffer(loud, dtype="