diff --git a/README.md b/README.md index baecdde..3064ef0 100644 --- a/README.md +++ b/README.md @@ -157,6 +157,8 @@ Prefix all commands with `uv run apps/ham2mon.py` (which runs the script using t uv run apps/ham2mon.py [options] ``` +Not sure how to run ham2mon for what you're trying to do? The [Scanning Guide](./doc/scanning-guide.md) walks through every mode of operation — from a bare sweep with no frequency file to bank-filtered monitoring — what each one is for, and what you'll see on screen. + ## Console Operation: The following is an example of the option switches for UHD with NBFM demodulation, although omission of any will use default values (shown below) that are optimal for the B200: @@ -209,6 +211,8 @@ uv run apps/ham2mon.py -a "file=gqrx.raw,rate=8E6,repeat=false,throttle=true,fre `/ = Frequency entry mode (Esc to exit)` +`b = Edit active banks (Enter to apply, Esc to cancel)` + `CTRL-C or SHIFT-Q = quit` > [!IMPORTANT] @@ -484,11 +488,41 @@ The frequency file contains metadata for individual frequencies and ranges of fr 2. Lockout frequencies 3. Frequency labeling 4. CTCSS (PL tone) filtering +5. Bank tags If an individual frequency or frequency range is specified more than once, an error will be generated and ham2mon will not load (unless the duplicate entry is used to specify an additional unique ctcss tone for that frequency). -For an example, see the [example frequencies file](./doc/example.freqs.yaml). +For an example, see the [example frequencies file](./doc/example.freqs.yaml). For a build-up from the bare minimum through every configuration option, see the [full example frequencies file](./doc/full-example.freqs.yaml). + +### Bank Filtering (`--banks`) +Banks are optional tags applied to frequency entries and to per-tone rules in the frequency file: + + - label: "Local repeater output" + single: 462.730 + banks: ["NET_A"] + +Select which banks to monitor with `--banks` (or `frequency_policies.active_banks` in YAML): + + uv run apps/ham2mon.py -a "airspy" -f 460.0-470.0 --banks NET_A NET_B + +`--banks` is a **filter**, not a scan-scope control. The scanner still sweeps the entire configured band (or range) every scan cycle; bank filtering only controls which channels are demodulated and which captured transmissions are kept. A channel whose resolved bank tags do not intersect the selected set is never assigned a demodulator, and a transmission already captured on such a channel is discarded. + +Bank filtering is fail-closed: if a `--banks` tag matches no configured frequency or tone bank, no channel is demodulated and ham2mon logs a startup warning. Without `--banks`, all channels are monitored. Two special tags are available: `SEARCH` lets unconfigured spectrum hits be monitored, and `UNTAGGED` matches channels that carry no bank tag while filtering is active. + +The RECEIVER panel shows the active selection in its **Banks** row (`NET_A, NET_B`, or `none` without `--banks`), and each CHANNELS entry displays its resolved tags as a bracketed block just before the CTCSS readout (e.g. `[NET_A,NET_B]`), dimmed while the channel is idle. `SEARCH` and `UNTAGGED` appear as literal tags here only while bank filtering is active. + +A bank tag can be written as a dict mapping the tag to a **per-bank display label** (`banks: {AREA_A: "Net A", AREA_B: "Net B"}`). The keys are the membership tags, exactly like the list form; the values are per-bank label overrides. When a hit resolves to one of those banks, its label is shown instead of the entry label (this is how one frequency reads differently per geographic area). Label precedence for a matched entry is: per-bank label, then per-tone label, then the entry label. An optional top-level `banks:` section in the frequency file attaches display-only names to tags for the RECEIVER Banks row (e.g. `BANKS: NET_A (Net A)`); it does not declare membership. See the [full example frequencies file](./doc/full-example.freqs.yaml). + +Use `--list-banks` to audit bank membership without running the scanner: it loads the frequency file, prints each configured bank with its channel members (using the top-level display label when present), and exits. + +For the full set of scanning modes — including running with no frequency file, and custom bank combinations such as `NET_A SEARCH` or a default bank for a range with specific-channel overrides — see the [Scanning Guide](./doc/scanning-guide.md). + +#### Changing banks at runtime + +Press `b` to edit the active banks in place: the Banks row switches to an editable text field pre-filled with the current selection. Type a comma- or space-separated list (letters, digits, `_`, `-`, `,`, space) and press `Enter` to apply, or `Esc` to cancel. Submitting an empty list (or the literal `none`) restores promiscuous mode. + +The change applies immediately on the next scan cycle: channels whose resolved tags no longer intersect the new selection are no longer assigned a demodulator, and any transmission still running on a now-deselected bank finishes naturally but its recording is discarded. Runtime changes are not persisted and do not affect the `--banks` setting for the next launch. ### Priority Handling Priorities can be assigned to frequencies and frequency ranges in the frequency file. Highest priority is 1. Frequencies can have equal priority. If no priority is assigned the default value is no priority. @@ -597,29 +631,31 @@ CTCSS (Continuous Tone-Coded Squelch System) allows filtering transmissions by r ### Configuration -CTCSS tones are configured per-channel inside the YAML frequencies file (specified via the `-F`/`--frequencies` command-line option): +CTCSS tones are configured per-channel inside the YAML frequencies file (specified via the `-F`/`--frequencies` command-line option) using the unified `tones:` key. Each list item is either a bare tone frequency in Hz or a tone rule dict with an optional per-tone label and per-tone banks: ```yaml frequencies: + # Single tone, bare frequency in Hz - label: "CTCSS Test Channel" single: 144.500 - ctcss: 100.0 # Configured expected CTCSS tone in Hz + tones: [100.0] ``` -To support **multiple valid CTCSS tones** on a single frequency or frequency range, declare the frequency block multiple times, changing only the `ctcss` tone frequency. These will merge into a single tuner entry at load time: +To support **multiple valid CTCSS tones** on a single frequency or frequency range, list them all under `tones:` on a single entry: ```yaml frequencies: - # Primary tone - - label: "CTCSS Test Channel" - single: 144.500 - ctcss: 100.0 - # Backup tone - label: "CTCSS Test Channel" single: 144.500 - ctcss: 141.3 + tones: + - ctcss: 100.0 # in Hz + label: "Primary" + - ctcss: 141.3 + label: "Backup" ``` +> Declaring the same frequency twice (even with different tones) is an error; multiple tones must live on one entry. + By default, **CTCSS demodulation is disabled (`--max-ctcss-tones` defaults to 0) for performance reasons**, as running CTCSS tone detection blocks on every channel incurs significant CPU overhead even when no signal is present. To enable CTCSS tone detection, you must specify a limit (e.g. `--max-ctcss-tones 3`). Any configured tones loaded beyond this limit are validated and rejected at configuration load time. ### User Options @@ -627,10 +663,10 @@ By default, **CTCSS demodulation is disabled (`--max-ctcss-tones` defaults to 0) Depending on your configuration in your `.freqs.yaml` file (specified via the `-F`/`--frequencies` option), the application operates in one of two modes: 1. **Carrier Squelch (CSQ) Mode (CTCSS Bypassed):** - * **Trigger:** Enabled for any channel configured in your `.freqs.yaml` file **without** a `ctcss` tone, or for any frequency not present in the file at all (such as dynamically discovered frequencies during a range scan). + * **Trigger:** Enabled for any channel configured in your `.freqs.yaml` file **without** a `tones:` entry, or for any frequency not present in the file at all (such as dynamically discovered frequencies during a range scan). * **Behavior:** The receiver will record and unmute any signal that is strong enough to break the RF carrier power squelch, regardless of whether a sub-audible tone is present or what its frequency is. Additionally, the 300Hz high-pass filter is dynamically bypassed in this mode to preserve full audio fidelity and bass (e.g. for broadcast FM music). 2. **Tone Squelch (CTCSS) Mode:** - * **Trigger:** Enabled for channels configured **with** a specific `ctcss` key (e.g. `ctcss: 100.0`). + * **Trigger:** Enabled for channels configured with a `tones:` entry (e.g. `tones: [100.0]`). * **Behavior:** The receiver will only unmute and keep the recording if the signal contains one of the configured CTCSS tones. Transmissions carrying a different tone or no tone at all are muted and discarded. ### GUI Display diff --git a/apps/channel_loggers.py b/apps/channel_loggers.py index 8579024..a167a98 100644 --- a/apps/channel_loggers.py +++ b/apps/channel_loggers.py @@ -20,7 +20,7 @@ class ActivityParams: ''' type: str dest: str - interval: int + interval: float | int class ActivityLogger(ABC): ''' @@ -29,10 +29,10 @@ class ActivityLogger(ABC): def __init__(self, params: ActivityParams, get_ctcss: Callable[[int], float | None] | None = None) -> None: logger.debug(f'Creating {self.__class__.__name__} channel logger') - self.interval: int = 0 # overridden by child classes - self.log_task: dict[int, asyncio.Task] = {} # activity logging tasks are channel specific - self.params = params - self.get_ctcss = get_ctcss # optional callback: bb_freq -> matched ctcss tone or None + self.interval: float | int = 0 # overridden by child classes + self.log_task: dict[int, asyncio.Task[None]] = {} # activity logging tasks are channel specific + self.params: ActivityParams = params + self.get_ctcss: Callable[[int], float | None] | None = get_ctcss # optional callback: bb_freq -> matched ctcss tone or None async def log(self, msg: ChannelMessage | None, record: TransmissionRecord | None = None) -> None: @@ -102,7 +102,10 @@ async def log_active(self, msg: ChannelMessage) -> None: rf=msg.rf, bb=msg.bb, channel=msg.channel, - matched_ctcss=live_ctcss)) + matched_ctcss=live_ctcss, + label=msg.label, + priority=msg.priority, + banks=msg.banks)) class NoOp(ActivityLogger): ''' @@ -112,7 +115,7 @@ def __init__(self, params: ActivityParams, get_ctcss: Callable[[int], float | None] | None = None) -> None: super().__init__(params, get_ctcss=get_ctcss) - self.interval: int = 0 + self.interval: float | int = 0 async def log(self, msg: ChannelMessage | None, record: TransmissionRecord | None = None) -> None: @@ -125,12 +128,12 @@ class FixedField(ActivityLogger): ''' Send channel events to a file with fixed field length records ''' - def __init__(self, params, + def __init__(self, params: ActivityParams, get_ctcss: Callable[[int], float | None] | None = None) -> None: super().__init__(params, get_ctcss=get_ctcss) - self.file_name = params.dest - self.interval = params.interval + self.file_name: str = params.dest + self.interval: float | int = params.interval async def log(self, msg: ChannelMessage | None, record: TransmissionRecord | None = None) -> None: @@ -140,11 +143,13 @@ async def log(self, msg: ChannelMessage | None, await super().log(msg) now = datetime.datetime.now() + banks_str: str = ",".join(msg.banks) if msg.banks else "" with open(self.file_name, 'a') as file: text = (f'{now.strftime("%Y-%m-%d, %H:%M:%S.%f")}: {msg.state:<4}{msg.rf:<10}' f'{msg.channel:<2}{msg.priority if msg.priority else "":<2}' f'{msg.classification if msg.classification else "":<2}' f'{f"{msg.matched_ctcss:.1f}" if msg.matched_ctcss else "":<7}' + f'{banks_str[:15]:<15}' f'{msg.file if msg.file else "":<50}\n' ) file.write(text) @@ -155,12 +160,12 @@ class JsonToServer(ActivityLogger): ''' Send channels events as json messages to a remote server ''' - def __init__(self, params, + def __init__(self, params: ActivityParams, get_ctcss: Callable[[int], float | None] | None = None) -> None: super().__init__(params, get_ctcss=get_ctcss) - self.server = params.dest - self.interval = params.interval + self.server: str = params.dest + self.interval: float | int = params.interval self.requests = import_module('requests') # urllib3 log suppression is configured at application startup in ham2mon.py diff --git a/apps/components/base.py b/apps/components/base.py index 5cd6d1b..49de0f5 100644 --- a/apps/components/base.py +++ b/apps/components/base.py @@ -63,6 +63,9 @@ class ChannelInfo: wav_tmp_path: str """Absolute path to the tmp WAV file being evaluated.""" + banks: list[str] = field(default_factory=list) + """Resolved scanner bank tags for this channel.""" + class Component(ABC): """Base abstract class for all ham2mon components.""" diff --git a/apps/config.py b/apps/config.py index b9ad8dc..ae044c2 100644 --- a/apps/config.py +++ b/apps/config.py @@ -174,11 +174,12 @@ def __post_init__(self): @dataclass(kw_only=True) class FrequencyPoliciesConfig: - """Frequencies file path, lockout settings, and priority overrides.""" + """Frequencies file path, lockout settings, priority overrides, and active banks.""" file: Optional[Path] = None disable_lockout: bool = False disable_priority: bool = False + active_banks: List[str] = field(default_factory=list) @dataclass(kw_only=True) diff --git a/apps/cursesgui.py b/apps/cursesgui.py index c96969b..c227086 100644 --- a/apps/cursesgui.py +++ b/apps/cursesgui.py @@ -12,7 +12,13 @@ import logging from pathlib import Path, PurePath from frequency_manager import ConfigFrequency, ChannelFrequency, ChannelList, FrequencyList -from utilities import baseband_to_bin, build_column_edges, index_to_column +from utilities import ( + baseband_to_bin, + build_column_edges, + format_active_banks, + format_channel_banks, + index_to_column, +) from ui_theme import THEME, ThemeConfiguration logger = logging.getLogger(f"ham2mon.{__name__}") @@ -440,7 +446,7 @@ def draw(self) -> None: label_start = 14 matched_ctcss = getattr(channel, 'matched_ctcss', None) - primary_ctcss = channel.ctcss or (channel.ctcss_tones[0] if channel.ctcss_tones else None) + primary_ctcss = channel.ctcss_tones[0] if channel.ctcss_tones else None has_multiple_ctcss = len(channel.ctcss_tones) > 1 is_testing_ctcss = channel.active and not channel.hanging and has_multiple_ctcss and matched_ctcss is None @@ -457,9 +463,26 @@ def draw(self) -> None: ctcss_str = f'{display_ctcss:>5.1f}' win.addnstr(row, col + self.width - 5, ctcss_str , 5, attributes[1] | curses.A_ITALIC) win.addnstr(row, col + self.width - 6, ' ', 1, attributes[0]) - label_end = self.width - 6 - else: - label_end = self.width + + # Right-align the bank tag block just before the CTCSS field (or + # the right border when no CTCSS), reserving its width from the + # label region. Empty banks draw nothing, so non-bank users see + # no layout shift. + ctcss_cols = 6 if display_ctcss is not None else 0 + bank_end = self.width - ctcss_cols + bank_str = format_channel_banks( + getattr(channel, "banks", None) or [], max(0, bank_end - 1) + ) + bank_start = bank_end - len(bank_str) + label_end = bank_start - 1 if bank_str else bank_end + + if bank_str: + bank_attr = ( + THEME.get("channel.bank_active") + if channel.active + else THEME.get("channel.bank_inactive") + ) + win.addnstr(row, col + bank_start, bank_str, len(bank_str), bank_attr) if label_end > label_start: remainder = label_end - label_start @@ -855,6 +878,9 @@ def __init__(self, screen, width=None): self.freq_max = 148E6 self.samp_rate = 2E6 self.freq_entry = 'None' + self.banks: set[str] = set() + self.bank_labels: dict[str, str] = {} + self.bank_entry: str | None = None self.squelch_db = -60 self.volume_db = 0 self.type_demod = 0 @@ -864,6 +890,9 @@ def __init__(self, screen, width=None): self.activity_dest = "" self.gains = None self.classifier_params = None + # Declared here (initialized in draw_frame like the other *_field + # attributes) so the read-only Banks value can be rendered safely. + self.banks_field: RxWindow.RxEntry | None = None self.demod_map = { 0: 'NBFM', @@ -1038,6 +1067,8 @@ def draw_frame(self) -> None: self.frequency_file_name_field = RxWindow.RxEntry( "Freq File", 2, 'left', False) + self.banks_field = RxWindow.RxEntry("Banks", 2, "left", False) + self.activity_type_field = RxWindow.RxEntry( "Activity Type", 2, 'left', False) @@ -1082,6 +1113,13 @@ def draw_rx(self) -> None: file_name = self.frequency_file_name.name if self.frequency_file_name else "none" self.frequency_file_name_field.set(file_name) + if self.bank_entry is not None: + banks_text = self.bank_entry + else: + banks_text = format_active_banks(self.banks, self.bank_labels) + if self.banks_field is not None: + self.banks_field.set(banks_text) + self.activity_type_field.set(self.activity_type) if self.activity_dest is not None: @@ -1156,6 +1194,10 @@ def proc_keyb_hard(self, keyb: int): # set mode to frequency entry self.freq_entry = '' return False + elif keyb == ord("b") and self.freq_entry == "None": + # set mode to bank entry (mutually exclusive with frequency entry) + self.start_bank_entry() + return False elif keyb == 27: # ESC # end frequncy entry mode without seting the frequency self.freq_entry = 'None' @@ -1178,6 +1220,39 @@ def proc_keyb_hard(self, keyb: int): else: return False + def start_bank_entry(self) -> None: + """Enter bank-entry mode, pre-filled with the current active banks. + + The live entry text is shown in the Banks receiver field; Enter + applies it (via the caller), ESC cancels. + """ + self.bank_entry = ",".join(sorted(self.banks)) + + def proc_keyb_bank_entry(self, keyb: int) -> bool: + """Process keystrokes in bank-entry mode. + + ESC cancels, Enter applies, Backspace deletes, and letters/digits/ + '_'/'-'/','/space build up the entry. Returns True only when Enter + was pressed, so the caller can apply the parsed value. The entry + text is consumed (set to None) on both ESC and Enter. + """ + if keyb == 27: # ESC + self.bank_entry = None + return False + entry = self.bank_entry + if entry is None: + return False + if keyb == ord('\n'): + self.bank_entry = None + return True + if keyb == curses.KEY_BACKSPACE: + self.bank_entry = entry[:-1] + return False + if chr(keyb).isalnum() or chr(keyb) in " _,-": + self.bank_entry = entry + chr(keyb) + return False + return False + def _adjust_gain_stage(self, index: int, delta: float) -> bool: if index < len(self.gains): self.gains[index]["value"] += delta diff --git a/apps/default.theme.yaml b/apps/default.theme.yaml index 781eb58..f1ce866 100644 --- a/apps/default.theme.yaml +++ b/apps/default.theme.yaml @@ -35,6 +35,8 @@ styles: channel.icon_inactive: { fg: 46, dim: true } channel.index_active: { fg: 39, bold: true } channel.index_inactive: { fg: 24 } + channel.bank_active: { fg: 51 } + channel.bank_inactive: { fg: 51, dim: true } channel.placeholder_index: { fg: 24, dim: true } # index label on empty "Scanning..." rows channel.placeholder_text: { fg: 244, dim: true } # the "Scanning..." text itself # Note: the CTCSS tone readout uses channel.icon_active/channel.icon_inactive diff --git a/apps/frequency_manager.py b/apps/frequency_manager.py index 0af5b62..f0d3155 100644 --- a/apps/frequency_manager.py +++ b/apps/frequency_manager.py @@ -9,13 +9,44 @@ import os from dataclasses import dataclass, field from pathlib import Path -from typing import TypeAlias # TypeAlias needed for python < 3.12 +from typing import Any, TypeAlias # TypeAlias needed for python < 3.12 import yaml from utilities import frequency_to_baseband logger = logging.getLogger(f"ham2mon.{__name__}") +# Match tolerances shared across FrequencyManager RF/tone matching +# (resolve_banks, get_label, get_ctcss_info, get_ctcss_tones). +FREQ_MATCH_TOLERANCE_MHZ = 1e-4 +TONE_MATCH_TOLERANCE_HZ = 0.5 + + +@dataclass(kw_only=True) +class ToneRule: + """Structured CTCSS tone rule for single frequencies or ranges.""" + ctcss: float + label: str | None = None + banks: list[str] = field(default_factory=list) + + def __post_init__(self) -> None: + try: + self.ctcss = float(self.ctcss) + except (ValueError, TypeError): + raise ValueError("CTCSS must be a float or integer representing frequency in Hz") + if self.ctcss <= 0: + raise ValueError("CTCSS must be a positive number") + if isinstance(self.banks, str): + self.banks = [self.banks] + + +def _coerce_tone_rule(rule: ToneRule | dict[str, Any] | float) -> ToneRule: + if isinstance(rule, ToneRule): + return rule + if isinstance(rule, (int, float)): + return ToneRule(ctcss=rule) + return ToneRule(**rule) + @dataclass(kw_only=True) class FrequencyInfo: @@ -25,7 +56,8 @@ class FrequencyInfo: label: str = field(default=None) locked: bool = field(default=False) priority: int | None = field(default=None) - ctcss: float | None = field(default=None) + banks: list[str] = field(default_factory=list) + bank_labels: dict[str, str] | None = field(default=None, compare=False) def __post_init__(self): if not isinstance(self.locked, bool): @@ -35,14 +67,18 @@ def __post_init__(self): if not isinstance(self.priority, int) or self.priority < 1: raise ValueError('Priority must be an integer >= 1') - if self.ctcss is not None: - try: - self.ctcss = float(self.ctcss) - except (ValueError, TypeError): - raise ValueError('CTCSS must be a float or integer representing frequency in Hz') - if self.ctcss <= 0: - raise ValueError('CTCSS must be a positive number') + if isinstance(self.banks, dict): + # Dict form: {bank tag: per-bank display label}. The keys are the + # membership tags (a channel can belong to several banks); the + # values are per-bank label overrides shown instead of the entry + # label when the hit resolves to that bank. + self.bank_labels = {str(k): v for k, v in self.banks.items()} + self.banks = list(self.bank_labels) + elif isinstance(self.banks, str): + self.banks = [self.banks] + if self.bank_labels is not None and not isinstance(self.bank_labels, dict): + raise ValueError('bank_labels must be a dict mapping bank tags to labels') @dataclass(kw_only=True, eq=False) @@ -67,14 +103,44 @@ class ConfigFrequency(FrequencyInfo): # if not a single it is a range is_single: bool | None = field(default=None) - # All CTCSS tones considered valid for this frequency/range. Populated from - # `ctcss` on construction, and may be extended later via FrequencyManager.add() - # when the same frequency/range is declared again in config with a different - # `ctcss` value (e.g. a repeater with a primary and a backup PL tone). `ctcss` - # itself remains the first/primary tone, kept for backward compatibility with - # code that only expects a single value (e.g. get_ctcss_info()). + # tones: list of CTCSS tone rules with optional per-tone labels and banks. + # Each item is a ToneRule, a dict, or a bare float (see _coerce_tone_rule). + # Config-domain only — ChannelFrequency and ChannelMessage do not carry tones. + tones: list[ToneRule] = field(default_factory=list) + + # All CTCSS tones considered valid for this frequency/range, in order. + # Derived from the normalized `tones` rules; the first entry is the + # primary tone surfaced by get_ctcss_info(). ctcss_tones: list[float] = field(default_factory=list, init=False, repr=False) - ctcss_labels: list[str] = field(default_factory=list, init=False, repr=False) + + def __post_init__(self): + # Call parent validation first (banks norm) + super().__post_init__() + + # Convert any raw dict items under tones: into ToneRule instances + if self.tones: + self.tones = [_coerce_tone_rule(t) for t in self.tones] + + # Ensure any tone rule missing banks inherits parent entry's banks + for tone_rule in self.tones: + if not tone_rule.banks and self.banks: + tone_rule.banks = list(self.banks) + + # Validate frequency types + self._validate_frequency_types() + + # Validate frequency specification (single or range) + self._validate_frequency_specification() + + # Validate frequency values + self._validate_frequency_values() + + # Set state + self.is_single = self.single is not None + self.ctcss_tones = [] + for tone_rule in self.tones: + if tone_rule.ctcss not in self.ctcss_tones: + self.ctcss_tones.append(tone_rule.ctcss) def calculate_baseband(self, center_freq: int, channel_spacing: int) -> None: @@ -106,25 +172,6 @@ def get_priority_at(self, bb: int) -> int | bool: return None - def __post_init__(self): - # Call parent validation first - super().__post_init__() - - # Validate frequency types - self._validate_frequency_types() - - # Validate frequency specification (single or range) - self._validate_frequency_specification() - - # Validate frequency values - self._validate_frequency_values() - - # Set state - self.is_single = self.single is not None - self.ctcss_tones = [self.ctcss] if self.ctcss is not None else [] - self.ctcss_labels = [self.label] if (self.ctcss is not None and self.label is not None) else [] - - def _validate_frequency_types(self): """Ensure all frequency values are floats if provided""" for attr_name, attr_value in [ @@ -226,6 +273,9 @@ def __str__(self) -> str: if self.label: parts.append(f"[{self.label}]") + if self.banks: + parts.append(f"[{','.join(self.banks)}]") + if self.file: parts.append(f"Saved: {os.path.basename(self.file)}") elif self.detail: @@ -276,6 +326,8 @@ class TransmissionRecord: metadata: dict = field(default_factory=dict) # type: ignore[reportGeneralTypeIssues] """Free-form metadata populated by TransmissionComponent processors in Phase 4 (e.g. transcription text, classifier confidence scores).""" + banks: list[str] = field(default_factory=list) + """List of scanner bank tags assigned to this transmission.""" def __str__(self) -> str: @@ -283,6 +335,8 @@ def __str__(self) -> str: parts = [f"{self.rf:.4f} MHz", f"{self.duration_sec:.1f}s"] if self.label: parts.append(f"[{self.label}]") + if self.banks: + parts.append(f"[{','.join(self.banks)}]") if self.priority is not None: parts.append(f"P{self.priority}") if self.classification: @@ -315,9 +369,140 @@ def __init__(self, config: FrequencyConfiguration, channel_spacing: int) -> None self.center_freq = None self.config = config self.frequencies: FrequencyList = [] + self.active_banks: set[str] = set() + # Display-only labels for bank tags, from the optional top-level + # ``banks:`` section of the frequency file. Membership lives on the + # channel entries themselves; this map only decorates the TUI. + self.bank_display_labels: dict[str, str] = {} + + def set_active_banks(self, banks: list[str] | set[str] | None) -> None: + """Set the active bank tags for filtering and stepping inclusion.""" + if banks is None: + self.active_banks = set() + elif isinstance(banks, str): + self.active_banks = {banks} + else: + self.active_banks = set(banks) - async def process_frequencies_data(self, frequencies_config) -> FrequencyList: - """Process pre-loaded frentryequencies configuration data.""" + def is_bank_active(self, bank_list: list[str]) -> bool: + """Return True if active_banks is empty (promiscuous scan all mode), + or if active_banks intersects with bank_list.""" + if not self.active_banks: + return True + return bool(self.active_banks.intersection(bank_list)) + + def configured_banks(self) -> set[str]: + """Union of all bank tags configured across loaded frequencies and tone rules.""" + banks: set[str] = set() + for freq in self.frequencies: + banks.update(freq.banks) + for tone_rule in freq.tones: + banks.update(tone_rule.banks) + return banks + + def bank_members(self, bank: str) -> list[ConfigFrequency]: + """Channel-level entries whose membership tags include ``bank``. + + Returns entries in configuration (load) order, not scanning priority + order. Tone rules are not listed: their banks tag individual tones, + and membership auditing is at the channel level. ``--list-banks`` + uses this to preview a bank's contents. + """ + return [freq for freq in self.frequencies if bank in freq.banks] + + def unknown_active_banks(self) -> set[str]: + """Active bank tags that match no configured bank tag. The "SEARCH" + and "UNTAGGED" dynamic tags are exempt. Returns empty set in + promiscuous mode.""" + if not self.active_banks: + return set() + return self.active_banks - self.configured_banks() - {"SEARCH", "UNTAGGED"} + + def _warn_on_unmatched_active_banks(self) -> None: + """Log a warning when active bank tags match no configured bank tag. + + Bank filtering is fail-closed (see is_bank_active), so a typo'd active + bank (or a missing/empty frequency file) silently disables every + channel — surface it at startup. + """ + unknown = self.unknown_active_banks() + if unknown: + logger.warning( + 'Active bank(s) %s match no configured frequency/tone banks; bank filtering is fail-closed so channels in these will not be monitored. Check the --banks spelling and that -F/--frequencies points at a file defining matching banks.', + sorted(unknown)) + + def _untagged_fallback(self) -> list[str]: + """Bank-filtering sentinel tag. + + "UNTAGGED" is only meaningful when the user opted into bank filtering + (non-empty active_banks); it lets untagged hits fail-closed via + is_bank_active(). In promiscuous/legacy mode (no --banks) return no + tags so bank metadata stays empty for non-participants instead of + injecting "UNTAGGED" into logs, fixed-field records, and JSON. + """ + return ["UNTAGGED"] if self.active_banks else [] + + def resolve_banks(self, rf: float, ctcss_hz: float | None = None) -> list[str]: + """Resolve bank tags for a carrier hit at rf with decoded ctcss_hz + using 5-tier precedence hierarchy: + 1. Explicit single entry tone match + 2. Explicit single entry base bank + 3. Range entry tone match + 4. Range entry base bank + 5. Dynamic fallback ("SEARCH" if active; else "UNTAGGED" when bank + filtering is active, otherwise no tags) + + Note: "UNTAGGED" also arises from tiers 2 & 4 (and their pre-tuning + union fallbacks) when a configured entry matches but sets no bank + tags -- all such paths return _untagged_fallback(), the same gated + sentinel as tier 5. Under bank filtering both an unbanked configured + entry and an unmatched hit resolve to UNTAGGED and fail closed via + is_bank_active(); in promiscuous mode both resolve to no tags. + """ + # Tier 1 & 2: Check single frequencies first + for freq in self.frequencies: + if freq.is_single and freq.single is not None and abs(freq.single - rf) < FREQ_MATCH_TOLERANCE_MHZ: + if ctcss_hz is not None: + for tone_rule in freq.tones: + if abs(tone_rule.ctcss - ctcss_hz) < TONE_MATCH_TOLERANCE_HZ: + return tone_rule.banks if tone_rule.banks else freq.banks + return freq.banks if freq.banks else self._untagged_fallback() + else: + # Pre-tuning (no tone decoded yet): return Union of all possible bank tags + union_banks = set(freq.banks) + for tone_rule in freq.tones: + union_banks.update(tone_rule.banks) + return list(union_banks) if union_banks else self._untagged_fallback() + + # Tier 3 & 4: Check range entries + for freq in self.frequencies: + if not freq.is_single and freq.lo is not None and freq.hi is not None and freq.lo <= rf <= freq.hi: + if ctcss_hz is not None: + for tone_rule in freq.tones: + if abs(tone_rule.ctcss - ctcss_hz) < TONE_MATCH_TOLERANCE_HZ: + return tone_rule.banks if tone_rule.banks else freq.banks + return freq.banks if freq.banks else self._untagged_fallback() + else: + # Pre-tuning (no tone decoded yet): return Union of all possible bank tags + union_banks = set(freq.banks) + for tone_rule in freq.tones: + union_banks.update(tone_rule.banks) + return list(union_banks) if union_banks else self._untagged_fallback() + + # Tier 5: Outside configured entries + if "SEARCH" in self.active_banks: + return ["SEARCH"] + return self._untagged_fallback() + + async def process_frequencies_data(self, frequencies_config: dict[str, Any]) -> FrequencyList: + """Process pre-loaded frequency configuration data.""" + + # Optional top-level ``banks:`` section carries display-only labels + # for the TUI Banks row. It does not define membership; membership is + # declared per-channel via each entry's ``banks:``. + banks_section = frequencies_config.get('banks') + if isinstance(banks_section, dict): + self.bank_display_labels = {str(k): str(v) for k, v in banks_section.items()} if 'frequencies' in frequencies_config: for freq in frequencies_config['frequencies']: @@ -331,6 +516,7 @@ async def load(self) -> FrequencyList: self.frequencies = [] if not self.config.file_name: + self._warn_on_unmatched_active_banks() return [] file = self.config.file_name @@ -340,7 +526,7 @@ async def load(self) -> FrequencyList: logger.debug(f'Loading frequencies from {file}') with file.open(mode='r') as file: try: - frequencies_config = yaml.safe_load(file) + frequencies_config: dict[str, Any] = yaml.safe_load(file) except yaml.YAMLError as e: if hasattr(e, 'problem_mark'): logger.error( @@ -351,18 +537,18 @@ async def load(self) -> FrequencyList: raise Exception( "Invalid yaml frequency file (enable debugging for more info)") - return await self.process_frequencies_data(frequencies_config) + _ = await self.process_frequencies_data(frequencies_config) + self._warn_on_unmatched_active_banks() + return self.frequencies async def add(self, entry: dict) -> FrequencyList: ''' Add frequency to channels if not already there. - A frequency/range may legitimately be declared more than once in config - when the only difference is the `ctcss` tone (e.g. a repeater that - answers to a primary and a backup PL tone). In that case, the new tone - is merged into the existing entry's `ctcss_tones` list rather than - raising. Any other kind of duplicate (an identical entry, or a repeat - that doesn't add a new tone) is still an error. + Each frequency or range must be declared exactly once. CTCSS tones + are configured with the ``tones: [...]`` array, where each item is a + tone rule dict (with optional per-tone labels and banks) or a bare + float. Args: entry (dict): Dictionary of frequency attributes @@ -379,37 +565,29 @@ async def add(self, entry: dict) -> FrequencyList: FrequencyList: List of frequencies Raises: - ValueError: If the frequency already occurs in the list and is not - a mergeable CTCSS-only variant of an existing entry. + ValueError: If the frequency already occurs in the list. ''' wanted = ConfigFrequency(**entry) - if wanted.ctcss is not None: + if wanted.ctcss_tones: if self.config.max_ctcss_tones <= 0: raise ValueError( f"CTCSS is disabled (max_ctcss_tones={self.config.max_ctcss_tones}) " - f"but frequency config specifies ctcss: {wanted.ctcss}") + f"but frequency config specifies {len(wanted.ctcss_tones)} CTCSS tone(s)") if len(wanted.ctcss_tones) > self.config.max_ctcss_tones: raise ValueError( - f"Frequency config specifies {len(wanted.ctcss_tones)} CTCSS tones " - f"but max_ctcss_tones is limited to {self.config.max_ctcss_tones}") + f"A frequency config specifies {len(wanted.ctcss_tones)} CTCSS tones " + f"on a channel but max_ctcss_tones is limited to {self.config.max_ctcss_tones}") # use the dataclass __eq__ functions to look for matches matching_frequencies = [existing for existing in self.frequencies if wanted == existing] if len(matching_frequencies) > 0: - existing = matching_frequencies[0] - - if wanted.ctcss is not None and wanted.ctcss not in existing.ctcss_tones: - self._merge_ctcss_tone(existing, wanted, entry) - return self.frequencies - - # Already one occurance, and no new tone to merge, so this is an error raise ValueError( - f'Frequency {wanted} already occurs in list') + f'Frequency {wanted} already occurs in list. Use tones: [...] on a single entry to configure multiple CTCSS tones.') - # add the basband if center frequency has been set + # add the baseband if center frequency has been set if self.center_freq: wanted.calculate_baseband(self.center_freq, self.channel_spacing) @@ -417,43 +595,6 @@ async def add(self, entry: dict) -> FrequencyList: return self.frequencies - def _merge_ctcss_tone(self, existing: ConfigFrequency, wanted: ConfigFrequency, entry: dict) -> None: - ''' - Merge an additional CTCSS tone into an already-loaded frequency/range. - - The first-seen entry's label/priority/locked remain authoritative for - the merged entry. A conflicting, explicitly-specified priority on the - duplicate entry is treated as a config error, since it's ambiguous - which one should apply. (locked/label conflicts are intentionally not - validated in this phase.) - - Args: - existing (ConfigFrequency): The already-loaded entry to merge into - wanted (ConfigFrequency): The newly parsed duplicate entry - entry (dict): The raw dict passed to add(), used to distinguish - "not specified" from "explicitly set" for conflict checks - ''' - if self.config.max_ctcss_tones <= 0: - raise ValueError( - f"CTCSS is disabled (max_ctcss_tones={self.config.max_ctcss_tones}) " - f"but frequency config specifies ctcss: {wanted.ctcss}") - - if len(existing.ctcss_tones) >= self.config.max_ctcss_tones: - raise ValueError( - f"Cannot merge CTCSS tone {wanted.ctcss} into " - f"{existing.label!r}: exceeds max_ctcss_tones limit of {self.config.max_ctcss_tones}") - - if 'priority' in entry and existing.priority is not None and wanted.priority != existing.priority: - raise ValueError( - f"Cannot merge CTCSS tone {wanted.ctcss} into " - f"{existing.label!r}: priority {wanted.priority} conflicts " - f"with existing priority {existing.priority}") - - existing.ctcss_tones.append(wanted.ctcss) - existing.ctcss_labels.append(wanted.label or existing.label or "") - logger.debug( - f'Merged CTCSS tone {wanted.ctcss} into existing frequency ' - f'{existing.label!r} (tones now {existing.ctcss_tones})') async def change(self, entry: dict) -> FrequencyList: @@ -481,11 +622,6 @@ async def change(self, entry: dict) -> FrequencyList: if field in entry: setattr(frequency, field, entry[field]) - # CTCSS tone merging in change() - if 'ctcss' in entry and entry['ctcss'] is not None: - if entry['ctcss'] not in frequency.ctcss_tones: - self._merge_ctcss_tone(frequency, new_values, entry) - return self.frequencies if 'mode' in entry and entry['mode'] == 'add': @@ -578,23 +714,23 @@ def get_priority_info(self, bb: int) -> tuple[int | None, bool]: def get_ctcss_info(self, rf_freq: float) -> float | None: """Get the primary CTCSS tone frequency (in Hz) for the given absolute RF frequency. - NOTE: if multiple tones are configured for this frequency (see - get_ctcss_tones), this returns only the first/primary one. Kept for - backward compatibility with callers that only support a single tone - (currently: the single-tone squelch in BaseTuner). Callers that want to - validate against all configured tones should use get_ctcss_tones(). + Returns the first tone from the ``tones:`` list (the primary tone). + Kept single-valued for callers (e.g. the UI) that display one tone — + use get_ctcss_tones() for the full set. """ # Check single frequencies first for frequency in self.frequencies: - if frequency.is_single and abs(frequency.single - rf_freq) < 1e-4: - if frequency.ctcss is not None: - return frequency.ctcss + if (frequency.is_single and frequency.single is not None + and abs(frequency.single - rf_freq) < FREQ_MATCH_TOLERANCE_MHZ + and frequency.tones): + return frequency.tones[0].ctcss # Then check ranges for frequency in self.frequencies: - if not frequency.is_single and frequency.lo <= rf_freq <= frequency.hi: - if frequency.ctcss is not None: - return frequency.ctcss + if (not frequency.is_single and frequency.lo is not None and frequency.hi is not None + and frequency.lo <= rf_freq <= frequency.hi + and frequency.tones): + return frequency.tones[0].ctcss return None @@ -606,15 +742,21 @@ def get_ctcss_tones(self, rf_freq: float) -> list[float]: """ # Check single frequencies first for frequency in self.frequencies: - if frequency.is_single and abs(frequency.single - rf_freq) < 1e-4: - if frequency.ctcss_tones: - return list(frequency.ctcss_tones) + if frequency.is_single and frequency.single is not None and abs(frequency.single - rf_freq) < FREQ_MATCH_TOLERANCE_MHZ: + tones = list(frequency.ctcss_tones) + for tone_rule in frequency.tones: + if tone_rule.ctcss not in tones: + tones.append(tone_rule.ctcss) + return tones # Then check ranges for frequency in self.frequencies: - if not frequency.is_single and frequency.lo <= rf_freq <= frequency.hi: - if frequency.ctcss_tones: - return list(frequency.ctcss_tones) + if not frequency.is_single and frequency.lo is not None and frequency.hi is not None and frequency.lo <= rf_freq <= frequency.hi: + tones = list(frequency.ctcss_tones) + for tone_rule in frequency.tones: + if tone_rule.ctcss not in tones: + tones.append(tone_rule.ctcss) + return tones return [] @@ -660,31 +802,62 @@ def generate_baseband_frequencies(self) -> None: self.center_freq, self.channel_spacing) - def get_label(self, rf: float, ctcss: float | None = None) -> str | None: + def _bank_label(self, freq_entry: FrequencyInfo, banks: list[str] | None) -> str | None: + """Pick the per-bank display label for a matched entry. + + The winning bank is the first resolved tag that is active and carries + a per-bank label (active mode); otherwise the first resolved tag with + a per-bank label (promiscuous mode). Returns None when the entry has + no per-bank labels or none of the resolved tags apply. + """ + if not banks or not freq_entry.bank_labels: + return None + for bank in banks: + if bank in self.active_banks and bank in freq_entry.bank_labels: + return freq_entry.bank_labels[bank] + for bank in banks: + if bank in freq_entry.bank_labels: + return freq_entry.bank_labels[bank] + return None + + def get_label(self, rf: float, ctcss: float | None = None, banks: list[str] | None = None) -> str | None: ''' Get the label for a frequency. If there is not a label for the frequency then return the label for the range of frequencies (if any) + Label precedence for a matched entry: + 1. Per-bank display label (from the resolved ``banks`` tags) + 2. Matched per-tone label + 3. Entry label + Args: rf (float): Radio frequency of tuned channel ctcss (float, optional): Matched CTCSS tone frequency + banks (list[str], optional): Resolved bank tags for the hit, used + to select a per-bank display label when one is configured ''' range_label: str | None = None for freq_entry in self.frequencies: if freq_entry.is_single: - if freq_entry.single == rf: - if ctcss is not None and ctcss in freq_entry.ctcss_tones: - idx = freq_entry.ctcss_tones.index(ctcss) - if idx < len(freq_entry.ctcss_labels): - return freq_entry.ctcss_labels[idx] + if freq_entry.single is not None and abs(freq_entry.single - rf) < FREQ_MATCH_TOLERANCE_MHZ: + bank_label = self._bank_label(freq_entry, banks) + if bank_label is not None: + return bank_label + if ctcss is not None: + for tone_rule in freq_entry.tones: + if abs(tone_rule.ctcss - ctcss) < TONE_MATCH_TOLERANCE_HZ and tone_rule.label: + return tone_rule.label return freq_entry.label else: - if freq_entry.lo <= rf <= freq_entry.hi: + if freq_entry.lo is not None and freq_entry.hi is not None and freq_entry.lo <= rf <= freq_entry.hi: range_label = freq_entry.label - if ctcss is not None and ctcss in freq_entry.ctcss_tones: - idx = freq_entry.ctcss_tones.index(ctcss) - if idx < len(freq_entry.ctcss_labels): - range_label = freq_entry.ctcss_labels[idx] + bank_label = self._bank_label(freq_entry, banks) + if bank_label is not None: + range_label = bank_label + elif ctcss is not None: + for tone_rule in freq_entry.tones: + if abs(tone_rule.ctcss - ctcss) < TONE_MATCH_TOLERANCE_HZ and tone_rule.label: + range_label = tone_rule.label return range_label diff --git a/apps/h2m_parser.py b/apps/h2m_parser.py index 71d51f0..bbdbcf4 100644 --- a/apps/h2m_parser.py +++ b/apps/h2m_parser.py @@ -77,6 +77,7 @@ class CliMapping: CliMapping("frequency_file_name", "frequency_policies", "file", Path), CliMapping("disable_lockout", "frequency_policies", "disable_lockout", bool), CliMapping("disable_priority", "frequency_policies", "disable_priority", bool), + CliMapping("active_banks", "frequency_policies", "active_banks", list), # Display CliMapping("max_db", "display", "max_db", float), @@ -227,6 +228,9 @@ def __init__(self, args: Optional[List[str]] = None) -> None: dest="threshold_db", default=None, help="Threshold in dB") + parser.add_argument("--banks", nargs="+", dest="active_banks", default=None, + help="Active scanner banks to monitor (e.g. --banks FRS_FAMILY SECURITY)") + parser.add_argument("-w", "--write", dest="record", action="store_true", default=None, help="Record (write) channels to disk") @@ -311,6 +315,10 @@ def __init__(self, args: Optional[List[str]] = None) -> None: dest="max_ctcss_tones", default=None, help="Maximum number of CTCSS tones configured per frequency") + parser.add_argument("--list-banks", action="store_true", default=False, + dest="list_banks", + help="Print each configured bank with its channel members, then exit without scanning") + if args is not None: options = parser.parse_args(args) else: @@ -337,6 +345,7 @@ def __init__(self, args: Optional[List[str]] = None) -> None: parser.error(str(err)) self.frequency_params = self._build_frequency_params() + self.list_banks = bool(options.list_banks) def _merge_cli_options(self, raw: dict[str, Any], options) -> dict[str, Any]: """Layer non-None CLI options onto the raw YAML dict, keyed by CLI_OPTION_MAP.""" diff --git a/apps/ham2mon.py b/apps/ham2mon.py index 2b21a7c..adc3c38 100644 --- a/apps/ham2mon.py +++ b/apps/ham2mon.py @@ -20,6 +20,8 @@ from os.path import realpath, dirname import _curses +from frequency_manager import FrequencyConfiguration, FrequencyManager +from utilities import parse_bank_entry logger = logging.getLogger("ham2mon") @@ -109,7 +111,11 @@ async def make_display(self) -> None: self.rxwin.record = self.scanner.record self.rxwin.type_demod = PARSER.master_config.receiver.mode self.rxwin.frequency_file_name = self.scanner.frequency_file_name + if self.scanner is not None: + self.rxwin.banks = self.scanner.frequency_manager.active_banks + self.rxwin.bank_labels = self.scanner.frequency_manager.bank_display_labels self.rxwin.activity_type = self.scanner.activity_params.type + # not all activity types use a dest if self.scanner.activity_params.type == 'fixed-field': dest = self.scanner.activity_params.dest @@ -145,6 +151,11 @@ async def cycle(self) -> None: self.chanwin.draw_channels(self.scanner.channels) self.specwin.draw_spectrum(self.scanner.spectrum, self.scanner.channels, self.chanwin.get_row_map()) self.lockoutwin.draw_channels(self.scanner.frequencies, self.scanner.channels) + # Refresh active banks each cycle so runtime changes ('b' key) show up + # on the next draw without requiring a window rebuild. + if self.scanner is not None: + self.rxwin.banks = self.scanner.frequency_manager.active_banks + self.rxwin.bank_labels = self.scanner.frequency_manager.bank_display_labels self.rxwin.draw_rx() # Update physical screen via optimized double buffering @@ -156,6 +167,11 @@ async def init_scanner(self) -> scnr.Scanner: scanner = scnr.Scanner(PARSER.master_config, frequency_params) + if PARSER.master_config.frequency_policies.active_banks: + scanner.frequency_manager.set_active_banks( + PARSER.master_config.frequency_policies.active_banks + ) + await scanner.load_frequencies() # Set the parameters scanner.set_center_freq(scanner.center_freq) @@ -175,6 +191,15 @@ def center_freq_changed(self): self.rxwin.steps = self.scanner.steps async def handle_char(self, keyb: int) -> None: + # Bank-entry mode consumes all keystrokes until Enter/ESC. This early + # return keeps bank entry and frequency entry mutually exclusive and + # suppresses every other handler while editing. + if self.rxwin.bank_entry is not None: + text = self.rxwin.bank_entry + if self.rxwin.proc_keyb_bank_entry(keyb) and self.scanner is not None: + self.scanner.set_active_banks(parse_bank_entry(text)) + return + # Send keystroke to spectrum window and update scanner if True if self.specwin.proc_keyb(keyb): self.scanner.set_threshold(self.specwin.threshold_db) @@ -214,6 +239,45 @@ async def display_main(stdscr) -> None: def main(stdscr) -> None: return asyncio.run(display_main(stdscr)) +async def list_banks() -> None: + """Print each configured bank with its channel members, then exit. + + A non-curses audit mode for --list-banks: builds a FrequencyManager (no + SDR, receiver, or component pipeline is touched) so bank membership can + be previewed before running the scanner. + """ + cfg = PARSER.master_config + frequency_manager = FrequencyManager( + FrequencyConfiguration( + file_name=cfg.frequency_policies.file, + disable_lockout=cfg.frequency_policies.disable_lockout, + disable_priority=cfg.frequency_policies.disable_priority, + max_ctcss_tones=cfg.receiver.max_ctcss_tones, + ), + cfg.receiver.channel_spacing, + ) + await frequency_manager.load() + + configured = frequency_manager.configured_banks() + if not configured: + print("No banks configured.") + return + + for bank in sorted(configured): + display_label = frequency_manager.bank_display_labels.get(bank) + header = f"{bank} - {display_label}" if display_label else bank + members = frequency_manager.bank_members(bank) + print(f"{header} ({len(members)} channel(s))") + for member in members: + if member.is_single and member.single is not None: + desc = f"{member.single:.4f} MHz" + else: + desc = f"{member.lo:.4f}-{member.hi:.4f} MHz" + name = member.label if member.label else "(no label)" + if member.bank_labels and bank in member.bank_labels: + name = f"{member.bank_labels[bank]} ({name})" + print(f" {desc} {name}") + if __name__ == '__main__': try: @@ -256,7 +320,10 @@ def main(stdscr) -> None: # Suppress chatty third-party loggers that would otherwise pollute output logging.getLogger("urllib3").setLevel(logging.WARNING) - wrapper(main) + if PARSER.list_banks: + asyncio.run(list_banks()) + else: + wrapper(main) except KeyboardInterrupt: pass except RuntimeError as error: diff --git a/apps/scanner.py b/apps/scanner.py index e046cca..125c505 100644 --- a/apps/scanner.py +++ b/apps/scanner.py @@ -12,18 +12,16 @@ import threading import time from dataclasses import dataclass, field -from pathlib import Path import estimate import h2m_parser as prsr import numpy as np import receiver as recvr -from components.base import ChannelInfo -from components.manager import ComponentManager from center_frequency_provider import FrequencyGroup, FrequencyProvider from channel_loggers import ActivityLogger, ActivityParams, ChannelMessage +from components.base import ChannelInfo +from components.manager import ComponentManager from config import GainConfig, MasterHam2MonConfig - from frequency_manager import ( ChannelFrequency, ChannelList, @@ -44,7 +42,6 @@ wav_duration_sec, ) - logger = logging.getLogger(f"ham2mon.{__name__}") @dataclass(kw_only=True) @@ -300,6 +297,12 @@ def _get_signal_strength(self, bb: int) -> float: return 10.0 * np.log10(power) - 70.0 + # active_banks can change at runtime (Scanner.set_active_banks / the 'b' + # key in the TUI). Bank filtering only gates assignment in + # _assign_channels_to_demodulators(), so a demodulator already tuned to a + # now-inactive-bank channel keeps running here until the transmission + # ends naturally; its recording is then discarded by the is_bank_active + # check in _process_completed_transmission. async def _process_current_demodulators(self, channels: ChannelList) -> None: the_now = time.time() @@ -362,6 +365,12 @@ async def _assign_channels_to_demodulators(self, channels: ChannelList) -> None: # If channel not in demodulators if channel.bb not in self.receiver.get_demod_freqs() and not channel.locked: + # Skip if channel's resolved bank tags are not active in current bank filter + # selection. Promiscuous mode (no --banks) is always active, so skip the whole + # check to avoid a per-cycle resolve_banks() scan for non-bank users. + if self.frequency_manager.active_banks and not self.frequency_manager.is_bank_active( + self.frequency_manager.resolve_banks(channel.rf, channel.matched_ctcss)): + continue # Sequence through each demodulator for idx in range(len(self.receiver.demodulators)): demodulator = self.receiver.demodulators[idx] @@ -407,6 +416,15 @@ def _add_metadata(self, active_channels: NDArray) -> ChannelList: if channel in demod_map: matched_tone = demod_map[channel].matched_ctcss_tone + # Resolve bank tags for the CHANNELS panel display. Gated on a + # non-empty active_banks to mirror the promiscuous short-circuit + # in _assign_channels_to_demodulators() and avoid a per-cycle + # resolve_banks() scan for non-bank users. + banks = ( + self.frequency_manager.resolve_banks(frequency, matched_tone) + if self.frequency_manager.active_banks else [] + ) + idx = 0 if priority is not None else len(sweep) # priority channels up front sweep.insert(idx, ChannelFrequency(bb=channel, rf=frequency, @@ -414,10 +432,10 @@ def _add_metadata(self, active_channels: NDArray) -> ChannelList: active=is_active, priority=priority, hanging=is_hanging, - ctcss=self.frequency_manager.get_ctcss_info(frequency), matched_ctcss=matched_tone, - label=self.frequency_manager.get_label(frequency, matched_tone), - ctcss_tones=self.frequency_manager.get_ctcss_tones(frequency))) + label=self.frequency_manager.get_label(frequency, matched_tone, banks), + ctcss_tones=self.frequency_manager.get_ctcss_tones(frequency), + banks=banks)) return sweep @@ -507,6 +525,16 @@ def set_threshold(self, threshold_db: int) -> None: """ self.threshold_db = threshold_db + def set_active_banks(self, banks: list[str] | None) -> None: + """Set the active bank tags at runtime. + + Empty or None restores promiscuous scan-all mode. Unknown tags are + warned about since bank filtering is fail-closed (a typo silently + disables every channel). + """ + self.frequency_manager.set_active_banks(banks) + self.frequency_manager._warn_on_unmatched_active_banks() + def _process_completed_transmission( self, msg: ChannelMessage ) -> tuple[ChannelMessage, TransmissionRecord | None]: @@ -539,6 +567,15 @@ def _process_completed_transmission( msg.detail = 'Discarded mismatched CTCSS' return msg, None + # 1b. Active bank selection discard: if the final resolved bank tags for this transmission do not match active_banks + resolved_banks = msg.banks or self.frequency_manager.resolve_banks(msg.rf, msg.matched_ctcss) + if not self.frequency_manager.is_bank_active(resolved_banks): + _delete_file(tmp_path, "inactive bank selection") + msg.detail = 'Discarded inactive bank selection' + return msg, None + + + # 2. Minimum duration check: reject recordings shorter than min_recording_sec. # wav_bytes_per_sec encapsulates DEFAULT_AUDIO_RATE and bit-depth, keeping # this formula consistent with the duration_sec calculation below. @@ -552,6 +589,7 @@ def _process_completed_transmission( # 3. Component evaluation (WavGatekeeper) classification: str | None = None comp_metadata: dict[str, object] = {} + if self._component_manager.has_wav_component(): info = ChannelInfo( @@ -564,6 +602,7 @@ def _process_completed_transmission( signal_db=msg.signal_db, timestamp=msg.started_at or time.time(), wav_tmp_path=tmp_path, + banks=resolved_banks, ) res = self._component_manager.process_wav(tmp_path, info) classification = res.classification @@ -614,6 +653,7 @@ def _process_completed_transmission( "rf": msg.rf, "channel": msg.channel, "label": msg.label, + "banks": resolved_banks, "classification": classification, "duration_sec": duration_sec, "metadata": comp_metadata, @@ -636,6 +676,7 @@ def _process_completed_transmission( started_at=started_at, duration_sec=duration_sec, metadata=comp_metadata, + banks=resolved_banks, ) return msg, record @@ -644,7 +685,7 @@ async def got_channel_activity(self, msg: ChannelMessage) -> None: This callback is to let the demodulators inform us about a transmission. - 1. Embellish metadata FIRST (label/priority) + 1. Embellish metadata FIRST (label/priority/banks) 2. Process recording persistence / classification SECOND 3. Log activity via channel logger 4. If channel is interesting, notify frequency provider @@ -654,8 +695,9 @@ async def got_channel_activity(self, msg: ChannelMessage) -> None: if msg is None: return - # 1. Embellish metadata FIRST so label/priority are present for persistence & classification - msg.label = self.frequency_manager.get_label(msg.rf, msg.matched_ctcss) + # 1. Embellish metadata FIRST so label/priority/banks are present for persistence & classification + msg.banks = self.frequency_manager.resolve_banks(msg.rf, msg.matched_ctcss) + msg.label = self.frequency_manager.get_label(msg.rf, msg.matched_ctcss, msg.banks) msg.priority = self.frequency_manager.is_priority(msg.bb) # 2. Process recording persistence / classification SECOND diff --git a/apps/tests/test_channel_loggers.py b/apps/tests/test_channel_loggers.py index 9407cdc..9b8f7ec 100644 --- a/apps/tests/test_channel_loggers.py +++ b/apps/tests/test_channel_loggers.py @@ -18,7 +18,7 @@ async def test_fixed_field_logger(tmp_path): params = ActivityParams(type="fixed-field", dest=str(log_file), interval=0) logger = FixedField(params) - # 1. Message WITH matched CTCSS + # 1. Message WITH matched CTCSS and banks msg1 = ChannelMessage( state="on", rf=145.5, @@ -27,11 +27,12 @@ async def test_fixed_field_logger(tmp_path): priority=1, classification="V", matched_ctcss=100.0, - file="test1.wav" + file="test1.wav", + banks=["NETWORK_A", "NETWORK_B"], ) await logger.log(msg1) - # 2. Message WITHOUT matched CTCSS + # 2. Message WITHOUT matched CTCSS or banks msg2 = ChannelMessage( state="off", rf=145.5, @@ -40,7 +41,7 @@ async def test_fixed_field_logger(tmp_path): priority=None, classification=None, matched_ctcss=None, - file="test2.wav" + file="test2.wav", ) await logger.log(msg2) @@ -48,10 +49,14 @@ async def test_fixed_field_logger(tmp_path): lines = log_file.read_text().splitlines() assert len(lines) == 2 - # Verify formatting of msg1: matched_ctcss '100.0 ' should be present right before the filename + # Verify formatting of msg1: matched_ctcss '100.0 ' and the banks column + # truncated to 15 chars ('NETWORK_A,NETWO'); the rest of the joined bank + # list must be dropped so the record stays fixed-width. line1 = lines[0] - # Check that we can find the tone and filename in the correct order/formatting + # Check that we can find the tone, truncated banks, and filename in the correct order/formatting assert "100.0 " in line1 + assert "NETWORK_A,NETWO" in line1 + assert "NETWORK_B" not in line1 assert "test1.wav" in line1 # Verify formatting of msg2: matched_ctcss is empty/omitted @@ -80,7 +85,8 @@ async def test_json_to_server_logger(): priority=1, classification="V", matched_ctcss=141.3, - file="test.wav" + file="test.wav", + banks=["NETWORK_A"], ) await logger.log(msg) @@ -94,6 +100,7 @@ async def test_json_to_server_logger(): assert posted_json["state"] == "on" assert posted_json["rf"] == 145.5 assert posted_json["file"] == "test.wav" + assert posted_json["banks"] == ["NETWORK_A"] class SpyLogger(ActivityLogger): diff --git a/apps/tests/test_component_base.py b/apps/tests/test_component_base.py index 6bcbf14..e944a88 100644 --- a/apps/tests/test_component_base.py +++ b/apps/tests/test_component_base.py @@ -65,7 +65,7 @@ def test_channel_info_immutable(): rf=460.125, bb_hz=0, channel=0, - label="Fire", + label="Net A", priority=1, matched_ctcss_hz=156.7, signal_db=-65, @@ -73,7 +73,7 @@ def test_channel_info_immutable(): wav_tmp_path="/tmp/test.wav", ) assert info.rf == 460.125 - assert info.label == "Fire" + assert info.label == "Net A" def test_component_recover(): diff --git a/apps/tests/test_frequency_manager.py b/apps/tests/test_frequency_manager.py index 3e012cd..2acd156 100644 --- a/apps/tests/test_frequency_manager.py +++ b/apps/tests/test_frequency_manager.py @@ -1,8 +1,16 @@ +import logging +from pathlib import Path + import pytest from frequency_manager import ( - FrequencyManager, FrequencyConfiguration, + ChannelMessage, + ConfigFrequency, + FrequencyConfiguration, + FrequencyInfo, + FrequencyManager, + ToneRule, + TransmissionRecord, ) -from pathlib import Path TEST_DIR = Path(__file__).parent @@ -48,8 +56,8 @@ async def load_inline_config(frequency_manager: FrequencyManager, *entries: dict it's meant to represent, e.g.: await load_inline_config(fm, - {'label': 'Security Patrol dispatch', 'single': 462.400, 'ctcss': 100.0}, - {'label': 'Security Patrol dispatch (Backup)', 'single': 462.400, 'ctcss': 67.0}, + {'label': 'Security Patrol dispatch', 'single': 462.400, 'tones': [100.0]}, + {'label': 'Security Patrol dispatch (Backup)', 'single': 462.400, 'tones': [67.0]}, ) Prefer this over calling frequency_manager.add() directly when the test's @@ -101,6 +109,34 @@ async def test_file_format_conditions(file, expected_exception, message): await FrequencyManager(config, CHANNEL_SPACING).load() +@pytest.mark.asyncio +async def test_full_example_frequencies_file_loads(): + """The progressive full-example frequency file must stay loadable.""" + config = FrequencyConfiguration( + file_name=Path(__file__).parents[2] / "doc" / "full-example.freqs.yaml", + disable_lockout=False, disable_priority=False, max_ctcss_tones=3) + fm = FrequencyManager(config, CHANNEL_SPACING) + freqs = await fm.load() + assert len(freqs) >= 15 + by_label = {f.label: f for f in freqs} + + # Entry-level bank as a plain string is coerced to a list. + assert by_label["Logistics net"].banks == ["LOG"] + + # Per-tone banks resolve per tone rule; a bare tone stays untagged. + multi_role = by_label["Multi-role repeater"] + assert multi_role.tones[0].banks == ["DISPATCH"] + assert multi_role.tones[1].ctcss == 127.3 + assert multi_role.tones[1].banks == [] + + # Kitchen-sink range: banks + two tones + priority on a range entry. + wide_area = by_label["Wide-area field operations"] + assert not wide_area.is_single + assert wide_area.banks == ["FIELD"] + assert [tone.ctcss for tone in wide_area.tones] == [110.9, 123.0] + assert wide_area.priority == 5 + + @pytest.mark.asyncio async def test_file_load_no_errors(fm_with_entries): await fm_with_entries.load() @@ -260,91 +296,47 @@ async def test_fail_duplicate_add_single_frequency(fm_empty): Real-world config sometimes needs more than one CTCSS tone valid for the same physical frequency (e.g. a repeater that answers to a primary and a backup PL -tone). Rather than requiring a new YAML shape, this is expressed by declaring -the same frequency/range twice, differing only by `ctcss` -- mirroring how -users already tend to write these configs: +tone). Configure them on a single entry with the unified ``tones:`` array: - label: "Security Patrol dispatch" single: 462.400 priority: 1 - ctcss: 100.0 - - label: "Security Patrol dispatch (Backup)" - single: 462.400 - priority: 1 - ctcss: 67.0 -""" - - -@pytest.mark.asyncio -async def test_add_merges_second_ctcss_tone_for_same_frequency(fm_empty): - - FREQ = 462.400 - - await fm_empty.add({'label': 'Security Patrol dispatch', - 'single': FREQ, 'priority': 1, 'ctcss': 100.0}) - frequencies = await fm_empty.add({'label': 'Security Patrol dispatch (Backup)', - 'single': FREQ, 'priority': 1, 'ctcss': 67.0}) - - # No new entry was added -- the second declaration was merged into the first - assert len(frequencies) == 1 + tones: + - ctcss: 100.0 # in Hz + - ctcss: 67.0 - merged = frequencies[LAST_ENTRY] - assert merged.label == 'Security Patrol dispatch' # first entry's label wins - assert merged.ctcss == 100.0 # primary tone, back-compat - assert merged.ctcss_tones == [100.0, 67.0] - - -@pytest.mark.asyncio -async def test_add_merges_third_ctcss_tone_for_same_frequency(fm_empty): - - FREQ = 462.400 - - await fm_empty.add({'label': 'Primary', 'single': FREQ, 'ctcss': 100.0}) - await fm_empty.add({'label': 'Backup 1', 'single': FREQ, 'ctcss': 67.0}) - frequencies = await fm_empty.add({'label': 'Backup 2', 'single': FREQ, 'ctcss': 82.5}) - - assert len(frequencies) == 1 - assert frequencies[LAST_ENTRY].ctcss_tones == [100.0, 67.0, 82.5] - - -@pytest.mark.asyncio -async def test_add_merges_ctcss_tone_for_same_range(fm_empty): - - FREQ = 450.0 - - await fm_empty.add({'label': 'Primary', 'lo': FREQ, 'hi': FREQ+1, 'ctcss': 100.0}) - frequencies = await fm_empty.add({'label': 'Backup', 'lo': FREQ, 'hi': FREQ+1, 'ctcss': 67.0}) +Declaring the same frequency twice is an error, even when the tone differs — +multiple tones must live on one entry. +""" - assert len(frequencies) == 1 - assert frequencies[LAST_ENTRY].ctcss_tones == [100.0, 67.0] @pytest.mark.asyncio -async def test_fail_duplicate_ctcss_tone_same_value(fm_empty): - """Repeating the exact same tone for the same frequency is a real duplicate, not a merge.""" +async def test_fail_duplicate_frequency(fm_empty): + """Declaring the same frequency twice is always an error. Use tones: [...] for multiple tones.""" FREQ = 462.400 - entry = {'label': 'Security Patrol dispatch', 'single': FREQ, 'ctcss': 100.0} - + entry = {'label': 'Security Patrol dispatch', 'single': FREQ, 'tones': [100.0]} await fm_empty.add(entry) + # Same entry exactly → error with pytest.raises(ValueError, match='already occurs in list'): await fm_empty.add(entry) + # Same frequency, different ctcss → also an error now + with pytest.raises(ValueError, match='already occurs in list'): + await fm_empty.add({'label': 'Backup', 'single': FREQ, 'tones': [67.0]}) + @pytest.mark.asyncio -async def test_fail_merge_ctcss_conflicting_priority(fm_empty): - """A duplicate-frequency entry with a different explicit priority is ambiguous config.""" - - FREQ = 462.400 +async def test_fail_duplicate_range(fm_empty): + """Declaring the same range twice is always an error.""" - await fm_empty.add({'label': 'Security Patrol dispatch', - 'single': FREQ, 'priority': 1, 'ctcss': 100.0}) + await fm_empty.add({'label': 'Primary', 'lo': 450.0, 'hi': 451.0, 'tones': [100.0]}) - with pytest.raises(ValueError, match='conflicts with existing priority'): - await fm_empty.add({'label': 'Security Patrol dispatch (Backup)', - 'single': FREQ, 'priority': 2, 'ctcss': 67.0}) + with pytest.raises(ValueError, match='already occurs in list'): + await fm_empty.add({'label': 'Backup', 'lo': 450.0, 'hi': 451.0, 'tones': [67.0]}) @pytest.mark.asyncio @@ -352,14 +344,37 @@ async def test_get_ctcss_tones_returns_all_configured_tones(fm_empty): FREQ = 462.400 - await fm_empty.add({'label': 'Primary', 'single': FREQ, 'ctcss': 100.0}) - await fm_empty.add({'label': 'Backup', 'single': FREQ, 'ctcss': 67.0}) + await fm_empty.add({ + 'single': FREQ, + 'label': 'Security Patrol', + 'tones': [ + {'ctcss': 100.0, 'label': 'Primary'}, + {'ctcss': 67.0, 'label': 'Backup'}, + ] + }) fm_empty.set_center(FREQ*1e6) assert fm_empty.get_ctcss_tones(FREQ) == [100.0, 67.0] + +@pytest.mark.asyncio +async def test_get_ctcss_tones_returns_tones_from_tones_array(fm_empty): + """get_ctcss_tones must return CTCSS tones specified under the unified tones: [...] array.""" + await fm_empty.add({ + 'single': 467.7125, + 'label': 'Base', + 'banks': ['FRS_FAMILY'], + 'tones': [ + {'ctcss': 67.0, 'label': 'Security', 'banks': ['SECURITY']}, + {'ctcss': 71.9, 'label': 'Operations', 'banks': ['OPERATIONS']}, + ] + }) + tones = fm_empty.get_ctcss_tones(467.7125) + assert set(tones) == {67.0, 71.9} + + @pytest.mark.asyncio async def test_get_ctcss_tones_empty_when_not_configured(fm_empty): @@ -370,33 +385,46 @@ async def test_get_ctcss_tones_empty_when_not_configured(fm_empty): @pytest.mark.asyncio -async def test_get_ctcss_info_still_returns_only_primary_tone(fm_empty): +async def test_get_ctcss_info_returns_first_tone(fm_empty): """ get_ctcss_info() is kept single-valued on purpose for callers (e.g. the - current single-tone squelch) that don't yet support multiple tones. + current single-tone squelch) that don't yet support multiple tones. It + returns the first tone configured under tones:. """ FREQ = 462.400 - await fm_empty.add({'label': 'Primary', 'single': FREQ, 'ctcss': 100.0}) - await fm_empty.add({'label': 'Backup', 'single': FREQ, 'ctcss': 67.0}) + await fm_empty.add({ + 'single': FREQ, + 'label': 'Security Patrol', + 'tones': [ + {'ctcss': 100.0, 'label': 'Primary'}, + {'ctcss': 67.0, 'label': 'Backup'}, + ] + }) assert fm_empty.get_ctcss_info(FREQ) == 100.0 @pytest.mark.asyncio -async def test_process_frequencies_data_merges_duplicate_ctcss_entries(fm_empty): - """End-to-end version of the merge, through the same path a real YAML config load uses.""" +async def test_get_ctcss_info_returns_none_when_no_tones(fm_empty): + """get_ctcss_info returns None for entries without any CTCSS tones.""" + await fm_empty.add({'single': 462.400, 'label': 'No tone'}) + + assert fm_empty.get_ctcss_info(462.400) is None - frequencies = await load_inline_config( - fm_empty, - {'label': 'Security Patrol dispatch', 'single': 462.400, - 'priority': 1, 'ctcss': 100.0}, - {'label': 'Security Patrol dispatch (Backup)', - 'single': 462.400, 'priority': 1, 'ctcss': 67.0}, - ) - assert len(frequencies) == 1 - assert frequencies[0].ctcss_tones == [100.0, 67.0] +@pytest.mark.asyncio +async def test_tones_accepts_bare_floats(fm_empty): + """The unified tones: key accepts bare floats as well as tone rule dicts.""" + await fm_empty.add({ + 'single': 462.400, + 'label': 'Security Patrol', + 'tones': [100.0, {'ctcss': 67.0, 'label': 'Backup'}], + }) + + assert fm_empty.get_ctcss_tones(462.400) == [100.0, 67.0] + assert fm_empty.get_ctcss_info(462.400) == 100.0 + assert fm_empty.get_label(462.400, 67.0) == 'Backup' @pytest.mark.asyncio @@ -999,32 +1027,49 @@ async def test_get_priority_info(fm_empty): @pytest.mark.asyncio -async def test_max_ctcss_tones_limit(fm_empty): - """Test that max_ctcss_tones limits CTCSS tone count and throws validation errors.""" +async def test_max_ctcss_tones_disabled(fm_empty): + """max_ctcss_tones=0 disables CTCSS entirely: adding a frequency with tones: raises.""" from frequency_manager import FrequencyConfiguration, FrequencyManager - # Set up config with max_ctcss_tones = 2 - config_2 = FrequencyConfiguration( + config_0 = FrequencyConfiguration( file_name=None, disable_lockout=False, disable_priority=False, - max_ctcss_tones=2 + max_ctcss_tones=0 ) - fm_2 = FrequencyManager(config_2, channel_spacing=5000) + fm_0 = FrequencyManager(config_0, channel_spacing=5000) + + with pytest.raises(ValueError, match="CTCSS is disabled"): + await fm_0.add({'single': 144.390, 'tones': [100.0], 'label': 'Frequency'}) + - # 1. Add first CTCSS tone: OK - await fm_2.add({'single': 144.390, 'ctcss': 100.0, 'label': 'Frequency'}) - assert fm_2.frequencies[0].ctcss_tones == [100.0] +@pytest.mark.asyncio +async def test_max_ctcss_tones_bypass_via_tones_raises_when_exceeding(fm_empty): + """max_ctcss_tones must cap tones: -only entries. + + Regression guard: the tone count check must key off all configured tones + (including the tones: array) so an entry with more tones than the receiver + supports is rejected at config load rather than truncated silently at + runtime. + """ + with pytest.raises(ValueError, match="max_ctcss_tones"): + await fm_empty.add({ + 'single': 467.7125, + 'label': 'FRS Family', + 'tones': [ + {'ctcss': 67.0, 'label': 'Tone 1'}, + {'ctcss': 71.9, 'label': 'Tone 2'}, + {'ctcss': 74.4, 'label': 'Tone 3'}, + {'ctcss': 77.0, 'label': 'Tone 4'}, + ], + }) - # 2. Add second CTCSS tone (merge): OK - await fm_2.add({'single': 144.390, 'ctcss': 141.3}) - assert fm_2.frequencies[0].ctcss_tones == [100.0, 141.3] - # 3. Add third CTCSS tone (merge): should raise ValueError due to limit - with pytest.raises(ValueError, match="exceeds max_ctcss_tones limit"): - await fm_2.add({'single': 144.390, 'ctcss': 151.4}) +@pytest.mark.asyncio +async def test_max_ctcss_tones_disabled_rejects_tones_only_entry(): + """max_ctcss_tones=0 disables CTCSS even for tones: -only entries.""" + from frequency_manager import FrequencyConfiguration, FrequencyManager - # Set up config with max_ctcss_tones = 0 (CTCSS disabled) config_0 = FrequencyConfiguration( file_name=None, disable_lockout=False, @@ -1033,48 +1078,769 @@ async def test_max_ctcss_tones_limit(fm_empty): ) fm_0 = FrequencyManager(config_0, channel_spacing=5000) - # 4. Adding with CTCSS should fail immediately when max_ctcss_tones is 0 with pytest.raises(ValueError, match="CTCSS is disabled"): - await fm_0.add({'single': 144.390, 'ctcss': 100.0, 'label': 'Frequency'}) - - -@pytest.mark.asyncio -async def test_change_ctcss_merging(fm_empty): - """Test that change() supports CTCSS tone merging and respects max limits.""" - # Default max_ctcss_tones is 3 - # 1. Add a frequency with a CTCSS tone - await fm_empty.add({'single': 144.390, 'ctcss': 100.0, 'label': 'Frequency'}) - assert fm_empty.frequencies[0].ctcss_tones == [100.0] - - # 2. Merge another CTCSS tone via change() - await fm_empty.change({'single': 144.390, 'ctcss': 141.3}) - assert fm_empty.frequencies[0].ctcss_tones == [100.0, 141.3] - - # 3. Merge a third tone via change() - await fm_empty.change({'single': 144.390, 'ctcss': 151.4}) - assert fm_empty.frequencies[0].ctcss_tones == [100.0, 141.3, 151.4] - - # 4. Merging a fourth tone should fail (max limit is 3) - with pytest.raises(ValueError, match="exceeds max_ctcss_tones limit"): - await fm_empty.change({'single': 144.390, 'ctcss': 162.2}) + await fm_0.add({ + 'single': 144.390, + 'label': 'Frequency', + 'tones': [ + {'ctcss': 100.0, 'label': 'Primary'}, + {'ctcss': 141.3, 'label': 'Backup'}, + ], + }) @pytest.mark.asyncio async def test_get_label_with_ctcss(fm_empty): """Test that get_label correctly resolves the label matching a specific CTCSS tone.""" - # 1. Add multiple entries for same frequency with different CTCSS tones and labels - await fm_empty.add({'single': 144.390, 'ctcss': 100.0, 'label': 'Tone 100'}) - await fm_empty.add({'single': 144.390, 'ctcss': 141.3, 'label': 'Tone 141'}) - await fm_empty.add({'single': 144.390, 'ctcss': 151.4, 'label': 'Tone 151'}) - - # 2. Assert that get_label without ctcss returns the primary label + await fm_empty.add({ + 'single': 144.390, + 'label': 'Tone 100', + 'tones': [ + {'ctcss': 100.0, 'label': 'Tone 100'}, + {'ctcss': 141.3, 'label': 'Tone 141'}, + {'ctcss': 151.4, 'label': 'Tone 151'}, + ] + }) + + # get_label without ctcss returns the primary label assert fm_empty.get_label(144.390) == 'Tone 100' - # 3. Assert that get_label with specific ctcss returns the corresponding label + # get_label with specific ctcss returns the corresponding label assert fm_empty.get_label(144.390, 100.0) == 'Tone 100' assert fm_empty.get_label(144.390, 141.3) == 'Tone 141' assert fm_empty.get_label(144.390, 151.4) == 'Tone 151' - # 4. Assert that get_label with unmatched/unknown ctcss falls back to primary label + # get_label with unmatched ctcss falls back to primary label assert fm_empty.get_label(144.390, 88.5) == 'Tone 100' + +def test_tone_rule_validation(): + """Test ToneRule dataclass construction and validation.""" + rule = ToneRule(ctcss=67.0, label="Net A", banks="NET_A") + assert rule.ctcss == 67.0 + assert rule.label == "Net A" + assert rule.banks == ["NET_A"] + + with pytest.raises(ValueError, match="positive number"): + ToneRule(ctcss=-5.0) + + with pytest.raises(ValueError, match="CTCSS must be a float"): + ToneRule(ctcss="invalid") + + +def test_frequency_info_banks_and_tones_normalization(): + """Test FrequencyInfo/ConfigFrequency normalization for banks and tones.""" + # 1. Scalar bank string auto-promotion (FrequencyInfo base class) + info1 = FrequencyInfo(banks="NET_A") + assert info1.banks == ["NET_A"] + + # 2. Bare float tones are coerced to ToneRule (ConfigFrequency only — tones lives there) + info2 = ConfigFrequency(single=462.400, tones=[ToneRule(ctcss=100.0)], banks=["NET_A"]) + assert len(info2.tones) == 1 + assert info2.tones[0].ctcss == 100.0 + assert info2.tones[0].banks == ["NET_A"] + + # 3. Parent bank inheritance for tones omitting banks (ConfigFrequency) + tone_no_bank = ToneRule(ctcss=141.3, label="Parks") + info3 = ConfigFrequency(single=462.400, banks=["COMMERCIAL"], tones=[tone_no_bank]) + assert info3.tones[0].banks == ["COMMERCIAL"] + + +def test_frequency_info_banks_dict_form_normalization(): + """Dict-form banks: keys become membership tags, values become per-bank labels.""" + info = ConfigFrequency(single=462.400, banks={"AREA_A": "Net A", "AREA_B": "Net B"}) + assert info.banks == ["AREA_A", "AREA_B"] + assert info.bank_labels == {"AREA_A": "Net A", "AREA_B": "Net B"} + + # List form leaves bank_labels unset (zero overhead for existing configs) + info_list = ConfigFrequency(single=462.400, banks=["NET_A"]) + assert info_list.banks == ["NET_A"] + assert info_list.bank_labels is None + + # String form is unchanged + info_str = FrequencyInfo(banks="NET_A") + assert info_str.banks == ["NET_A"] + assert info_str.bank_labels is None + + # Dict-form keys normalize to strings + info_int = ConfigFrequency(single=462.400, banks={1: "one"}) + assert info_int.banks == ["1"] + assert info_int.bank_labels == {"1": "one"} + + # Tone inheritance copies bank keys only, not labels + tone = ToneRule(ctcss=141.3, label="Parks") + info_tone = ConfigFrequency(single=462.400, banks={"AREA_A": "Net A"}, tones=[tone]) + assert info_tone.tones[0].banks == ["AREA_A"] + + +@pytest.mark.asyncio +async def test_add_dict_form_banks_does_not_break_duplicate_detection(fm_empty): + """Custom __eq__ compares only frequency values; dict-form banks don't create dupes.""" + await fm_empty.add({'single': 462.400, 'banks': {'AREA_A': 'Net A'}}) + with pytest.raises(ValueError, match='already occurs'): + await fm_empty.add({'single': 462.400, 'banks': {'AREA_B': 'Net B'}}) + assert len(fm_empty.frequencies) == 1 + + +@pytest.mark.asyncio +async def test_get_label_per_bank_overrides_tone_label(fm_empty): + """Per-bank label beats per-tone label for a matched entry.""" + await fm_empty.add({ + 'single': 462.400, + 'label': 'Base', + 'banks': {'AREA_A': 'Net A', 'AREA_B': 'Net B'}, + 'tones': [ + {'ctcss': 100.0, 'label': 'Tone 100'}, + {'ctcss': 141.3, 'label': 'Tone 141'}, + ], + }) + + # Bank label wins over tone label when the resolved bank carries one + assert fm_empty.get_label(462.400, 100.0, ['AREA_A']) == 'Net A' + assert fm_empty.get_label(462.400, 141.3, ['AREA_B']) == 'Net B' + + # Unresolved / unmatched bank falls back to the tone label + assert fm_empty.get_label(462.400, 100.0, ['OTHER']) == 'Tone 100' + + # No banks argument: existing behavior preserved + assert fm_empty.get_label(462.400, 100.0) == 'Tone 100' + assert fm_empty.get_label(462.400) == 'Base' + + +@pytest.mark.asyncio +async def test_get_label_per_bank_active_preference(fm_empty): + """When several resolved tags carry labels, the active bank is preferred.""" + await fm_empty.add({ + 'single': 462.400, + 'label': 'Base', + 'banks': {'AREA_A': 'Net A', 'AREA_B': 'Net B'}, + }) + fm_empty.set_active_banks({'AREA_B'}) + assert fm_empty.get_label(462.400, None, ['AREA_A', 'AREA_B']) == 'Net B' + fm_empty.set_active_banks(set()) + # Promiscuous mode: first resolved tag with a label wins + assert fm_empty.get_label(462.400, None, ['AREA_A', 'AREA_B']) == 'Net A' + + +@pytest.mark.asyncio +async def test_get_range_label_with_bank_override(fm_empty): + """Ranges get per-bank labels too; bank beats tone beats entry.""" + await fm_empty.add({ + 'lo': 460.0, 'hi': 468.0, + 'label': 'Region', + 'banks': {'AREA_A': 'Net A'}, + 'tones': [{'ctcss': 100.0, 'label': 'Tone 100'}], + }) + assert fm_empty.get_label(464.0) == 'Region' + assert fm_empty.get_label(464.0, 100.0) == 'Tone 100' + assert fm_empty.get_label(464.0, 100.0, ['AREA_A']) == 'Net A' + assert fm_empty.get_label(464.0, None, ['AREA_A']) == 'Net A' + + +@pytest.mark.asyncio +async def test_bank_display_labels_section_and_members(fm_empty): + """Top-level banks: section is display-only; bank_members audits membership.""" + await fm_empty.process_frequencies_data({ + 'banks': {'AREA_A': 'Net A', 'AREA_B': 'Net B'}, + 'frequencies': [ + {'single': 462.400, 'banks': ['AREA_A']}, + {'single': 463.000, 'banks': ['AREA_B']}, + {'lo': 460.0, 'hi': 468.0, 'banks': ['AREA_A']}, + ], + }) + assert fm_empty.bank_display_labels == {'AREA_A': 'Net A', 'AREA_B': 'Net B'} + + members_a = fm_empty.bank_members('AREA_A') + assert [m.single for m in members_a if m.is_single] == [462.400] + assert len(members_a) == 2 # the single plus the range + assert [m.single for m in fm_empty.bank_members('AREA_B') if m.is_single] == [463.000] + assert fm_empty.bank_members('NOPE') == [] + + +def test_transmission_record_banks_field(): + """Test TransmissionRecord includes banks list.""" + rec = TransmissionRecord( + rf=460.125, + bb_hz=0, + channel=0, + label="Test", + priority=1, + matched_ctcss_hz=67.0, + signal_db=-40, + classification="V", + wav_path="/tmp/test.wav", + started_at=1000.0, + duration_sec=5.0, + banks=["NET_A", "NET_B"], + ) + assert rec.banks == ["NET_A", "NET_B"] + + +# --------------------------------------------------------------------------- +# Bank support: set_active_banks / is_bank_active +# --------------------------------------------------------------------------- + +def test_set_active_banks_accepts_list(fm_empty): + """set_active_banks stores the supplied list as a set.""" + fm_empty.set_active_banks(["NET_A", "RAILROAD"]) + assert fm_empty.active_banks == {"NET_A", "RAILROAD"} + + +def test_set_active_banks_accepts_set(fm_empty): + fm_empty.set_active_banks({"NET_A"}) + assert fm_empty.active_banks == {"NET_A"} + + +def test_set_active_banks_accepts_none(fm_empty): + """None resets to promiscuous (empty set).""" + fm_empty.set_active_banks(["NET_A"]) + fm_empty.set_active_banks(None) + assert fm_empty.active_banks == set() + + +def test_is_bank_active_empty_active_banks_is_promiscuous(fm_empty): + """Empty active_banks means every bank matches (scan-all mode).""" + fm_empty.set_active_banks(None) + assert fm_empty.is_bank_active(["NET_A"]) is True + assert fm_empty.is_bank_active([]) is True + + +def test_is_bank_active_intersection(fm_empty): + fm_empty.set_active_banks(["NET_A"]) + assert fm_empty.is_bank_active(["NET_A", "NET_B"]) is True + assert fm_empty.is_bank_active(["RAILROAD"]) is False + assert fm_empty.is_bank_active([]) is False + + +# --------------------------------------------------------------------------- +# Bank support: resolve_banks() — 5-tier precedence hierarchy +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_resolve_banks_tier1_single_with_matching_tone(fm_empty): + """Tier 1: explicit single entry + matching tone rule → tone rule's banks.""" + await fm_empty.add({ + 'single': 462.5625, + 'banks': ['COMMERCIAL'], + 'tones': [ + {'ctcss': 67.0, 'label': 'Net A 1', 'banks': ['NET_A']}, + {'ctcss': 141.3, 'label': 'City Parks', 'banks': ['PARKS_MAINT']}, + ], + }) + + result = fm_empty.resolve_banks(462.5625, ctcss_hz=67.0) + assert result == ['NET_A'] + + result = fm_empty.resolve_banks(462.5625, ctcss_hz=141.3) + assert result == ['PARKS_MAINT'] + + +@pytest.mark.asyncio +async def test_resolve_banks_tier1_tone_inherits_parent_bank_when_tone_has_no_banks(fm_empty): + """Tier 1: tone rule without its own banks inherits the parent entry's banks.""" + await fm_empty.add({ + 'single': 462.5625, + 'banks': ['COMMERCIAL'], + 'tones': [ + {'ctcss': 141.3, 'label': 'City Parks'}, # no banks key → should inherit COMMERCIAL + ], + }) + + result = fm_empty.resolve_banks(462.5625, ctcss_hz=141.3) + assert result == ['COMMERCIAL'] + + +@pytest.mark.asyncio +async def test_resolve_banks_tier2_single_csq_no_tone(fm_empty): + """Tier 2: explicit single entry, no decoded tone → base entry banks.""" + await fm_empty.add({ + 'single': 460.125, + 'banks': ['NET_A', 'NET_B'], + }) + + result = fm_empty.resolve_banks(460.125, ctcss_hz=None) + assert set(result) == {'NET_A', 'NET_B'} + + +@pytest.mark.asyncio +async def test_resolve_banks_tier2_single_with_tone_no_ctcss_decoded(fm_empty): + """Tier 2: single entry has tone rules, but no CTCSS was decoded → union of base and tone banks returned.""" + await fm_empty.add({ + 'single': 462.5625, + 'banks': ['COMMERCIAL'], + 'tones': [ + {'ctcss': 67.0, 'banks': ['NET_A']}, + ], + }) + + # No ctcss_hz decoded → returns Union of base and tone rules + result = fm_empty.resolve_banks(462.5625, ctcss_hz=None) + assert set(result) == {'COMMERCIAL', 'NET_A'} + + +@pytest.mark.asyncio +async def test_resolve_banks_tier2_single_untagged_fallback(fm_empty): + """Tier 2: single entry exists but has no banks → no tags (promiscuous).""" + await fm_empty.add({'single': 460.050, 'label': 'Untagged channel'}) + + result = fm_empty.resolve_banks(460.050, ctcss_hz=None) + assert result == [] + + +@pytest.mark.asyncio +async def test_resolve_banks_tier3_range_with_matching_tone(fm_empty): + """Tier 3: frequency inside a range + matching tone rule → tone rule's banks.""" + await fm_empty.add({ + 'lo': 462.200, + 'hi': 462.400, + 'banks': ['COMMERCIAL'], + 'tones': [ + {'ctcss': 67.0, 'label': 'Net A Segment', 'banks': ['NET_A']}, + ], + }) + + result = fm_empty.resolve_banks(462.300, ctcss_hz=67.0) + assert result == ['NET_A'] + + +@pytest.mark.asyncio +async def test_resolve_banks_tier4_range_csq(fm_empty): + """Tier 4: frequency inside a range, no tone decoded → range's base banks.""" + await fm_empty.add({ + 'lo': 462.200, + 'hi': 462.400, + 'banks': ['COMMERCIAL'], + }) + + result = fm_empty.resolve_banks(462.300, ctcss_hz=None) + assert result == ['COMMERCIAL'] + + +@pytest.mark.asyncio +async def test_resolve_banks_tier4_range_untagged_fallback(fm_empty): + """Tier 4: frequency inside an untagged range → no tags (promiscuous).""" + await fm_empty.add({'lo': 462.200, 'hi': 462.400, 'label': 'Untagged range'}) + + result = fm_empty.resolve_banks(462.300) + assert result == [] + + +@pytest.mark.asyncio +async def test_resolve_banks_tier5_unconfigured_returns_no_tags_in_promiscuous(fm_empty): + """Tier 5: frequency outside all configured entries → no tags (no SEARCH, + no bank filtering active).""" + await fm_empty.add({'single': 460.125, 'banks': ['NET_A']}) + + result = fm_empty.resolve_banks(999.999) + assert result == [] + + +@pytest.mark.asyncio +async def test_resolve_banks_tier5_unconfigured_returns_search_when_active(fm_empty): + """Tier 5: SEARCH active + unconfigured frequency → ['SEARCH'] returned.""" + await fm_empty.add({'single': 460.125, 'banks': ['NET_A']}) + fm_empty.set_active_banks(['SEARCH']) + + result = fm_empty.resolve_banks(999.999) + assert result == ['SEARCH'] + + +@pytest.mark.asyncio +async def test_resolve_banks_single_takes_precedence_over_range(fm_empty): + """Single entry (Tier 1/2) always beats a containing range (Tier 3/4).""" + await fm_empty.add({'lo': 460.000, 'hi': 461.000, 'banks': ['RAILROAD']}) + await fm_empty.add({'single': 460.500, 'banks': ['NET_A']}) + + result = fm_empty.resolve_banks(460.500) + assert result == ['NET_A'] + + +@pytest.mark.asyncio +async def test_resolve_banks_tone_match_tolerance(fm_empty): + """Tone matching uses a ±0.5 Hz tolerance; exact and near-exact both hit; far miss falls back.""" + await fm_empty.add({ + 'single': 462.5625, + 'banks': ['COMMERCIAL'], + 'tones': [{'ctcss': 67.0, 'banks': ['NET_A']}], + }) + + # Exact match + assert fm_empty.resolve_banks(462.5625, ctcss_hz=67.0) == ['NET_A'] + # Within tolerance + assert fm_empty.resolve_banks(462.5625, ctcss_hz=67.4) == ['NET_A'] + # Outside tolerance → falls back to base entry banks (Tier 2) + assert fm_empty.resolve_banks(462.5625, ctcss_hz=68.0) == ['COMMERCIAL'] + + +# --------------------------------------------------------------------------- +# Bank support: "SEARCH" dynamic tag — opt-in unconfigured spectrum scanning +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_search_tag_in_active_banks_resolves_unconfigured_hits(fm_empty): + """SEARCH in active_banks causes unconfigured spectrum hits to resolve as ['SEARCH'].""" + await fm_empty.add({'single': 460.125, 'banks': ['NET_A']}) + fm_empty.set_active_banks(['NET_A', 'SEARCH']) + + # Configured hit still resolves normally + assert fm_empty.resolve_banks(460.125) == ['NET_A'] + + # Unconfigured hit resolves as SEARCH (not UNTAGGED) + assert fm_empty.resolve_banks(999.000) == ['SEARCH'] + + +@pytest.mark.asyncio +async def test_search_tag_not_active_unconfigured_returns_untagged(fm_empty): + """Without SEARCH in active_banks, unconfigured spectrum hits return UNTAGGED.""" + await fm_empty.add({'single': 460.125, 'banks': ['NET_A']}) + fm_empty.set_active_banks(['NET_A']) + + assert fm_empty.resolve_banks(999.000) == ['UNTAGGED'] + + +@pytest.mark.asyncio +async def test_search_tag_alone_in_active_banks(fm_empty): + """SEARCH as the only active bank: resolve_banks still returns SEARCH for unconfigured + spectrum and resolves configured entries normally.""" + await fm_empty.add({'single': 460.125, 'banks': ['NET_A']}) + fm_empty.set_active_banks(['SEARCH']) + + # resolve_banks: unconfigured hit → SEARCH + assert fm_empty.resolve_banks(999.000) == ['SEARCH'] + + # resolve_banks: configured single still resolves correctly (SEARCH not in active_banks does + # not suppress configured resolution; it only affects Tier 5 unconfigured hits) + assert fm_empty.resolve_banks(460.125) == ['NET_A'] + + +@pytest.mark.asyncio +async def test_search_tag_promiscuous_mode_search_inactive(fm_empty): + """In promiscuous mode (empty active_banks), SEARCH is inactive and + unconfigured spectrum returns no tags.""" + await fm_empty.add({'single': 460.125, 'banks': ['NET_A']}) + fm_empty.set_active_banks(None) # promiscuous + + assert fm_empty.resolve_banks(999.000) == [] + + +@pytest.mark.asyncio +async def test_resolve_banks_untagged_sentinel_gated_on_bank_filtering(fm_empty): + """The UNTAGGED sentinel appears only when bank filtering is active. + + Without --banks (empty active_banks), untagged hits resolve to no tags so + bank metadata stays empty for non-participants. With --banks active, the + same hits resolve to UNTAGGED so is_bank_active() can fail them closed. + """ + await fm_empty.add({'single': 460.125, 'label': 'Untagged channel'}) # Promiscuous / legacy: no sentinel injected + assert fm_empty.resolve_banks(460.125) == [] + assert fm_empty.resolve_banks(999.000) == [] + + # Bank filtering active: untagged hits resolve to the sentinel + fm_empty.set_active_banks(['NET_A']) + assert fm_empty.resolve_banks(460.125) == ['UNTAGGED'] + assert fm_empty.resolve_banks(999.000) == ['UNTAGGED'] + + # Untagged sentinel never matches a non-dynamic active bank (fail-closed) + assert fm_empty.is_bank_active(['UNTAGGED']) is False + + +def test_channel_message_str_omits_bank_bracket_when_no_tags() -> None: + """The syslog debug line (str(msg)) must skip the bank field when no bank + information is available instead of printing [UNTAGGED].""" + no_banks = ChannelMessage(state='off', rf=460.125, bb=0, channel=3) + text = str(no_banks) + assert '[UNTAGGED]' not in text + assert '[NET_A]' not in text + + tagged = ChannelMessage(state='off', rf=460.125, bb=0, channel=3, + banks=['NET_A']) + assert '[NET_A]' in str(tagged) + + +# --------------------------------------------------------------------------- +# Mutation-killing boundary tests (plugging gaps found by mutmut) +# --------------------------------------------------------------------------- + +def test_set_active_banks_accepts_bare_string(fm_empty): + """set_active_banks with a bare string wraps it in a set (isinstance str branch). + Kills mutant that sets active_banks = None instead of {banks}. + """ + fm_empty.set_active_banks("NET_A") + assert fm_empty.active_banks == {"NET_A"} + # Confirm it actually filters correctly — not promiscuous + assert fm_empty.is_bank_active(["NET_A"]) is True + assert fm_empty.is_bank_active(["RAILROAD"]) is False + + +@pytest.mark.asyncio +async def test_resolve_banks_single_proximity_threshold(fm_empty): + """Frequency 200 Hz outside the ±100 Hz (1e-4 MHz) single-match window + must NOT match the entry. Kills mutations that widen the threshold to + <= 1e-4 or < 1.0001 MHz. + """ + await fm_empty.add({'single': 460.1250, 'banks': ['NET_A']}) + + # Exactly on target → must match + assert fm_empty.resolve_banks(460.1250) == ['NET_A'] + # 200 Hz away (0.0002 MHz) → outside 100 Hz window → must NOT match + assert fm_empty.resolve_banks(460.1252) == [] + assert fm_empty.resolve_banks(460.1248) == [] + + +@pytest.mark.asyncio +async def test_resolve_banks_range_boundary_inclusive_at_lo_and_hi(fm_empty): + """A frequency exactly at lo or hi must be inside the range (inclusive <=). + Kills mutations that change lo <= to lo < or <= hi to < hi. + """ + await fm_empty.add({'lo': 462.200, 'hi': 462.400, 'banks': ['COMMERCIAL']}) + + # Exactly at lo — must be inside (<=, not <) + assert fm_empty.resolve_banks(462.200) == ['COMMERCIAL'] + # Exactly at hi — must be inside (<= not <) + assert fm_empty.resolve_banks(462.400) == ['COMMERCIAL'] + # Just outside both ends — must NOT match + assert fm_empty.resolve_banks(462.199) == [] + assert fm_empty.resolve_banks(462.401) == [] + + +@pytest.mark.asyncio +async def test_resolve_banks_single_tone_tolerance_exact_boundary(fm_empty): + """Tone ±0.5 Hz is the exclusive upper boundary: abs(diff) must be + strictly < 0.5 to match. Kills mutations that change < 0.5 to <= 0.5 + or < 1.5 on single-entry tone rules (Tier 1). + """ + await fm_empty.add({ + 'single': 462.5625, + 'banks': ['COMMERCIAL'], + 'tones': [{'ctcss': 67.0, 'banks': ['NET_A']}], + }) + + # 0.49 Hz away — strictly inside window → must match tone + assert fm_empty.resolve_banks(462.5625, ctcss_hz=67.49) == ['NET_A'] + assert fm_empty.resolve_banks(462.5625, ctcss_hz=66.51) == ['NET_A'] + # Exactly 0.5 Hz away — NOT strictly < 0.5 → must NOT match, falls back to base + assert fm_empty.resolve_banks(462.5625, ctcss_hz=67.5) == ['COMMERCIAL'] + assert fm_empty.resolve_banks(462.5625, ctcss_hz=66.5) == ['COMMERCIAL'] + # 0.9 Hz away — well outside → must NOT match (kills < 1.5 widening mutation) + assert fm_empty.resolve_banks(462.5625, ctcss_hz=67.9) == ['COMMERCIAL'] + + +@pytest.mark.asyncio +async def test_resolve_banks_range_tone_tolerance_exact_boundary(fm_empty): + """Same ±0.5 Hz exclusive boundary check, but for range-entry tone rules + (Tier 3). Kills mutations that change < 0.5 to <= 0.5 or < 1.5. + """ + await fm_empty.add({ + 'lo': 462.200, + 'hi': 462.400, + 'banks': ['COMMERCIAL'], + 'tones': [{'ctcss': 67.0, 'banks': ['NET_A']}], + }) + + # 0.49 Hz away — strictly inside window → must match tone + assert fm_empty.resolve_banks(462.300, ctcss_hz=67.49) == ['NET_A'] + assert fm_empty.resolve_banks(462.300, ctcss_hz=66.51) == ['NET_A'] + # Exactly 0.5 Hz away — NOT strictly < 0.5 → falls back to base range banks + assert fm_empty.resolve_banks(462.300, ctcss_hz=67.5) == ['COMMERCIAL'] + assert fm_empty.resolve_banks(462.300, ctcss_hz=66.5) == ['COMMERCIAL'] + # 0.9 Hz away — well outside → must NOT match (kills < 1.5 widening mutation) + assert fm_empty.resolve_banks(462.300, ctcss_hz=67.9) == ['COMMERCIAL'] + + +# --------------------------------------------------------------------------- +# get_label / get_ctcss_info match-tolerance consistency +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_get_label_rf_match_tolerance(fm_empty): + """get_label must match RF within the same strict <1e-4 MHz window as + resolve_banks (not exact equality). Kills regressions to `==` matching. + """ + await fm_empty.add({'single': 460.1250, 'label': 'Patrol'}) + + # Exactly on target → matches + assert fm_empty.get_label(460.1250) == 'Patrol' + # 50 Hz away (0.00005 MHz) → strictly inside window → matches + assert fm_empty.get_label(460.12505) == 'Patrol' + # 200 Hz away (0.0002 MHz) → outside window → no match + assert fm_empty.get_label(460.1252) is None + assert fm_empty.get_label(460.1248) is None + + +@pytest.mark.asyncio +async def test_get_label_uses_tone_tolerance_for_tone_rules(fm_empty): + """get_label tone matching uses the same ±0.5 Hz exclusive tolerance as + resolve_banks: near tones hit, exactly-0.5-away and far tones fall back to + the entry label. Kills regressions that return the exact-value tone label + only (old ctcss_labels path) or widen the tolerance. + """ + await fm_empty.add({ + 'single': 462.5625, + 'label': 'Base', + 'tones': [ + {'ctcss': 67.0, 'label': 'Net A'}, + {'ctcss': 100.0, 'label': 'Net B'}, + ], + }) + + # 0.49 Hz away — strictly inside → matches the tone label + assert fm_empty.get_label(462.5625, 67.49) == 'Net A' + assert fm_empty.get_label(462.5625, 99.51) == 'Net B' + # Exactly 0.5 Hz away — NOT strictly < 0.5 → falls back to entry label + assert fm_empty.get_label(462.5625, 67.5) == 'Base' + # Well outside → falls back to entry label + assert fm_empty.get_label(462.5625, 68.0) == 'Base' + # No ctcss provided → entry label + assert fm_empty.get_label(462.5625) == 'Base' + + +@pytest.mark.asyncio +async def test_get_ctcss_info_tones_only_returns_first_tone(fm_empty): + """get_ctcss_info must surface a primary tone for tones:-only entries.""" + await fm_empty.add({ + 'single': 462.400, + 'label': 'Security Patrol', + 'tones': [ + {'ctcss': 100.0, 'label': 'Primary'}, + {'ctcss': 67.0, 'label': 'Backup'}, + ] + }) + + assert fm_empty.get_ctcss_info(462.400) == 100.0 + + +@pytest.mark.asyncio +async def test_get_ctcss_info_none_when_no_tone(fm_empty): + """get_ctcss_info returns None for entries with no CTCSS tone configured.""" + await fm_empty.add({'single': 462.400, 'label': 'No tone'}) + + assert fm_empty.get_ctcss_info(462.400) is None + assert fm_empty.get_ctcss_info(999.0) is None + + +# --------------------------------------------------------------------------- +# Active-bank startup sanity check (typo detection) +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_unknown_active_banks_matches_configured_set(fm_empty): + """unknown_active_banks must flag only requested banks with no configured + match, across both top-level and tone-rule banks.""" + await fm_empty.add({ + 'single': 462.5625, + 'banks': ['COMMERCIAL'], + 'tones': [ + {'ctcss': 67.0, 'banks': ['NET_A']}, + {'ctcss': 71.9, 'banks': ['SECURITY']}, + ], + }) + await fm_empty.add({'single': 467.7125, 'banks': ['OPERATIONS']}) + + fm_empty.set_active_banks(['NET_A', 'OPERATIONS', 'FIRE_TAK', 'NOPE']) + + assert fm_empty.unknown_active_banks() == {'FIRE_TAK', 'NOPE'} + + +def test_unknown_active_banks_exempts_search(fm_empty): + """SEARCH is a pseudo-bank, not a configured tag — must never be flagged.""" + fm_empty.set_active_banks(['SEARCH']) + assert fm_empty.unknown_active_banks() == set() + + +def test_unknown_active_banks_empty_in_promiscuous_mode(fm_empty): + """Promiscuous mode (no active banks) means nothing to validate.""" + fm_empty.set_active_banks(None) + assert fm_empty.unknown_active_banks() == set() + + +@pytest.mark.asyncio +async def test_load_warns_on_unmatched_active_banks(fm_empty, tmp_path: Path, caplog: pytest.LogCaptureFixture): + """A typo'd active bank (vs. configured tags) must log a WARNING on load().""" + freqs_file = tmp_path / "freqs.yaml" + freqs_file.write_text( + "frequencies:\n" + " - single: 460.125\n" + " label: Patrol\n" + " banks: [NET_A]\n" + ) + + fm_empty.config.file_name = freqs_file + fm_empty.set_active_banks(["FIRE_TAK"]) + + with caplog.at_level(logging.WARNING, logger="ham2mon.frequency_manager"): + await fm_empty.load() + + assert any( + "FIRE_TAK" in message and "match no configured" in message + for message in caplog.messages + ) + + +@pytest.mark.asyncio +async def test_load_no_warning_when_all_active_banks_match(fm_empty, tmp_path: Path, caplog: pytest.LogCaptureFixture): + """Matching active banks produce no startup warning on load().""" + freqs_file = tmp_path / "freqs.yaml" + freqs_file.write_text( + "frequencies:\n" + " - single: 460.125\n" + " label: Patrol\n" + " banks: [NET_A]\n" + ) + + fm_empty.config.file_name = freqs_file + fm_empty.set_active_banks(["NET_A"]) + + with caplog.at_level(logging.WARNING, logger="ham2mon.frequency_manager"): + await fm_empty.load() + + assert not any("match no configured" in message for message in caplog.messages) + + +@pytest.mark.asyncio +async def test_load_warns_on_unmatched_active_banks_without_file(fm_empty, caplog: pytest.LogCaptureFixture): + """--banks X without -F/--frequencies silently resolves everything to + UNTAGGED, which never matches X — load() must warn at startup.""" + fm_empty.set_active_banks(["NET_A"]) + + with caplog.at_level(logging.WARNING, logger="ham2mon.frequency_manager"): + await fm_empty.load() + + assert any( + "NET_A" in message and "match no configured" in message + for message in caplog.messages + ) + + +@pytest.mark.asyncio +async def test_load_no_warning_without_file_when_search_only(fm_empty, caplog: pytest.LogCaptureFixture): + """--banks SEARCH without a frequency file is functional (Tier 5 SEARCH + matches), so no startup warning.""" + fm_empty.set_active_banks(["SEARCH"]) + + with caplog.at_level(logging.WARNING, logger="ham2mon.frequency_manager"): + await fm_empty.load() + + assert not any("match no configured" in message for message in caplog.messages) + + +@pytest.mark.asyncio +async def test_load_no_warning_without_file_when_untagged_only(fm_empty, caplog: pytest.LogCaptureFixture): + """--banks UNTAGGED without a frequency file is functional (Tier 5 UNTAGGED + matches), so no startup warning.""" + fm_empty.set_active_banks(["UNTAGGED"]) + + with caplog.at_level(logging.WARNING, logger="ham2mon.frequency_manager"): + await fm_empty.load() + + assert not any("match no configured" in message for message in caplog.messages) + + +@pytest.mark.asyncio +async def test_active_bank_without_frequencies_resolves_untagged(fm_empty): + """Without a frequency file, every hit resolves to UNTAGGED which never + matches a non-dynamic active bank — the fail-closed no-op that the + load()-time warning exists to surface.""" + fm_empty.set_active_banks(["NET_A"]) + + assert fm_empty.resolve_banks(460.125) == ["UNTAGGED"] + assert fm_empty.is_bank_active(["UNTAGGED"]) is False diff --git a/apps/tests/test_receiver.py b/apps/tests/test_receiver.py index 540862e..2e7a47c 100644 --- a/apps/tests/test_receiver.py +++ b/apps/tests/test_receiver.py @@ -849,7 +849,7 @@ def get_demod_freqs(self): 'single': 144.03, 'priority': 1, 'label': 'Priority 1', - 'ctcss': 100.0 + 'tones': [100.0] }) # Tune the mock demodulator to +30 kHz diff --git a/apps/tests/test_scanner.py b/apps/tests/test_scanner.py index 4d2cd17..4d8be9e 100644 --- a/apps/tests/test_scanner.py +++ b/apps/tests/test_scanner.py @@ -9,7 +9,7 @@ import numpy as np import pytest from conftest import make_test_scanner -from frequency_manager import ChannelMessage, TransmissionRecord +from frequency_manager import ChannelFrequency, ChannelMessage, TransmissionRecord from scanner import Scanner from utilities import ( DEFAULT_AUDIO_RATE, @@ -92,7 +92,7 @@ def test_transmission_record_built_on_kept_wav(tmp_path: Path) -> None: started_at = 1_700_000_000.0 msg = ChannelMessage( state='off', - rf=460_125_000.0, + rf=460.125, bb=0, channel=0, wav_tmp_path=tmp_wav, @@ -136,7 +136,7 @@ def test_transmission_record_duration_set_on_msg(tmp_path: Path) -> None: _write_wav(tmp_wav, num_samples=8_000) # 1-second recording msg = ChannelMessage( - state='off', rf=460_125_000.0, bb=0, channel=0, + state='off', rf=460.125, bb=0, channel=0, wav_tmp_path=tmp_wav, started_at=1_700_000_000.0, ) @@ -160,7 +160,7 @@ def test_transmission_record_none_on_ctcss_discard(tmp_path: Path) -> None: _write_wav(tmp_wav, num_samples=16_000) msg = ChannelMessage( - state='off', rf=460_125_000.0, bb=0, channel=0, + state='off', rf=460.125, bb=0, channel=0, wav_tmp_path=tmp_wav, discard=True, ) scanner = make_test_scanner(wav_dir=wav_dir) @@ -179,7 +179,7 @@ def test_transmission_record_none_on_short_recording(tmp_path: Path) -> None: _write_wav(tmp_wav, num_samples=0) msg = ChannelMessage( - state='off', rf=460_125_000.0, bb=0, channel=0, + state='off', rf=460.125, bb=0, channel=0, wav_tmp_path=tmp_wav, ) scanner = make_test_scanner(wav_dir=wav_dir) @@ -283,7 +283,7 @@ def test_scanner_interesting_matrix( msg = ChannelMessage( state=state, - rf=145_000_000.0, + rf=145.5, bb=0, channel=0, file=msg_file, @@ -291,3 +291,213 @@ def test_scanner_interesting_matrix( ) assert scanner.interesting(msg) is expected_interesting + + +@pytest.mark.asyncio +async def test_assign_channels_skips_inactive_banks(tmp_path: Path) -> None: + """_assign_channels_to_demodulators must skip channels whose resolved banks are not active.""" + wav_dir = str(tmp_path / "wav") + os.makedirs(wav_dir, exist_ok=True) + scanner = make_test_scanner(wav_dir=wav_dir) + + # Set active bank filter to OPERATIONS only + from frequency_manager import FrequencyConfiguration, FrequencyManager + fm = FrequencyManager(FrequencyConfiguration(file_name=None, disable_lockout=False, disable_priority=False), 5000) + fm.set_active_banks(["OPERATIONS"]) + await fm.add({'single': 462.5625, 'banks': ['FRS_FAMILY'], 'label': 'FRS Ch 1'}) + await fm.add({'single': 467.7125, 'banks': ['OPERATIONS'], 'label': 'FRS Ch 14 Ops'}) + scanner.frequency_manager = fm + scanner.mismatched_freqs = {} + + # Create two channels: one in FRS_FAMILY (inactive) and one in OPERATIONS (active) + ch_inactive = ChannelFrequency( + rf=462.5625, bb=10000, active=False, hanging=False, locked=False, + label="FRS Ch 1" + ) + ch_active = ChannelFrequency( + rf=467.7125, bb=20000, active=False, hanging=False, locked=False, + label="FRS Ch 14 Ops" + ) + + # Mock receiver & demodulators (1 free demodulator) + from unittest.mock import AsyncMock + demod = MagicMock() + demod.center_freq = 0 + demod.set_center_freq = AsyncMock() + scanner.receiver = MagicMock() + scanner.receiver.demodulators = [demod] + scanner.receiver.get_demod_freqs = MagicMock(return_value=[]) + scanner.center_freq = 460000000 + scanner._demod_signal_stats = {0: (0.0, 0)} + + await scanner._assign_channels_to_demodulators([ch_inactive, ch_active]) + + # The active channel (467.7125 MHz, bb=20000) should have been assigned, while inactive (bb=10000) was skipped + demod.set_center_freq.assert_called_once() + assert demod.set_center_freq.call_args[0][0] == 20000 + + +@pytest.mark.asyncio +async def test_assign_channels_promiscuous_skips_bank_scan(tmp_path: Path) -> None: + """Without --banks (promiscuous), _assign_channels_to_demodulators must not + call resolve_banks() per channel per cycle — is_bank_active() is always + active, so the scan is pure waste. Channels must still be assigned.""" + wav_dir = str(tmp_path / "wav") + os.makedirs(wav_dir, exist_ok=True) + scanner = make_test_scanner(wav_dir=wav_dir) + + from frequency_manager import FrequencyConfiguration, FrequencyManager + fm = FrequencyManager(FrequencyConfiguration(file_name=None, disable_lockout=False, disable_priority=False), 5000) + await fm.add({'single': 462.5625, 'banks': ['FRS_FAMILY'], 'label': 'FRS Ch 1'}) + scanner.frequency_manager = fm + scanner.mismatched_freqs = {} + + resolve_banks_calls = {"n": 0} + original_resolve_banks = fm.resolve_banks + + def _spy_resolve_banks(rf: float, ctcss_hz: float | None = None) -> list[str]: + resolve_banks_calls["n"] += 1 + return original_resolve_banks(rf, ctcss_hz) + + fm.resolve_banks = _spy_resolve_banks # type: ignore[method-assign] + + ch = ChannelFrequency( + rf=462.5625, bb=10000, active=False, hanging=False, locked=False, + label="FRS Ch 1" + ) + + from unittest.mock import AsyncMock + demod = MagicMock() + demod.center_freq = 0 + demod.set_center_freq = AsyncMock() + scanner.receiver = MagicMock() + scanner.receiver.demodulators = [demod] + scanner.receiver.get_demod_freqs = MagicMock(return_value=[]) + scanner.center_freq = 460000000 + scanner._demod_signal_stats = {0: (0.0, 0)} + + await scanner._assign_channels_to_demodulators([ch]) + + assert resolve_banks_calls["n"] == 0 + demod.set_center_freq.assert_called_once() + assert demod.set_center_freq.call_args[0][0] == 10000 + + +def test_process_completed_transmission_discards_inactive_banks(tmp_path: Path) -> None: + """_process_completed_transmission must discard WAV files whose final resolved banks are not active.""" + wav_dir = str(tmp_path / "wav") + os.makedirs(wav_dir, exist_ok=True) + tmp_wav = str(tmp_path / "tmp_inactive_bank.wav") + _write_wav(tmp_wav, num_samples=16_000) + + scanner = make_test_scanner(wav_dir=wav_dir) + from frequency_manager import FrequencyConfiguration, FrequencyManager + fm = FrequencyManager(FrequencyConfiguration(file_name=None, disable_lockout=False, disable_priority=False), 5000) + fm.set_active_banks(["SECURITY"]) + scanner.frequency_manager = fm + + msg = ChannelMessage( + state='off', rf=467.7125, bb=0, channel=0, + wav_tmp_path=tmp_wav, banks=["FRS_FAMILY"], + ) + + _msg, record = scanner._process_completed_transmission(msg) + + assert record is None + assert not os.path.exists(tmp_wav), "Discarded WAV must be deleted" + assert _msg.detail == "Discarded inactive bank selection" + + +@pytest.mark.asyncio +async def test_add_metadata_populates_banks_when_filtering(tmp_path: Path) -> None: + """_add_metadata must populate ChannelFrequency.banks when active_banks is + set, and leave it [] in promiscuous mode -- mirroring the gated resolve in + _assign_channels_to_demodulators so non-bank users pay no per-cycle cost.""" + wav_dir = str(tmp_path / "wav") + os.makedirs(wav_dir, exist_ok=True) + scanner = make_test_scanner(wav_dir=wav_dir) + + from frequency_manager import FrequencyConfiguration, FrequencyManager + fm = FrequencyManager(FrequencyConfiguration(file_name=None, disable_lockout=False, disable_priority=False), 5000) + await fm.add({'single': 462.5625, 'banks': ['FRS_FAMILY'], 'label': 'FRS Ch 1'}) + scanner.frequency_manager = fm + + # Receiver stub: no live demodulators, center frequency 460 MHz. The test + # baseband 2,562,500 Hz resolves to 462.5625 MHz (the FRS_FAMILY entry). + scanner.receiver = MagicMock() + scanner.receiver.get_demod_freq_map = MagicMock(return_value={}) + scanner.receiver.center_freq = 460000000 + bb = 2562500 + + # Promiscuous mode (no --banks): no bank tags resolved. + fm.set_active_banks(None) + sweep = scanner._add_metadata(np.array([bb])) + assert sweep[0].banks == [] + + # Bank filtering active: tags resolved onto the channel. + fm.set_active_banks(["FRS_FAMILY"]) + sweep = scanner._add_metadata(np.array([bb])) + assert sweep[0].banks == ["FRS_FAMILY"] + + # Resolved tags reflect the entry's configured banks even when the active + # selection differs; discarding by active_banks happens downstream in + # _process_current_demodulators via is_bank_active(). + fm.set_active_banks(["OPERATIONS"]) + sweep = scanner._add_metadata(np.array([bb])) + assert sweep[0].banks == ["FRS_FAMILY"] + + # A hit outside every configured entry resolves to the UNTAGGED sentinel + # when bank filtering is active (fail-closed), mirroring tier 5. + sweep = scanner._add_metadata(np.array([0])) + assert sweep[0].banks == ["UNTAGGED"] + + +@pytest.mark.asyncio +async def test_set_active_banks_takes_effect_next_assignment(tmp_path: Path) -> None: + """set_active_banks() must flip which channels _assign_channels_to_demodulators + assigns on the next cycle (immediate-apply semantics).""" + wav_dir = str(tmp_path / "wav") + os.makedirs(wav_dir, exist_ok=True) + scanner = make_test_scanner(wav_dir=wav_dir) + + from frequency_manager import FrequencyConfiguration, FrequencyManager + fm = FrequencyManager(FrequencyConfiguration(file_name=None, disable_lockout=False, disable_priority=False), 5000) + await fm.add({'single': 462.5625, 'banks': ['FRS_FAMILY'], 'label': 'FRS Ch 1'}) + await fm.add({'single': 467.7125, 'banks': ['OPERATIONS'], 'label': 'FRS Ch 14 Ops'}) + scanner.frequency_manager = fm + scanner.mismatched_freqs = {} + + from unittest.mock import AsyncMock + demod = MagicMock() + demod.center_freq = 0 + demod.set_center_freq = AsyncMock() + scanner.receiver = MagicMock() + scanner.receiver.demodulators = [demod] + scanner.receiver.get_demod_freqs = MagicMock(return_value=[]) + scanner.center_freq = 460000000 + scanner._demod_signal_stats = {0: (0.0, 0)} + + ch_family = ChannelFrequency( + rf=462.5625, bb=10000, active=False, hanging=False, locked=False, + label="FRS Ch 1" + ) + ch_ops = ChannelFrequency( + rf=467.7125, bb=20000, active=False, hanging=False, locked=False, + label="FRS Ch 14 Ops" + ) + + # Cycle 1: OPERATIONS active -> the OPERATIONS channel is assigned. + scanner.set_active_banks(["OPERATIONS"]) + await scanner._assign_channels_to_demodulators([ch_family, ch_ops]) + assert demod.set_center_freq.call_args[0][0] == 20000 + + # Reset the demodulator to a free slot for the next cycle. + demod.set_center_freq.reset_mock() + demod.center_freq = 0 + + # Cycle 2: switch at runtime to FRS_FAMILY -> the FRS_FAMILY channel is assigned. + scanner.set_active_banks(["FRS_FAMILY"]) + await scanner._assign_channels_to_demodulators([ch_family, ch_ops]) + demod.set_center_freq.assert_called_once() + assert demod.set_center_freq.call_args[0][0] == 10000 + diff --git a/apps/tests/test_scanner_dispatch.py b/apps/tests/test_scanner_dispatch.py index bf6ec9d..5e92e45 100644 --- a/apps/tests/test_scanner_dispatch.py +++ b/apps/tests/test_scanner_dispatch.py @@ -7,19 +7,14 @@ from unittest.mock import AsyncMock, MagicMock import pytest - - from components.base import ChannelInfo, ComponentResult, WavGatekeeper from components.manager import ComponentManager - from config import MasterHam2MonConfig from frequency_manager import ( FrequencyConfiguration, - FrequencyList, FrequencyManager, TransmissionRecord, ) - from scanner import Scanner @@ -78,6 +73,18 @@ def test_scanner_sidecar_json_written_on_kept_wav(tmp_path: Path): scanner._wav_dir = wav_dir scanner.frequency_manager = fm + # Spy on resolve_banks: this path (kept WAV + component, empty msg.banks) + # must resolve banks exactly once, not once per fallback site (scanner.py + # 544/578/621 pre-dedup). + resolve_banks_calls = {"n": 0} + original_resolve_banks = fm.resolve_banks + + def _spy_resolve_banks(rf: float, ctcss_hz: float | None = None) -> list[str]: + resolve_banks_calls["n"] += 1 + return original_resolve_banks(rf, ctcss_hz) + + fm.resolve_banks = _spy_resolve_banks # type: ignore[method-assign] + gk = MockGatekeeper( {"keep": True, "classification": "VOICE", "metadata": {"confidence": 0.98}} ) @@ -89,7 +96,7 @@ def test_scanner_sidecar_json_written_on_kept_wav(tmp_path: Path): msg = ChannelMessage( state="off", - rf=460_125_000.0, + rf=460.125, bb=0, channel=0, wav_tmp_path=tmp_wav, @@ -98,6 +105,8 @@ def test_scanner_sidecar_json_written_on_kept_wav(tmp_path: Path): _msg, record = scanner._process_completed_transmission(msg) + assert resolve_banks_calls["n"] == 1 + assert record is not None assert record.classification == "VOICE" assert record.metadata == {"confidence": 0.98} @@ -113,6 +122,7 @@ def test_scanner_sidecar_json_written_on_kept_wav(tmp_path: Path): assert sidecar_data["rf"] == 460.125 assert sidecar_data["classification"] == "VOICE" assert sidecar_data["metadata"] == {"confidence": 0.98} + assert sidecar_data["banks"] == [] def test_scanner_no_sidecar_json_on_discard(tmp_path: Path): @@ -152,7 +162,7 @@ def test_scanner_no_sidecar_json_on_discard(tmp_path: Path): msg = ChannelMessage( state="off", - rf=460_125_000.0, + rf=460.125, bb=0, channel=0, wav_tmp_path=tmp_wav, @@ -211,7 +221,7 @@ async def test_got_channel_activity_dispatches_notifiers_only_when_interesting( msg = ChannelMessage( state="off", - rf=460_125_000.0, + rf=460.125, bb=0, channel=0, wav_tmp_path=tmp_wav, @@ -235,7 +245,7 @@ async def test_got_channel_activity_dispatches_notifiers_only_when_interesting( msg_unwanted = ChannelMessage( state="off", - rf=460_125_000.0, + rf=460.125, bb=0, channel=0, wav_tmp_path=tmp_mismatch_wav, diff --git a/apps/tests/test_utilities.py b/apps/tests/test_utilities.py index 4673a02..3149206 100644 --- a/apps/tests/test_utilities.py +++ b/apps/tests/test_utilities.py @@ -5,7 +5,10 @@ baseband_to_bin, bin_to_baseband, build_column_edges, + format_active_banks, + format_channel_banks, index_to_column, + parse_bank_entry, wav_bytes_per_sec, wav_duration_sec, ) @@ -85,3 +88,48 @@ def test_index_to_column_matches_bar_column() -> None: for bin_idx in [0, 1, 500, 1024, 2000, 2047]: col: int = index_to_column(bin_idx, edges) assert edges[col] <= bin_idx < edges[col + 1] or col == num_cols - 1 + + +def test_format_active_banks_sorted_comma_separated() -> None: + assert format_active_banks({"NET_B", "NET_A"}) == "NET_A, NET_B" + assert format_active_banks({"NET_A"}) == "NET_A" + + +def test_format_active_banks_empty_is_none() -> None: + assert format_active_banks(set()) == "none" + + +def test_format_active_banks_with_bank_labels() -> None: + labels = {"NET_A": "Net A", "NET_B": "Net B"} + assert format_active_banks({"NET_A"}, labels) == "NET_A (Net A)" + assert format_active_banks({"NET_B", "NET_A"}, labels) == "NET_A (Net A), NET_B (Net B)" + # Unknown tags are left bare (sorted alphabetically) + assert format_active_banks({"NET_A", "MISC"}, labels) == "MISC, NET_A (Net A)" + # Empty set still renders "none" regardless of labels + assert format_active_banks(set(), labels) == "none" + + +def test_format_channel_banks_empty_is_blank() -> None: + assert format_channel_banks([], 20) == "" + + +def test_format_channel_banks_bracketed_join() -> None: + assert format_channel_banks(["NET_A", "NET_B"], 20) == "[NET_A,NET_B]" + + +def test_format_channel_banks_truncated_to_max_len() -> None: + assert format_channel_banks(["NET_A", "NET_B"], 6) == "[NET_A" + assert len(format_channel_banks(["NET_A", "NET_B"], 6)) == 6 + + +def test_parse_bank_entry_comma_and_space_separated() -> None: + assert parse_bank_entry("NET_A, NET_B") == ["NET_A", "NET_B"] + assert parse_bank_entry("NET_A NET_B") == ["NET_A", "NET_B"] + assert parse_bank_entry(" NET_A , NET_B ") == ["NET_A", "NET_B"] + + +def test_parse_bank_entry_empty_is_promiscuous() -> None: + assert parse_bank_entry("") == [] + assert parse_bank_entry(" ") == [] + assert parse_bank_entry("none") == [] + assert parse_bank_entry("NONE") == [] diff --git a/apps/ui_theme.py b/apps/ui_theme.py index 1562fcb..05567f1 100644 --- a/apps/ui_theme.py +++ b/apps/ui_theme.py @@ -131,6 +131,8 @@ def _resolve_color(raw: str | int) -> int: 'channel.icon_inactive': {'fg': 'green', 'dim': True}, 'channel.index_active': {'fg': 'blue'}, 'channel.index_inactive': {'fg': 'blue', 'dim': True}, + 'channel.bank_active': {'fg': 'cyan'}, + 'channel.bank_inactive': {'fg': 'cyan', 'dim': True}, 'channel.placeholder_index': {'fg': 'blue', 'dim': True}, 'channel.placeholder_text': {'fg': 'white', 'dim': True}, diff --git a/apps/utilities.py b/apps/utilities.py index c52a0c1..d632730 100644 --- a/apps/utilities.py +++ b/apps/utilities.py @@ -109,4 +109,50 @@ def index_to_column(index: int, col_edges: list[int]) -> int: """Given an index into the original data and the edges from build_column_edges, return which column it falls in.""" col = bisect.bisect_right(col_edges, index) - 1 - return max(0, min(len(col_edges) - 2, col)) \ No newline at end of file + return max(0, min(len(col_edges) - 2, col)) + + +def format_active_banks(banks: set[str], bank_labels: dict[str, str] | None = None) -> str: + """Format the active bank set for the RECEIVER panel "Banks" row. + + Returns sorted, comma-separated tags (e.g. "NET_A, NET_B"), or "none" when + the set is empty (promiscuous scan-all mode with no --banks). When + ``bank_labels`` is provided, a matching display label is appended to each + tag in parentheses (e.g. "NET_A (Net A)"); unknown tags are left + bare. + """ + if not banks: + return "none" + if bank_labels: + return ", ".join( + f"{tag} ({bank_labels[tag]})" if tag in bank_labels else tag + for tag in sorted(banks)) + return ", ".join(sorted(banks)) + + +def format_channel_banks(banks: list[str], max_len: int) -> str: + """Format a channel's resolved bank tags for the CHANNELS panel. + + Returns a "[NET_A,NET_B]" style tag block, truncated to max_len characters + when it does not fit the reserved label region, or "" when there are no + banks (nothing is drawn, so non-bank users see no layout shift). + """ + if not banks: + return "" + text = f"[{','.join(banks)}]" + if len(text) > max_len: + return text[:max_len] + return text + + +def parse_bank_entry(text: str) -> list[str]: + """Parse a bank-entry-mode input string into a list of bank tags. + + Tags are separated by commas and/or whitespace, stripped, and empty + entries dropped. A literal "none" (case-insensitive) selects promiscuous + scan-all mode and returns []; an empty/blank input also returns []. + """ + stripped = text.strip() + if not stripped or stripped.lower() == "none": + return [] + return [tag for tag in stripped.replace(",", " ").split() if tag] diff --git a/doc/example.freqs.yaml b/doc/example.freqs.yaml index ef3efa8..b20f2c3 100644 --- a/doc/example.freqs.yaml +++ b/doc/example.freqs.yaml @@ -4,15 +4,28 @@ # 1. Priority frequencies (priority: , lower number = higher priority) # 2. Lockout frequencies (locked: true) # 3. Frequency labeling (label: "...") -# 4. CTCSS tone squelch (ctcss: ) +# 4. CTCSS tone squelch (tones: [...]) # # Each entry is either a single frequency (single: ) or a frequency # range (lo: , hi: ), but never both. All frequencies are in MHz. # # CTCSS: -# If a "ctcss" tone (in Hz) is given, the channel uses Tone Squelch mode: -# the demodulator only opens when the carrier includes that sub-audible tone. -# Channels without "ctcss" operate in Carrier Squelch (CSQ) mode. +# Tones are configured with the unified "tones:" key as a list of tone +# values. Each entry is either a bare tone frequency in Hz or a tone rule +# dict with a tone frequency (ctcss: ), an optional per-tone label, and +# optional per-tone banks: +# +# tones: +# - ctcss: 67.0 # in Hz +# label: "Tone A" +# banks: ["FIRE"] +# - 71.9 # bare float, in Hz +# +# A single tone can be written as "tones: [100.0]". +# +# A channel with tones uses Tone Squelch mode: the demodulator only opens +# when the carrier includes one of the configured sub-audible tones. +# Channels without any tone operate in Carrier Squelch (CSQ) mode. # See the "CTCSS Squelch and Tone Filtering" section of the README for the # full list of valid tone frequencies. @@ -28,11 +41,19 @@ frequencies: - label: "Some dispatch" single: 462.400 priority: 1 - ctcss: 100.0 + tones: [100.0] - label: "General talkaround" single: 462.610 - ctcss: 167.9 + tones: [167.9] + + - label: "Multi-tone repeater" + single: 462.550 + tones: + - ctcss: 100.0 + label: "Primary" + banks: ["FIRE"] + - 127.3 # The scanner will detect and record any active channel in these ranges.