Mirror/intercept
1
0
Fork 0
mirror of https://github.com/smittix/intercept.git synced 2026-07-05 16:18:12 -07:00
Code Issues Packages Projects Releases Wiki Activity

Merge upstream/main and resolve weather-satellite.js conflict

Browse Source 
Keep allPasses assignment for satellite filtering support.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
mitchross
2026-02-26 18:37:09 -05:00
parent f652011366 32ce758172
commit bbab991c1f
27 changed files with 5771 additions and 1619 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/dependencies.py
+32 -11
View File
@@ -1,10 +1,11 @@
from __future__ import annotations
import logging
import os
import shutil
import subprocess
from typing import Any
import logging
import os
import platform
import shutil
import subprocess
from typing import Any
logger = logging.getLogger('intercept.dependencies')
@@ -17,12 +18,32 @@ def check_tool(name: str) -> bool:
return get_tool_path(name) is not None
def get_tool_path(name: str) -> str | None:
"""Get the full path to a tool, checking standard PATH and extra locations."""
# First check standard PATH
path = shutil.which(name)
if path:
return path
def get_tool_path(name: str) -> str | None:
"""Get the full path to a tool, checking standard PATH and extra locations."""
# Optional explicit override, e.g. INTERCEPT_RTL_FM_PATH=/opt/homebrew/bin/rtl_fm
env_key = f"INTERCEPT_{name.upper().replace('-', '_')}_PATH"
env_path = os.environ.get(env_key)
if env_path and os.path.isfile(env_path) and os.access(env_path, os.X_OK):
return env_path
# Prefer native Homebrew binaries on Apple Silicon to avoid mixing Rosetta
# /usr/local tools with arm64 Python/runtime.
if platform.system() == 'Darwin':
machine = platform.machine().lower()
preferred_paths: list[str] = []
if machine in {'arm64', 'aarch64'}:
preferred_paths.append('/opt/homebrew/bin')
preferred_paths.append('/usr/local/bin')
for base in preferred_paths:
full_path = os.path.join(base, name)
if os.path.isfile(full_path) and os.access(full_path, os.X_OK):
return full_path
# First check standard PATH
path = shutil.which(name)
if path:
return path
# Check additional paths (e.g., /usr/sbin for aircrack-ng on Debian)
for extra_path in EXTRA_TOOL_PATHS:
utils/dsc/constants.py
+9
View File
@@ -89,6 +89,15 @@ TELECOMMAND_CODES = {
201: 'POLL_RESPONSE', # Poll response
}
# Full 0-127 telecommand lookup (maps unknown codes to "UNKNOWN")
TELECOMMAND_CODES_FULL = {i: TELECOMMAND_CODES.get(i, "UNKNOWN") for i in range(128)}
# Format codes that carry telecommand fields
TELECOMMAND_FORMATS = {112, 114, 116, 120, 123}
# Minimum symbols (after phasing strip) before an EOS can be accepted
MIN_SYMBOLS_FOR_FORMAT = 12
# =============================================================================
# DSC Symbol Definitions
utils/dsc/decoder.py
+37 -11
View File
@@ -43,6 +43,8 @@ from .constants import (
FORMAT_CODES,
DISTRESS_NATURE_CODES,
VALID_EOS,
TELECOMMAND_FORMATS,
MIN_SYMBOLS_FOR_FORMAT,
)
# Configure logging
@@ -222,13 +224,14 @@ class DSCDecoder:
Detect DSC dot pattern for synchronization.
The dot pattern is at least 200 alternating bits (1010101...).
We look for at least 20 consecutive alternations.
We require at least 100 consecutive alternations to avoid
false sync triggers from noise.
"""
if len(self.bit_buffer) < 40:
if len(self.bit_buffer) < 200:
return False
# Check last 40 bits for alternating pattern
last_bits = self.bit_buffer[-40:]
# Check last 200 bits for alternating pattern
last_bits = self.bit_buffer[-200:]
alternations = 0
for i in range(1, len(last_bits)):
@@ -237,7 +240,7 @@ class DSCDecoder:
else:
alternations = 0
if alternations >= 20:
if alternations >= 100:
return True
return False
@@ -263,27 +266,37 @@ class DSCDecoder:
if end <= len(self.message_bits):
symbol_bits = self.message_bits[start:end]
symbol_value = self._bits_to_symbol(symbol_bits)
if symbol_value == -1:
logger.debug("DSC symbol check bit failure, aborting decode")
return None
symbols.append(symbol_value)
# Strip phasing sequence (RX/DX symbols 120-126) from the
# start of the message. Per ITU-R M.493, after the dot pattern
# there are 7 phasing symbols before the format specifier.
# Bound to max 7 — if more are present, this is a bad sync.
msg_start = 0
for i, sym in enumerate(symbols):
if 120 <= sym <= 126:
msg_start = i + 1
else:
break
if msg_start > 7:
logger.debug("DSC bad sync: >7 phasing symbols stripped")
return None
symbols = symbols[msg_start:]
if len(symbols) < 5:
return None
# Look for EOS (End of Sequence) - symbols 117, 122, or 127
# EOS must appear after at least MIN_SYMBOLS_FOR_FORMAT symbols
eos_found = False
eos_index = -1
for i, sym in enumerate(symbols):
if sym in VALID_EOS:
if i < MIN_SYMBOLS_FOR_FORMAT:
continue # Too early — not a real EOS
eos_found = True
eos_index = i
break
@@ -300,7 +313,9 @@ class DSCDecoder:
Convert 10 bits to symbol value.
DSC uses 10-bit symbols: 7 information bits + 3 error bits.
We extract the 7-bit value.
The 3 check bits provide parity such that the total number of
'1' bits across all 10 bits should be even (even parity).
Returns -1 if the check bits are invalid.
"""
if len(bits) != 10:
return -1
@@ -311,6 +326,11 @@ class DSCDecoder:
if bits[i]:
value |= (1 << i)
# Validate check bits: total number of 1s should be even
ones = sum(bits)
if ones % 2 != 0:
return -1
return value
def _decode_symbols(self, symbols: list[int]) -> dict | None:
@@ -356,9 +376,13 @@ class DSCDecoder:
# Decode MMSI from symbols 1-5 (destination/address)
dest_mmsi = self._decode_mmsi(symbols[1:6])
if dest_mmsi is None:
return None
# Decode self-ID from symbols 6-10 (source)
source_mmsi = self._decode_mmsi(symbols[6:11])
if source_mmsi is None:
return None
message = {
'type': 'dsc',
@@ -387,8 +411,9 @@ class DSCDecoder:
if position:
message['position'] = position
# Telecommand fields (usually last two before EOS)
if len(remaining) >= 2:
# Telecommand fields (last two before EOS) — only for formats
# that carry telecommand fields per ITU-R M.493
if format_code in TELECOMMAND_FORMATS and len(remaining) >= 2:
message['telecommand1'] = remaining[-2]
message['telecommand2'] = remaining[-1]
@@ -402,20 +427,21 @@ class DSCDecoder:
logger.warning(f"DSC decode error: {e}")
return None
def _decode_mmsi(self, symbols: list[int]) -> str:
def _decode_mmsi(self, symbols: list[int]) -> str | None:
"""
Decode MMSI from 5 DSC symbols.
Each symbol represents 2 BCD digits (00-99).
5 symbols = 10 digits, but MMSI is 9 digits (first symbol has leading 0).
Returns None if any symbol is out of valid BCD range.
"""
if len(symbols) < 5:
return '000000000'
return None
digits = []
for sym in symbols:
if sym < 0 or sym > 99:
sym = 0
return None
# Each symbol is 2 BCD digits
digits.append(f'{sym:02d}')
utils/dsc/parser.py
+4 -1
View File
@@ -248,7 +248,10 @@ def parse_dsc_message(raw_line: str) -> dict[str, Any] | None:
msg['priority'] = get_category_priority(msg['category'])
# Mark if this is a critical alert
msg['is_critical'] = msg['category'] in ('DISTRESS', 'ALL_SHIPS_URGENCY_SAFETY')
msg['is_critical'] = msg['category'] in (
'DISTRESS', 'DISTRESS_ACK', 'DISTRESS_RELAY',
'URGENCY', 'SAFETY', 'ALL_SHIPS_URGENCY_SAFETY',
)
return msg
utils/morse.py
+1279 -276
View File
File diff suppressed because it is too large Load Diff
utils/sdr/rtlsdr.py
+33 -20
View File
@@ -14,16 +14,16 @@ from typing import Optional
from .base import CommandBuilder, SDRCapabilities, SDRDevice, SDRType
from utils.dependencies import get_tool_path
logger = logging.getLogger('intercept.sdr.rtlsdr')
def _rtl_fm_demod_mode(modulation: str) -> str:
"""Map app/UI modulation names to rtl_fm demod tokens."""
mod = str(modulation or '').lower().strip()
return 'wbfm' if mod == 'wfm' else mod
def _get_dump1090_bias_t_flag(dump1090_path: str) -> Optional[str]:
logger = logging.getLogger('intercept.sdr.rtlsdr')
def _rtl_fm_demod_mode(modulation: str) -> str:
"""Map app/UI modulation names to rtl_fm demod tokens."""
mod = str(modulation or '').lower().strip()
return 'wbfm' if mod == 'wfm' else mod
def _get_dump1090_bias_t_flag(dump1090_path: str) -> Optional[str]:
"""Detect the correct bias-t flag for the installed dump1090 variant.
Different dump1090 forks use different flags:
@@ -86,22 +86,27 @@ class RTLSDRCommandBuilder(CommandBuilder):
ppm: Optional[int] = None,
modulation: str = "fm",
squelch: Optional[int] = None,
bias_t: bool = False
bias_t: bool = False,
direct_sampling: Optional[int] = None,
) -> list[str]:
"""
Build rtl_fm command for FM demodulation.
Used for pager decoding. Supports local devices and rtl_tcp connections.
Args:
direct_sampling: Enable direct sampling mode (0=off, 1=I-branch,
2=Q-branch). Use 2 for HF reception below 24 MHz.
"""
rtl_fm_path = get_tool_path('rtl_fm') or 'rtl_fm'
demod_mode = _rtl_fm_demod_mode(modulation)
cmd = [
rtl_fm_path,
'-d', self._get_device_arg(device),
'-f', f'{frequency_mhz}M',
'-M', demod_mode,
'-s', str(sample_rate),
]
rtl_fm_path = get_tool_path('rtl_fm') or 'rtl_fm'
demod_mode = _rtl_fm_demod_mode(modulation)
cmd = [
rtl_fm_path,
'-d', self._get_device_arg(device),
'-f', f'{frequency_mhz}M',
'-M', demod_mode,
'-s', str(sample_rate),
]
if gain is not None and gain > 0:
cmd.extend(['-g', str(gain)])
@@ -112,6 +117,14 @@ class RTLSDRCommandBuilder(CommandBuilder):
if squelch is not None and squelch > 0:
cmd.extend(['-l', str(squelch)])
if direct_sampling is not None:
# Older rtl_fm builds (common in Docker/distro packages) don't
# support -D; they use -E direct / -E direct2 instead.
if direct_sampling == 1:
cmd.extend(['-E', 'direct'])
elif direct_sampling == 2:
cmd.extend(['-E', 'direct2'])
if bias_t:
cmd.extend(['-T'])