44c98b11e3
Pressing `l` during a scan locked a frequency out for that run only, so the same birdie had to be locked out again on every later one. It now writes back to the settings file the run started from -- only that one key, since a scan's config also holds whatever was passed on the command line for this run and saving all of it would quietly make those permanent. The file is read, its lock-outs replaced, the rest left as it was. `save_lockouts` turns it off for anyone who would rather their config were never touched. Lock-outs were also single frequencies only. They are now a list of frequencies and spans -- "162.55M, 450M-455M, 88M to 108M" -- which is what a pager band or a noisy stretch of spectrum actually is. A point is still widened by the lock-out width; a span is taken exactly as written, because whoever typed it already said how wide it is. The scanner matched lock-outs by rounding a frequency into a bucket of the lock-out width, which cannot express a span and was never exact at the edges. It now holds intervals and tests them directly. Settings files that predate this hold a bare number per lock-out, and still mean the same thing: Lockout.coerce takes numbers, strings, pairs and dicts, so old profiles load untouched. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
418 lines
15 KiB
Python
Executable file
418 lines
15 KiB
Python
Executable file
"""Frequency ranges, parsing, and turning them into a sweep plan."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import re
|
|
from dataclasses import asdict, dataclass
|
|
|
|
from .bandplan import BandPreset, by_key, fmt_hz, presets_covering
|
|
|
|
__all__ = ["ScanRange", "TuneStep", "Lockout", "parse_frequency", "parse_range",
|
|
"parse_range_list", "parse_lockout_list", "build_plan", "fmt_hz",
|
|
"RangeError"]
|
|
|
|
|
|
class RangeError(ValueError):
|
|
"""Raised for un-parseable user frequency input."""
|
|
|
|
|
|
_SUFFIX = {
|
|
"": 1.0, "hz": 1.0,
|
|
"k": 1e3, "khz": 1e3,
|
|
"m": 1e6, "mhz": 1e6,
|
|
"g": 1e9, "ghz": 1e9,
|
|
}
|
|
|
|
_NUM = re.compile(r"^\s*([0-9]*\.?[0-9]+)\s*([a-zA-Z]*)\s*$")
|
|
|
|
|
|
def parse_frequency(text: str, default_unit: str = "") -> float:
|
|
"""Parse ``146.52M``, ``146520000``, ``433.92 MHz``, ``14074k`` -> Hz.
|
|
|
|
A bare number with no suffix is read as MHz when it is small enough to be
|
|
unambiguous (under 10000), otherwise as Hz -- which is how people
|
|
actually write frequencies.
|
|
"""
|
|
if text is None:
|
|
raise RangeError("empty frequency")
|
|
s = str(text).strip().replace(",", "").replace("_", "")
|
|
if not s:
|
|
raise RangeError("empty frequency")
|
|
m = _NUM.match(s)
|
|
if not m:
|
|
raise RangeError(f"cannot read {text!r} as a frequency")
|
|
value, suffix = float(m.group(1)), m.group(2).lower()
|
|
|
|
if not suffix:
|
|
suffix = default_unit.lower()
|
|
if not suffix:
|
|
# No unit anywhere: guess from magnitude.
|
|
return value * 1e6 if value < 10_000 else value
|
|
if suffix not in _SUFFIX:
|
|
raise RangeError(f"unknown frequency unit {m.group(2)!r} in {text!r}")
|
|
return value * _SUFFIX[suffix]
|
|
|
|
|
|
@dataclass
|
|
class ScanRange:
|
|
"""One start/stop pair the user wants swept."""
|
|
|
|
start: float
|
|
stop: float
|
|
step: float = 12_500.0
|
|
mode: str = "auto"
|
|
bandwidth: float = 0.0
|
|
label: str = ""
|
|
preset_key: str = ""
|
|
enabled: bool = True
|
|
threshold_db: float | None = None
|
|
|
|
def __post_init__(self):
|
|
self.start = float(self.start)
|
|
self.stop = float(self.stop)
|
|
if self.stop < self.start:
|
|
self.start, self.stop = self.stop, self.start
|
|
if self.stop == self.start:
|
|
# A single frequency: give it enough width to actually measure.
|
|
pad = max(self.step, 12_500.0) / 2.0
|
|
self.start -= pad
|
|
self.stop += pad
|
|
if self.step <= 0:
|
|
self.step = 12_500.0
|
|
if not self.label:
|
|
self.label = f"{fmt_hz(self.start)}-{fmt_hz(self.stop)}"
|
|
|
|
@property
|
|
def span(self) -> float:
|
|
return self.stop - self.start
|
|
|
|
@classmethod
|
|
def from_preset(cls, preset: BandPreset, **over) -> "ScanRange":
|
|
kw = dict(start=preset.start, stop=preset.stop, step=preset.step,
|
|
mode=preset.mode, bandwidth=preset.bandwidth,
|
|
label=preset.name, preset_key=preset.key)
|
|
kw.update(over)
|
|
return cls(**kw)
|
|
|
|
def _covering(self, hz: float, need_bandwidth: bool = False):
|
|
"""Band-plan segments to resolve against at one frequency.
|
|
|
|
A range that came from a band-plan preset resolves against segments of
|
|
the same service first. Without that, scanning the whole of 70 cm
|
|
would hand 433-435 MHz to the ISM preset that overlaps it there, and
|
|
an amateur repeater would be treated as an unlicensed device.
|
|
"""
|
|
covering = [p for p in presets_covering(hz)
|
|
if (p.bandwidth > 0 if need_bandwidth else p.mode != "auto")]
|
|
own = by_key(self.preset_key) if self.preset_key else None
|
|
if own is not None:
|
|
same = [p for p in covering if p.category == own.category]
|
|
if same:
|
|
return same
|
|
return covering
|
|
|
|
def resolved_mode(self, freq: float | None = None) -> str:
|
|
"""The demodulator to use, resolving ``auto`` against the band plan."""
|
|
if self.mode and self.mode != "auto":
|
|
return self.mode
|
|
hz = freq if freq is not None else 0.5 * (self.start + self.stop)
|
|
# Whole-band entries are skipped: they defer to the segments beneath
|
|
# them, so resolving to one would hand "auto" straight back.
|
|
covering = self._covering(hz)
|
|
if covering:
|
|
return min(covering, key=lambda p: p.span).mode
|
|
return "nfm"
|
|
|
|
def resolved_bandwidth(self, freq: float | None = None) -> float:
|
|
if self.bandwidth > 0:
|
|
return self.bandwidth
|
|
hz = freq if freq is not None else 0.5 * (self.start + self.stop)
|
|
covering = self._covering(hz, need_bandwidth=True)
|
|
if covering:
|
|
return min(covering, key=lambda p: p.span).bandwidth
|
|
return max(self.step, 12_500.0)
|
|
|
|
def to_dict(self) -> dict:
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: dict) -> "ScanRange":
|
|
known = {f for f in cls.__dataclass_fields__}
|
|
return cls(**{k: v for k, v in d.items() if k in known})
|
|
|
|
def describe(self) -> str:
|
|
state = "" if self.enabled else " (disabled)"
|
|
return (f"{self.label}: {fmt_hz(self.start)} - {fmt_hz(self.stop)} "
|
|
f"step {fmt_hz(self.step)} mode {self.mode}{state}")
|
|
|
|
|
|
_RANGE_SPLIT = re.compile(r"\s*(?:-{1,2}|\.{2}|to|:{2})\s*", re.I)
|
|
|
|
|
|
def parse_range(text: str) -> ScanRange:
|
|
"""Parse one range specification.
|
|
|
|
Accepted forms::
|
|
|
|
144M-148M explicit start and stop
|
|
144-148M unit carried over to both ends
|
|
146.52M a single frequency
|
|
144M-148M/25k with an explicit step
|
|
144M-148M/25k@nfm with a step and a demodulator
|
|
gmrs any band-plan preset key
|
|
|
|
"""
|
|
if text is None:
|
|
raise RangeError("empty range")
|
|
s = str(text).strip()
|
|
if not s:
|
|
raise RangeError("empty range")
|
|
|
|
mode = "auto"
|
|
if "@" in s:
|
|
s, _, mode = s.partition("@")
|
|
mode = mode.strip().lower() or "auto"
|
|
|
|
step = None
|
|
if "/" in s:
|
|
s, _, step_s = s.partition("/")
|
|
step = parse_frequency(step_s, default_unit="khz")
|
|
|
|
s = s.strip()
|
|
preset = by_key(s)
|
|
if preset is not None and preset.is_group:
|
|
raise RangeError(
|
|
f"{preset.key!r} covers {len(preset.expand())} separate ranges; "
|
|
"it can only be used where a list of ranges is accepted")
|
|
if preset is not None:
|
|
r = ScanRange.from_preset(preset)
|
|
if step:
|
|
r.step = step
|
|
if mode != "auto":
|
|
r.mode = mode
|
|
return r
|
|
|
|
parts = [p for p in _RANGE_SPLIT.split(s) if p.strip()]
|
|
if len(parts) == 1:
|
|
hz = parse_frequency(parts[0])
|
|
r = ScanRange(hz, hz, step or 12_500.0, mode)
|
|
elif len(parts) == 2:
|
|
# "144-148M": the unit on the right end applies to the left too.
|
|
right_unit = _NUM.match(parts[1])
|
|
unit = right_unit.group(2) if right_unit else ""
|
|
lo = parse_frequency(parts[0], default_unit=unit)
|
|
hi = parse_frequency(parts[1])
|
|
r = ScanRange(lo, hi, step or 12_500.0, mode)
|
|
else:
|
|
raise RangeError(f"cannot read {text!r} as a frequency range")
|
|
return r
|
|
|
|
|
|
def parse_range_list(text: str | list[str]) -> list[ScanRange]:
|
|
"""Parse a comma/newline separated list into ranges (unlimited count).
|
|
|
|
A band-plan key that stands for a set of others -- ``all-cw``, say --
|
|
expands here into one range per member.
|
|
"""
|
|
if isinstance(text, (list, tuple)):
|
|
items = list(text)
|
|
else:
|
|
items = re.split(r"[,\n;]+", str(text))
|
|
out = []
|
|
for item in items:
|
|
item = item.strip()
|
|
if not item:
|
|
continue
|
|
preset = by_key(item)
|
|
if preset is not None and preset.is_group:
|
|
out.extend(ScanRange.from_preset(m) for m in preset.expand())
|
|
else:
|
|
out.append(parse_range(item))
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sweep planning
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class TuneStep:
|
|
"""One tuner position, plus the slice of its spectrum we trust."""
|
|
|
|
center: float
|
|
low: float # first frequency this step is responsible for
|
|
high: float # last frequency this step is responsible for
|
|
range_index: int
|
|
range_label: str = ""
|
|
mode: str = "auto"
|
|
index: int = 0 # position in the plan; keys the noise-floor memory
|
|
|
|
@property
|
|
def width(self) -> float:
|
|
return self.high - self.low
|
|
|
|
def describe(self) -> str:
|
|
return f"{fmt_hz(self.center)} covering {fmt_hz(self.low)}-{fmt_hz(self.high)}"
|
|
|
|
|
|
def build_plan(ranges: list[ScanRange], sample_rate: float,
|
|
usable_fraction: float = 0.75,
|
|
dc_guard: float = 8_000.0,
|
|
max_steps: int = 200_000,
|
|
tunable=None) -> list[TuneStep]:
|
|
"""Split every enabled range into tuner steps.
|
|
|
|
The tuner is deliberately *not* centred on the span it covers. An RTL2832
|
|
always shows a DC spike at whatever it is tuned to, so a centred step would
|
|
blank a hole in the middle of every range -- and for a range narrower than
|
|
one capture, that hole is the whole point of interest. Instead each step
|
|
parks the local oscillator ``dc_guard`` below the span it is responsible
|
|
for, so the covered frequencies sit entirely on one side of DC, clear of
|
|
both the spike and the anti-alias roll-off.
|
|
|
|
``tunable`` is an optional predicate; when the offset LO would fall outside
|
|
the hardware's range the step falls back to centring.
|
|
"""
|
|
if sample_rate <= 0 or usable_fraction <= 0:
|
|
raise RangeError("sample rate and usable fraction must be positive")
|
|
# Half the usable width, because only one side of DC is used per step.
|
|
usable = float(sample_rate) * float(usable_fraction) / 2.0
|
|
if usable <= dc_guard:
|
|
raise RangeError("usable bandwidth is too small for the DC guard band")
|
|
|
|
steps: list[TuneStep] = []
|
|
for i, r in enumerate(ranges):
|
|
if not r.enabled:
|
|
continue
|
|
n = max(1, int(math.ceil(r.span / usable)))
|
|
if len(steps) + n > max_steps:
|
|
raise RangeError(
|
|
f"plan would need over {max_steps} tuner steps; "
|
|
"narrow the ranges or raise the sample rate"
|
|
)
|
|
width = r.span / n
|
|
for k in range(n):
|
|
low = r.start + k * width
|
|
high = low + width
|
|
center = low - dc_guard
|
|
if tunable is not None and not tunable(center):
|
|
center = 0.5 * (low + high) # fall back rather than lose the step
|
|
steps.append(TuneStep(
|
|
center=center,
|
|
low=low, high=high,
|
|
range_index=i, range_label=r.label,
|
|
mode=r.mode, index=len(steps),
|
|
))
|
|
return steps
|
|
|
|
|
|
def plan_summary(ranges: list[ScanRange], steps: list[TuneStep],
|
|
dwell_seconds: float) -> str:
|
|
total = sum(r.span for r in ranges if r.enabled)
|
|
n_en = sum(1 for r in ranges if r.enabled)
|
|
cycle = len(steps) * dwell_seconds
|
|
return (f"{n_en} range(s), {fmt_hz(total)} total span, {len(steps)} tuner "
|
|
f"steps, ~{cycle:.1f} s per full sweep")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lock-outs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class Lockout:
|
|
"""A frequency, or a span of them, the scan must never stop on.
|
|
|
|
A single frequency is stored with ``stop`` equal to ``start`` and is
|
|
widened by the lock-out width setting when it is applied, so changing that
|
|
width still moves it. A span is taken exactly as given -- a pager band or
|
|
a noisy stretch of spectrum is a definite width, not a point with a guess
|
|
around it.
|
|
"""
|
|
|
|
start: float
|
|
stop: float = 0.0
|
|
|
|
def __post_init__(self):
|
|
self.start = float(self.start)
|
|
self.stop = float(self.stop or self.start)
|
|
if self.stop < self.start:
|
|
self.start, self.stop = self.stop, self.start
|
|
|
|
@property
|
|
def is_span(self) -> bool:
|
|
return self.stop > self.start
|
|
|
|
def interval(self, width: float) -> tuple[float, float]:
|
|
"""The span this covers, given the width a point lock-out gets."""
|
|
if self.is_span:
|
|
return self.start, self.stop
|
|
half = max(1.0, float(width)) / 2.0
|
|
return self.start - half, self.start + half
|
|
|
|
def describe(self) -> str:
|
|
if self.is_span:
|
|
return f"{fmt_hz(self.start)}-{fmt_hz(self.stop)}"
|
|
return fmt_hz(self.start)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {"start": self.start, "stop": self.stop}
|
|
|
|
@classmethod
|
|
def coerce(cls, value) -> "Lockout":
|
|
"""Accept anything a saved file or a caller might reasonably hold.
|
|
|
|
Older settings files stored a bare number per lock-out, and callers
|
|
pass plain frequencies, so both still mean what they always did.
|
|
"""
|
|
if isinstance(value, cls):
|
|
return value
|
|
if isinstance(value, dict):
|
|
return cls(value.get("start", 0.0), value.get("stop", 0.0))
|
|
if isinstance(value, (list, tuple)):
|
|
if len(value) == 1:
|
|
return cls(float(value[0]))
|
|
if len(value) == 2:
|
|
return cls(float(value[0]), float(value[1]))
|
|
raise RangeError(f"cannot read {value!r} as a lock-out")
|
|
if isinstance(value, str):
|
|
return parse_lockout(value)
|
|
return cls(float(value))
|
|
|
|
|
|
def parse_lockout(text: str) -> Lockout:
|
|
"""One lock-out: ``162.55M`` or ``162.4M-162.6M``."""
|
|
if text is None:
|
|
raise RangeError("empty lock-out")
|
|
s = str(text).strip()
|
|
if not s:
|
|
raise RangeError("empty lock-out")
|
|
parts = [p for p in _RANGE_SPLIT.split(s) if p.strip()]
|
|
if len(parts) == 1:
|
|
return Lockout(parse_frequency(parts[0]))
|
|
if len(parts) == 2:
|
|
# "162.4-162.6M": the unit on the right end applies to the left too.
|
|
right = _NUM.match(parts[1])
|
|
unit = right.group(2) if right else ""
|
|
return Lockout(parse_frequency(parts[0], default_unit=unit),
|
|
parse_frequency(parts[1]))
|
|
raise RangeError(f"cannot read {text!r} as a lock-out")
|
|
|
|
|
|
def parse_lockout_list(text) -> list[Lockout]:
|
|
"""A comma or semicolon separated list of frequencies and spans."""
|
|
if text is None:
|
|
return []
|
|
if isinstance(text, (list, tuple)):
|
|
items = text
|
|
else:
|
|
items = re.split(r"[,;\n]", str(text))
|
|
out = []
|
|
for item in items:
|
|
if isinstance(item, str):
|
|
item = item.strip()
|
|
if not item or item.lower() in ("(none)", "none", "-"):
|
|
continue
|
|
out.append(Lockout.coerce(item))
|
|
return out
|