"""Timestamp-based labeling for captured recordings.""" from __future__ import annotations from typing import Optional from ria_toolkit_oss.data.recording import Recording from .campaign import CaptureStep def label_recording( recording: Recording, device_id: str, step: CaptureStep, capture_timestamp: float, campaign_name: Optional[str] = None, ) -> Recording: """Apply device identity and capture configuration labels to a recording's metadata. Labels are stored in the ``ria:*`` namespace when the recording is saved as SigMF, via the existing ``update_metadata`` mechanism. Args: recording: The recording to label. device_id: Identifier for the transmitting device (e.g. "iphone13_wifi_001"). step: The capture step that was active during this recording. capture_timestamp: Unix timestamp (float) of when capture started. campaign_name: Optional campaign name for cross-recording reference. Returns: The same recording with updated metadata. """ recording.update_metadata("device_id", device_id) recording.update_metadata("capture_timestamp", capture_timestamp) recording.update_metadata("step_label", step.label) recording.update_metadata("step_duration_s", step.duration) if campaign_name: recording.update_metadata("campaign", campaign_name) # WiFi-specific labels if step.channel is not None: recording.update_metadata("wifi_channel", step.channel) if step.bandwidth_mhz is not None: recording.update_metadata("wifi_bandwidth_mhz", step.bandwidth_mhz) # Bluetooth-specific labels if step.connection_interval_ms is not None: recording.update_metadata("bt_connection_interval_ms", step.connection_interval_ms) # Traffic pattern (WiFi + BT) if step.traffic is not None: recording.update_metadata("traffic_pattern", step.traffic) # TX power if step.power_dbm is not None: recording.update_metadata("tx_power_dbm", step.power_dbm) return recording def build_output_filename(device_id: str, step: CaptureStep) -> str: """Generate a deterministic filename for a labeled recording. Format: ``/`` Args: device_id: Device identifier string. step: Capture step. Returns: Relative path string (no extension) to use as ``filename`` in ``to_sigmf()``. """ safe_id = device_id.replace("/", "_").replace(" ", "_") safe_label = step.label.replace("/", "_").replace(" ", "_") return f"{safe_id}/{safe_label}"