Ryan Malloy
8da486719a
Textual's ContentSwitcher expects regular Widget/Container children, not Screen subclasses. Screen's on_mount fires for all children at once regardless of visibility, so demo workers for all 5 modes started simultaneously and competed for the bridge. Container children get proper on_show/on_hide from ContentSwitcher's visibility toggling — only the active panel's worker runs.
221 lines
7.7 KiB
Python
221 lines
7.7 KiB
Python
"""L-Band screen — direct input spectrum analyzer with allocation annotations.
|
|
|
|
Same sweep mechanics as the spectrum screen, but with LNB disabled (direct input)
|
|
and band allocation overlays showing what service each frequency range belongs to.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "tools"))
|
|
|
|
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_lib import LBAND_ALLOCATIONS
|
|
|
|
from skywalker_tui.widgets.spectrum_plot import SpectrumPlot
|
|
from skywalker_tui.widgets.waterfall import WaterfallDisplay
|
|
|
|
|
|
def _alloc_table(start: float, stop: float) -> str:
|
|
"""Build a Rich-markup allocation reference for the visible range."""
|
|
lines = ["[#00d4aa bold]L-Band Allocations in range:[/]"]
|
|
colors = ["#60a0c0", "#80b060", "#c0a050", "#a06080", "#50a0a0", "#a08060", "#6080a0"]
|
|
for i, (lo, hi, name) in enumerate(LBAND_ALLOCATIONS):
|
|
if lo < stop and hi > start:
|
|
overlap_lo = max(lo, start)
|
|
overlap_hi = min(hi, stop)
|
|
c = colors[i % len(colors)]
|
|
lines.append(f" [{c}]{overlap_lo:.0f}-{overlap_hi:.0f} MHz {name}[/]")
|
|
if len(lines) == 1:
|
|
lines.append(" [#506878](none in range)[/]")
|
|
return "\n".join(lines)
|
|
|
|
|
|
class LBandScreen(Container):
|
|
"""L-band direct input analyzer with allocation annotations."""
|
|
|
|
DEFAULT_CSS = """
|
|
LBandScreen {
|
|
layout: vertical;
|
|
}
|
|
LBandScreen #lband-main {
|
|
height: 1fr;
|
|
layout: horizontal;
|
|
}
|
|
LBandScreen #lband-plot-col {
|
|
width: 2fr;
|
|
layout: vertical;
|
|
}
|
|
LBandScreen #lband-info-col {
|
|
width: 1fr;
|
|
padding: 1;
|
|
background: #0e1420;
|
|
border-left: solid #1a2a3a;
|
|
layout: vertical;
|
|
}
|
|
LBandScreen #lband-alloc-panel {
|
|
height: auto;
|
|
padding: 1;
|
|
}
|
|
LBandScreen #lband-progress {
|
|
height: 3;
|
|
layout: horizontal;
|
|
padding: 0 2;
|
|
background: #0e1018;
|
|
}
|
|
LBandScreen #lband-progress Static {
|
|
width: auto;
|
|
margin: 1 1 0 0;
|
|
}
|
|
LBandScreen #lband-progress ProgressBar {
|
|
width: 1fr;
|
|
margin: 1 1 0 0;
|
|
}
|
|
LBandScreen #lband-controls {
|
|
height: auto;
|
|
padding: 1 2;
|
|
background: #0e1018;
|
|
border-top: solid #1a2a3a;
|
|
layout: horizontal;
|
|
}
|
|
LBandScreen #lband-controls Label {
|
|
width: auto;
|
|
margin: 1 1 0 0;
|
|
color: #506878;
|
|
}
|
|
LBandScreen #lband-controls Input {
|
|
width: 10;
|
|
margin: 0 1;
|
|
}
|
|
LBandScreen #lband-controls Button {
|
|
margin: 0 1;
|
|
}
|
|
"""
|
|
|
|
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 Horizontal(id="lband-main"):
|
|
with Vertical(id="lband-plot-col"):
|
|
yield SpectrumPlot(title="L-Band Spectrum (Direct Input)",
|
|
id="lband-plot")
|
|
yield WaterfallDisplay(title="Waterfall", id="lband-waterfall")
|
|
with Vertical(id="lband-info-col"):
|
|
yield Static(_alloc_table(950, 2150), id="lband-alloc-panel")
|
|
|
|
with Horizontal(id="lband-progress"):
|
|
yield Static("[#506878]Ready[/]", id="lband-status")
|
|
yield ProgressBar(total=100, show_eta=False, id="lband-pbar")
|
|
|
|
with Horizontal(id="lband-controls"):
|
|
yield Label("Start:")
|
|
yield Input("950", id="lband-start")
|
|
yield Label("Stop:")
|
|
yield Input("2150", id="lband-stop")
|
|
yield Label("Step:")
|
|
yield Input("2", id="lband-step")
|
|
yield Label("Dwell:")
|
|
yield Input("20", id="lband-dwell")
|
|
yield Button("23cm", id="lband-23cm-btn")
|
|
yield Button("Sweep", id="lband-sweep-btn", variant="success")
|
|
yield Button("Stop", id="lband-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 == "lband-sweep-btn":
|
|
self._start_sweep()
|
|
elif event.button.id == "lband-stop-btn":
|
|
self._stop_sweep()
|
|
elif event.button.id == "lband-23cm-btn":
|
|
self.query_one("#lband-start", Input).value = "1240"
|
|
self.query_one("#lband-stop", Input).value = "1300"
|
|
self.query_one("#lband-step", Input).value = "0.5"
|
|
# Update allocation display
|
|
self.query_one("#lband-alloc-panel", Static).update(
|
|
_alloc_table(1240, 1300)
|
|
)
|
|
|
|
def _start_sweep(self) -> None:
|
|
if self._sweeping:
|
|
return
|
|
self._sweeping = True
|
|
|
|
start = float(self.query_one("#lband-start", Input).value or "950")
|
|
stop = float(self.query_one("#lband-stop", Input).value or "2150")
|
|
step = float(self.query_one("#lband-step", Input).value or "2")
|
|
dwell = int(self.query_one("#lband-dwell", Input).value or "20")
|
|
|
|
# Update allocation panel for current range
|
|
self.query_one("#lband-alloc-panel", Static).update(
|
|
_alloc_table(start, stop)
|
|
)
|
|
|
|
self._sweep_worker = self._do_sweep(start, stop, step, dwell)
|
|
|
|
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) -> None:
|
|
"""L-band sweep with LNB disabled."""
|
|
try:
|
|
self._bridge.ensure_booted()
|
|
# Disable LNB for direct input
|
|
self._bridge.configure_lnb(disable_lnb=True)
|
|
except Exception:
|
|
pass
|
|
|
|
def progress_cb(freq, step_num, total, result):
|
|
pct = (step_num + 1) / total * 100
|
|
self.app.call_from_thread(self._update_progress, pct, freq)
|
|
|
|
self.app.call_from_thread(self._set_status, "Sweeping...")
|
|
|
|
freqs, powers, results = self._bridge.sweep_spectrum(
|
|
start, stop, step, dwell, sr_ksps=20000, callback=progress_cb,
|
|
)
|
|
|
|
self.app.call_from_thread(self._show_results, freqs, powers, results)
|
|
self._sweeping = False
|
|
|
|
def _update_progress(self, pct: float, freq: float) -> None:
|
|
if not self.is_mounted:
|
|
return
|
|
self.query_one("#lband-pbar", ProgressBar).update(progress=pct)
|
|
self.query_one("#lband-status", Static).update(
|
|
f"[#00d4aa]{freq:.1f} MHz[/]"
|
|
)
|
|
|
|
def _set_status(self, msg: str) -> None:
|
|
if not self.is_mounted:
|
|
return
|
|
self.query_one("#lband-status", Static).update(f"[#506878]{msg}[/]")
|
|
|
|
def _show_results(self, freqs, powers, results) -> None:
|
|
if not self.is_mounted:
|
|
return
|
|
self.query_one("#lband-plot", SpectrumPlot).update_data(
|
|
freqs, powers, results, lnb_lo=0.0,
|
|
)
|
|
self.query_one("#lband-waterfall", WaterfallDisplay).add_sweep(powers)
|
|
self.query_one("#lband-status", Static).update("[#506878]Complete[/]")
|
|
self.query_one("#lband-pbar", ProgressBar).update(progress=100)
|