Harbor Radio Signals

1"""Harbor Signals media pipeline built with Pixeltable.
2
3Ingest short VHF/MF/HF radio clips with a cover still and a transcript. Pixeltable
4derives audio metadata, peak-normalized / faded / volume-scaled copies, draws
5annotated cover boxes via vision helpers, and fans the transcript into a
6sentence view with StringSplitter. UDFs tag band lane, callsign, and duration.
7
8Layout: UDFs live in udfs.py; this file holds table models and @pxt.query functions.
9No HTTP router — drive the pipeline with `python run.py`.
10
11Run locally:
12    pxt schema update app.py harbor
13    python run.py
14"""
15from __future__ import annotations
16
17import pixeltable as pxt
18import pixeltable.functions as pxtf
19
20from udfs import band_lane, callsign_tag, duration_bucket, transcript_word_count
21
22TableModel = pxt.model_base()
23
24
25class Signals(TableModel, name='signals', has_default_idxs=False):
26    """One harbor radio transmission: clip + cover still + transcript."""
27
28    signal_id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
29    callsign: pxt.String
30    harbor: pxt.String
31    band: pxt.String  # vhf / mf / hf
32    transcript: pxt.String
33    clip: pxt.Audio
34    cover: pxt.Image
35    # ground-truth and predicted boxes on the cover (xyxy ints)
36    gt_boxes: pxt.Json  # list[list[int]]
37    pred_boxes: pxt.Json
38    box_labels: pxt.Json  # list[str]
39
40    # --- fn.audio primary ---
41    meta = pxtf.audio.get_metadata(clip)
42    normalized = pxtf.audio.normalize(clip)
43    faded_in = pxtf.audio.fade_in(normalized, duration=0.25)
44    faded = pxtf.audio.fade_out(faded_in, duration=0.25)
45    quiet = pxtf.audio.multiply_volume(clip, factor=0.4)
46
47    # duration from metadata (seconds); nested json path is best-effort
48    duration_s: pxt.Float | None
49
50    # --- fn.vision secondary ---
51    annotated = pxtf.vision.bboxes_draw(
52        cover,
53        pred_boxes,
54        labels=box_labels,
55        color='lime',
56        width=2,
57    )
58    padded_pred = pxtf.vision.bboxes_pad(pred_boxes, 'xyxy', x=4, y=4)
59
60    # UDF chain
61    lane = band_lane(band)
62    tag = callsign_tag(callsign, harbor, band)
63    words = transcript_word_count(transcript)
64    dur_bucket = duration_bucket(duration_s)
65
66    __indexes__ = [
67        pxt.BtreeIndex(harbor),
68        pxt.BtreeIndex(band),
69        pxt.BtreeIndex(lane),
70    ]
71
72
73class Sentences(
74    TableModel,
75    name='sentences',
76    base=Signals,
77    iterator=pxtf.string.string_splitter(Signals.transcript, separators='sentence'),
78):
79    """Component view: one row per sentence from each signal transcript."""
80
81    # 'text' is an output column of string_splitter
82    length = pxtf.string.len(text)  # type: ignore[name-defined]  # noqa: F821
83    upper = pxtf.string.upper(text)  # type: ignore[name-defined]  # noqa: F821
84
85
86@pxt.query
87def by_harbor(harbor: str):
88    """Signals for one harbor (harbor index)."""
89    return (
90        Signals.where(Signals.harbor == harbor)
91        .select(
92            Signals.signal_id,
93            Signals.callsign,
94            Signals.band,
95            Signals.tag,
96            Signals.lane,
97            Signals.words,
98            Signals.dur_bucket,
99        )
100        .order_by(Signals.callsign)
101    )
102
103
104@pxt.query
105def by_lane(lane: str):
106    """Signals in an ops lane (lane index)."""
107    return (
108        Signals.where(Signals.lane == lane)
109        .select(Signals.signal_id, Signals.callsign, Signals.harbor, Signals.band, Signals.tag)
110        .order_by(Signals.harbor)
111    )
112
113
114@pxt.query
115def long_sentences(min_len: int):
116    """Sentences at least min_len characters (StringSplitter view)."""
117    return (
118        Sentences.where(Sentences.length >= min_len)
119        .select(Sentences.callsign, Sentences.harbor, Sentences.text, Sentences.length)
120        .order_by(Sentences.length, asc=False)
121    )