"""Spectrum screen — sweep analyzer across the IF range. Displays a bar chart of power vs. frequency, optionally with a rolling waterfall beneath it. Uses threaded workers for the blocking USB sweep. """ from textual.app import ComposeResult from textual.containers import Container, Horizontal, Vertical from textual.widgets import Label, Input, Button, Static, ProgressBar, Checkbox from textual import work from textual.worker import Worker from skywalker_tui.widgets.spectrum_plot import SpectrumPlot from skywalker_tui.widgets.waterfall import WaterfallDisplay class SpectrumScreen(Container): """Spectrum analyzer with bar chart and optional waterfall.""" DEFAULT_CSS = """ SpectrumScreen { layout: vertical; } SpectrumScreen #spec-main { height: 1fr; layout: vertical; } SpectrumScreen #spec-progress-row { height: 3; layout: horizontal; padding: 0 2; background: #0e1018; } SpectrumScreen #spec-progress-row Static { width: auto; margin: 1 1 0 0; } SpectrumScreen #spec-progress-row ProgressBar { width: 1fr; margin: 1 1 0 0; } SpectrumScreen #spec-controls { height: auto; padding: 1 2; background: #0e1018; border-top: solid #1a2a3a; layout: horizontal; } SpectrumScreen #spec-controls Label { width: auto; margin: 1 1 0 0; color: #506878; } SpectrumScreen #spec-controls Input { width: 10; margin: 0 1; } SpectrumScreen #spec-controls Button { margin: 0 1; } SpectrumScreen #spec-controls Checkbox { margin: 1 1 0 0; } """ def __init__(self, bridge, **kwargs): super().__init__(**kwargs) self._bridge = bridge self._sweeping = False self._sweep_worker: Worker | None = None def compose(self) -> ComposeResult: with Vertical(id="spec-main"): yield SpectrumPlot(title="Spectrum Analyzer", id="spec-plot") yield WaterfallDisplay(title="Waterfall", id="spec-waterfall") with Horizontal(id="spec-progress-row"): yield Static("[#506878]Ready[/]", id="spec-status") yield ProgressBar(total=100, show_eta=False, id="spec-pbar") with Horizontal(id="spec-controls"): yield Label("Start:") yield Input("950", id="spec-start") yield Label("Stop:") yield Input("2150", id="spec-stop") yield Label("Step:") yield Input("5", id="spec-step") yield Label("Dwell:") yield Input("10", id="spec-dwell") yield Label("LNB LO:") yield Input("0", id="spec-lnb") yield Checkbox("Continuous", id="spec-continuous") yield Button("Sweep", id="spec-sweep-btn", variant="success") yield Button("Stop", id="spec-stop-btn", variant="error") def on_show(self) -> None: if self._bridge.is_demo and not self._sweeping: self._start_sweep() def on_hide(self) -> None: self._stop_sweep() def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "spec-sweep-btn": self._start_sweep() elif event.button.id == "spec-stop-btn": self._stop_sweep() def _start_sweep(self) -> None: if self._sweeping: return # Validate inputs try: start = float(self.query_one("#spec-start", Input).value or "950") stop = float(self.query_one("#spec-stop", Input).value or "2150") step = float(self.query_one("#spec-step", Input).value or "5") dwell = int(float(self.query_one("#spec-dwell", Input).value or "10")) lnb_lo = float(self.query_one("#spec-lnb", Input).value or "0") except ValueError: self._update_status("Invalid input — check numeric fields") return if not (950 <= start <= 2150) or not (950 <= stop <= 2150): self._update_status("Frequency out of range (950–2150 MHz)") return if start >= stop: self._update_status("Start must be less than Stop") return if not (0.1 <= step <= 500): self._update_status("Step out of range (0.1–500 MHz)") return continuous = self.query_one("#spec-continuous", Checkbox).value self._sweeping = True self._sweep_worker = self._do_sweep(start, stop, step, dwell, lnb_lo, continuous) def _stop_sweep(self) -> None: self._sweeping = False if self._sweep_worker: self._sweep_worker.cancel() self._sweep_worker = None @work(thread=True) def _do_sweep(self, start: float, stop: float, step: float, dwell: int, lnb_lo: float, continuous: bool) -> None: """Background sweep worker.""" import time try: self._bridge.ensure_booted() except Exception: pass sweep_num = 0 while self._sweeping: sweep_num += 1 step_count = [0] def progress_cb(freq, step_num, total, result): step_count[0] = step_num + 1 pct = (step_num + 1) / total * 100 self.app.call_from_thread(self._update_progress, pct, freq, sweep_num) self.app.call_from_thread(self._update_status, f"Sweeping #{sweep_num}...") freqs, powers, results = self._bridge.sweep_spectrum( start, stop, step, dwell, sr_ksps=20000, callback=progress_cb, ) self.app.call_from_thread(self._update_plot, freqs, powers, results, lnb_lo) self.app.call_from_thread(self._update_waterfall, powers) if not continuous: break time.sleep(0.1) self._sweeping = False self.app.call_from_thread(self._update_status, "Complete") def _update_progress(self, pct: float, freq: float, sweep_num: int) -> None: if not self.is_mounted: return self.query_one("#spec-pbar", ProgressBar).update(progress=pct) self.query_one("#spec-status", Static).update( f"[#00d4aa]Sweep #{sweep_num}[/] [#506878]{freq:.0f} MHz[/]" ) def _update_status(self, msg: str) -> None: if not self.is_mounted: return self.query_one("#spec-status", Static).update(f"[#506878]{msg}[/]") self.query_one("#spec-pbar", ProgressBar).update(progress=0) def _update_plot(self, freqs, powers, results, lnb_lo) -> None: if not self.is_mounted: return self.query_one("#spec-plot", SpectrumPlot).update_data( freqs, powers, results, lnb_lo=lnb_lo, ) def _update_waterfall(self, powers) -> None: if not self.is_mounted: return self.query_one("#spec-waterfall", WaterfallDisplay).add_sweep(powers)