diff --git a/tests/test_gsm_spy.py b/tests/test_gsm_spy.py
deleted file mode 100644
index 797f794..0000000
--- a/tests/test_gsm_spy.py
+++ /dev/null
@@ -1,360 +0,0 @@
-"""Unit tests for GSM Spy parsing and validation functions."""
-
-import pytest
-from routes.gsm_spy import (
- parse_grgsm_scanner_output,
- parse_tshark_output,
- arfcn_to_frequency,
- validate_band_names,
- REGIONAL_BANDS
-)
-
-
-class TestParseGrgsmScannerOutput:
- """Tests for parse_grgsm_scanner_output()."""
-
- def test_valid_output_line(self):
- """Test parsing a valid grgsm_scanner output line."""
- line = "ARFCN: 23, Freq: 940.6M, CID: 31245, LAC: 1234, MCC: 214, MNC: 01, Pwr: -48"
- result = parse_grgsm_scanner_output(line)
-
- assert result is not None
- assert result['type'] == 'tower'
- assert result['arfcn'] == 23
- assert result['frequency'] == 940.6
- assert result['cid'] == 31245
- assert result['lac'] == 1234
- assert result['mcc'] == 214
- assert result['mnc'] == 1
- assert result['signal_strength'] == -48.0
- assert 'timestamp' in result
-
- def test_freq_without_suffix(self):
- """Test parsing frequency without M suffix."""
- line = "ARFCN: 975, Freq: 925.2, CID: 13522, LAC: 38722, MCC: 262, MNC: 1, Pwr: -58"
- result = parse_grgsm_scanner_output(line)
- assert result is not None
- assert result['frequency'] == 925.2
-
- def test_config_line(self):
- """Test that configuration lines are skipped."""
- line = " Configuration: 1 CCCH, not combined"
- result = parse_grgsm_scanner_output(line)
- assert result is None
-
- def test_neighbour_line(self):
- """Test that neighbour cell lines are skipped."""
- line = " Neighbour Cells: 57, 61, 70, 71, 72, 86"
- result = parse_grgsm_scanner_output(line)
- assert result is None
-
- def test_cell_arfcn_line(self):
- """Test that cell ARFCN lines are skipped."""
- line = " Cell ARFCNs: 63, 76"
- result = parse_grgsm_scanner_output(line)
- assert result is None
-
- def test_progress_line(self):
- """Test that progress/status lines are skipped."""
- line = "Scanning GSM900 band..."
- result = parse_grgsm_scanner_output(line)
- assert result is None
-
- def test_empty_line(self):
- """Test handling of empty lines."""
- result = parse_grgsm_scanner_output("")
- assert result is None
-
- def test_invalid_data(self):
- """Test handling of non-numeric values."""
- line = "ARFCN: abc, Freq: xyz, CID: bad, LAC: data, MCC: bad, MNC: bad, Pwr: bad"
- result = parse_grgsm_scanner_output(line)
- assert result is None
-
- def test_no_identity_filtered(self):
- """Test that MCC=0/MNC=0 entries (no network identity) are filtered out."""
- line = "ARFCN: 115, Freq: 925.0M, CID: 0, LAC: 0, MCC: 0, MNC: 0, Pwr: -100"
- result = parse_grgsm_scanner_output(line)
- assert result is None
-
- def test_mcc_zero_mnc_zero_filtered(self):
- """Test that MCC=0/MNC=0 even with valid CID is filtered out."""
- line = "ARFCN: 113, Freq: 924.6M, CID: 1234, LAC: 5678, MCC: 0, MNC: 0, Pwr: -90"
- result = parse_grgsm_scanner_output(line)
- assert result is None
-
- def test_cid_zero_valid_mcc_passes(self):
- """Test that CID=0 with valid MCC/MNC passes (partially decoded cell)."""
- line = "ARFCN: 115, Freq: 958.0M, CID: 0, LAC: 21864, MCC: 234, MNC: 10, Pwr: -51"
- result = parse_grgsm_scanner_output(line)
- assert result is not None
- assert result['cid'] == 0
- assert result['mcc'] == 234
- assert result['signal_strength'] == -51.0
-
- def test_valid_cid_nonzero(self):
- """Test that valid non-zero CID/MCC entries pass through."""
- line = "ARFCN: 115, Freq: 925.0M, CID: 19088, LAC: 21864, MCC: 234, MNC: 10, Pwr: -58"
- result = parse_grgsm_scanner_output(line)
- assert result is not None
- assert result['cid'] == 19088
- assert result['signal_strength'] == -58.0
-
-
-class TestParseTsharkOutput:
- """Tests for parse_tshark_output()."""
-
- def test_valid_full_output(self):
- """Test parsing tshark output with all fields."""
- line = "5\t0xABCD1234\t123456789012345\t1234\t31245"
- result = parse_tshark_output(line)
-
- assert result is not None
- assert result['type'] == 'device'
- assert result['ta_value'] == 5
- assert result['tmsi'] == '0xABCD1234'
- assert result['imsi'] == '123456789012345'
- assert result['lac'] == 1234
- assert result['cid'] == 31245
- assert result['distance_meters'] == 5 * 554 # TA * 554 meters
- assert 'timestamp' in result
-
- def test_missing_optional_fields(self):
- """Test parsing with missing optional fields (empty tabs).
-
- A packet with TA but no TMSI/IMSI is discarded since there's
- no device identifier to track.
- """
- line = "3\t\t\t1234\t31245"
- result = parse_tshark_output(line)
- assert result is None
-
- def test_missing_optional_fields_with_tmsi(self):
- """Test parsing with TMSI but missing TA, IMSI, CID."""
- line = "\t0xABCD\t\t1234\t"
- result = parse_tshark_output(line)
-
- assert result is not None
- assert result['ta_value'] is None
- assert result['tmsi'] == '0xABCD'
- assert result['imsi'] is None
- assert result['lac'] == 1234
- assert result['cid'] is None
-
- def test_no_ta_value(self):
- """Test parsing without TA value (empty first field)."""
- line = "\t0xABCD1234\t123456789012345\t1234\t31245"
- result = parse_tshark_output(line)
-
- assert result is not None
- assert result['ta_value'] is None
- assert result['tmsi'] == '0xABCD1234'
- assert result['imsi'] == '123456789012345'
- assert result['lac'] == 1234
- assert result['cid'] == 31245
-
- def test_invalid_line(self):
- """Test handling of invalid tshark output."""
- line = "invalid data"
- result = parse_tshark_output(line)
- assert result is None
-
- def test_empty_line(self):
- """Test handling of empty lines."""
- result = parse_tshark_output("")
- assert result is None
-
- def test_partial_fields(self):
- """Test with fewer than 5 fields."""
- line = "5\t0xABCD1234" # Only 2 fields
- result = parse_tshark_output(line)
- assert result is None
-
-
-class TestArfcnToFrequency:
- """Tests for arfcn_to_frequency()."""
-
- def test_gsm850_arfcn(self):
- """Test ARFCN in GSM850 band."""
- # GSM850: ARFCN 128-251, 869-894 MHz
- arfcn = 128
- freq = arfcn_to_frequency(arfcn)
- assert freq == 869000000 # 869 MHz
-
- arfcn = 251
- freq = arfcn_to_frequency(arfcn)
- assert freq == 893600000 # 893.6 MHz
-
- def test_egsm900_arfcn(self):
- """Test ARFCN in EGSM900 band."""
- # EGSM900: ARFCN 0-124, DL = 935 + 0.2*ARFCN MHz
- arfcn = 0
- freq = arfcn_to_frequency(arfcn)
- assert freq == 935000000 # 935.0 MHz
-
- arfcn = 22
- freq = arfcn_to_frequency(arfcn)
- assert freq == 939400000 # 939.4 MHz
-
- arfcn = 124
- freq = arfcn_to_frequency(arfcn)
- assert freq == 959800000 # 959.8 MHz
-
- def test_egsm900_ext_arfcn(self):
- """Test ARFCN in EGSM900 extension band."""
- # EGSM900_EXT: ARFCN 975-1023, DL = 925.2 + 0.2*(ARFCN-975) MHz
- arfcn = 975
- freq = arfcn_to_frequency(arfcn)
- assert freq == 925200000 # 925.2 MHz
-
- arfcn = 1023
- freq = arfcn_to_frequency(arfcn)
- assert freq == 934800000 # 934.8 MHz
-
- def test_dcs1800_arfcn(self):
- """Test ARFCN in DCS1800 band."""
- # DCS1800: ARFCN 512-885, 1805-1880 MHz
- # Note: ARFCN 512 also exists in PCS1900 and will match that first
- # Use ARFCN 811+ which is only in DCS1800
- arfcn = 811 # Beyond PCS1900 range (512-810)
- freq = arfcn_to_frequency(arfcn)
- # 811 is ARFCN offset from 512, so freq = 1805MHz + (811-512)*200kHz
- expected = 1805000000 + (811 - 512) * 200000
- assert freq == expected
-
- arfcn = 885
- freq = arfcn_to_frequency(arfcn)
- assert freq == 1879600000 # 1879.6 MHz
-
- def test_pcs1900_arfcn(self):
- """Test ARFCN in PCS1900 band."""
- # PCS1900: ARFCN 512-810, 1930-1990 MHz
- # Note: overlaps with DCS1800 ARFCN range, but different frequencies
- arfcn = 512
- freq = arfcn_to_frequency(arfcn)
- # Will match first band (DCS1800 in Europe config)
- assert freq > 0
-
- def test_invalid_arfcn(self):
- """Test ARFCN outside known ranges."""
- with pytest.raises(ValueError, match="not found in any known GSM band"):
- arfcn_to_frequency(9999)
-
- with pytest.raises(ValueError):
- arfcn_to_frequency(-1)
-
- def test_arfcn_200khz_spacing(self):
- """Test that ARFCNs are 200kHz apart."""
- arfcn1 = 128
- arfcn2 = 129
- freq1 = arfcn_to_frequency(arfcn1)
- freq2 = arfcn_to_frequency(arfcn2)
- assert freq2 - freq1 == 200000 # 200 kHz
-
-
-class TestValidateBandNames:
- """Tests for validate_band_names()."""
-
- def test_valid_americas_bands(self):
- """Test valid band names for Americas region."""
- bands = ['GSM850', 'PCS1900']
- result, error = validate_band_names(bands, 'Americas')
- assert result == bands
- assert error is None
-
- def test_valid_europe_bands(self):
- """Test valid band names for Europe region."""
- # Note: Europe uses EGSM900, not GSM900
- bands = ['EGSM900', 'DCS1800', 'GSM850', 'GSM800']
- result, error = validate_band_names(bands, 'Europe')
- assert result == bands
- assert error is None
-
- def test_valid_asia_bands(self):
- """Test valid band names for Asia region."""
- # Note: Asia uses EGSM900, not GSM900
- bands = ['EGSM900', 'DCS1800']
- result, error = validate_band_names(bands, 'Asia')
- assert result == bands
- assert error is None
-
- def test_invalid_band_for_region(self):
- """Test invalid band name for a region."""
- bands = ['GSM900', 'INVALID_BAND']
- result, error = validate_band_names(bands, 'Americas')
- assert result == []
- assert error is not None
- assert 'Invalid bands' in error
- assert 'INVALID_BAND' in error
-
- def test_invalid_region(self):
- """Test invalid region name."""
- bands = ['GSM900']
- result, error = validate_band_names(bands, 'InvalidRegion')
- assert result == []
- assert error is not None
- assert 'Invalid region' in error
-
- def test_empty_bands_list(self):
- """Test with empty bands list."""
- result, error = validate_band_names([], 'Americas')
- assert result == []
- assert error is None
-
- def test_single_valid_band(self):
- """Test with single valid band."""
- bands = ['GSM850']
- result, error = validate_band_names(bands, 'Americas')
- assert result == ['GSM850']
- assert error is None
-
- def test_case_sensitive_band_names(self):
- """Test that band names are case-sensitive."""
- bands = ['gsm850'] # lowercase
- result, error = validate_band_names(bands, 'Americas')
- assert result == []
- assert error is not None
-
- def test_multiple_invalid_bands(self):
- """Test with multiple invalid bands."""
- bands = ['INVALID1', 'GSM850', 'INVALID2']
- result, error = validate_band_names(bands, 'Americas')
- assert result == []
- assert error is not None
- assert 'INVALID1' in error
- assert 'INVALID2' in error
-
-
-class TestRegionalBandsConfig:
- """Tests for REGIONAL_BANDS configuration."""
-
- def test_all_regions_defined(self):
- """Test that all expected regions are defined."""
- assert 'Americas' in REGIONAL_BANDS
- assert 'Europe' in REGIONAL_BANDS
- assert 'Asia' in REGIONAL_BANDS
-
- def test_all_bands_have_required_fields(self):
- """Test that all bands have required configuration fields."""
- for region, bands in REGIONAL_BANDS.items():
- for band_name, band_config in bands.items():
- assert 'start' in band_config
- assert 'end' in band_config
- assert 'arfcn_start' in band_config
- assert 'arfcn_end' in band_config
-
- def test_frequency_ranges_valid(self):
- """Test that frequency ranges are positive and start < end."""
- for region, bands in REGIONAL_BANDS.items():
- for band_name, band_config in bands.items():
- assert band_config['start'] > 0
- assert band_config['end'] > 0
- assert band_config['start'] < band_config['end']
-
- def test_arfcn_ranges_valid(self):
- """Test that ARFCN ranges are valid."""
- for region, bands in REGIONAL_BANDS.items():
- for band_name, band_config in bands.items():
- assert band_config['arfcn_start'] >= 0
- assert band_config['arfcn_end'] >= 0
- assert band_config['arfcn_start'] <= band_config['arfcn_end']
diff --git a/utils/constants.py b/utils/constants.py
index f51124e..2b5edff 100644
--- a/utils/constants.py
+++ b/utils/constants.py
@@ -275,13 +275,3 @@ MAX_DEAUTH_ALERTS_AGE_SECONDS = 300 # 5 minutes
# Deauth detector sniff timeout (seconds)
DEAUTH_SNIFF_TIMEOUT = 0.5
-
-# =============================================================================
-# GSM SPY (Cellular Intelligence)
-# =============================================================================
-
-# Maximum age for GSM tower/device data in DataStore (seconds)
-MAX_GSM_AGE_SECONDS = 300 # 5 minutes
-
-# Timing Advance conversion to meters
-GSM_TA_METERS_PER_UNIT = 554
diff --git a/utils/database.py b/utils/database.py
index 9e62d87..3211624 100644
--- a/utils/database.py
+++ b/utils/database.py
@@ -453,134 +453,6 @@ def init_db() -> None:
ON tscm_cases(status, created_at)
''')
- # =====================================================================
- # GSM (Global System for Mobile) Intelligence Tables
- # =====================================================================
-
- # gsm_cells - Known cell towers (OpenCellID cache)
- conn.execute('''
- CREATE TABLE IF NOT EXISTS gsm_cells (
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- mcc INTEGER NOT NULL,
- mnc INTEGER NOT NULL,
- lac INTEGER NOT NULL,
- cid INTEGER NOT NULL,
- lat REAL,
- lon REAL,
- azimuth INTEGER,
- range_meters INTEGER,
- samples INTEGER,
- radio TEXT,
- operator TEXT,
- first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
- last_verified TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
- metadata TEXT,
- UNIQUE(mcc, mnc, lac, cid)
- )
- ''')
-
- # gsm_rogues - Detected rogue towers / IMSI catchers
- conn.execute('''
- CREATE TABLE IF NOT EXISTS gsm_rogues (
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- arfcn INTEGER NOT NULL,
- mcc INTEGER,
- mnc INTEGER,
- lac INTEGER,
- cid INTEGER,
- signal_strength REAL,
- reason TEXT NOT NULL,
- threat_level TEXT DEFAULT 'medium',
- detected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
- location_lat REAL,
- location_lon REAL,
- acknowledged BOOLEAN DEFAULT 0,
- notes TEXT,
- metadata TEXT
- )
- ''')
-
- # gsm_signals - 60-day archive of signal observations
- conn.execute('''
- CREATE TABLE IF NOT EXISTS gsm_signals (
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- imsi TEXT,
- tmsi TEXT,
- mcc INTEGER,
- mnc INTEGER,
- lac INTEGER,
- cid INTEGER,
- ta_value INTEGER,
- signal_strength REAL,
- arfcn INTEGER,
- timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
- metadata TEXT
- )
- ''')
-
- # gsm_tmsi_log - 24-hour raw pings for crowd density
- conn.execute('''
- CREATE TABLE IF NOT EXISTS gsm_tmsi_log (
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- tmsi TEXT NOT NULL,
- lac INTEGER,
- cid INTEGER,
- ta_value INTEGER,
- timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
- )
- ''')
-
- # gsm_velocity_log - 1-hour buffer for movement tracking
- conn.execute('''
- CREATE TABLE IF NOT EXISTS gsm_velocity_log (
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- device_id TEXT NOT NULL,
- prev_ta INTEGER,
- curr_ta INTEGER,
- prev_cid INTEGER,
- curr_cid INTEGER,
- timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
- estimated_velocity REAL,
- metadata TEXT
- )
- ''')
-
- # GSM indexes for performance
- conn.execute('''
- CREATE INDEX IF NOT EXISTS idx_gsm_cells_location
- ON gsm_cells(lat, lon)
- ''')
-
- conn.execute('''
- CREATE INDEX IF NOT EXISTS idx_gsm_cells_identity
- ON gsm_cells(mcc, mnc, lac, cid)
- ''')
-
- conn.execute('''
- CREATE INDEX IF NOT EXISTS idx_gsm_rogues_severity
- ON gsm_rogues(threat_level, detected_at)
- ''')
-
- conn.execute('''
- CREATE INDEX IF NOT EXISTS idx_gsm_signals_cell_time
- ON gsm_signals(cid, lac, timestamp)
- ''')
-
- conn.execute('''
- CREATE INDEX IF NOT EXISTS idx_gsm_signals_device
- ON gsm_signals(imsi, tmsi, timestamp)
- ''')
-
- conn.execute('''
- CREATE INDEX IF NOT EXISTS idx_gsm_tmsi_log_time
- ON gsm_tmsi_log(timestamp)
- ''')
-
- conn.execute('''
- CREATE INDEX IF NOT EXISTS idx_gsm_velocity_log_device
- ON gsm_velocity_log(device_id, timestamp)
- ''')
-
# =====================================================================
# DSC (Digital Selective Calling) Tables
# =====================================================================
@@ -2298,60 +2170,3 @@ def cleanup_old_payloads(max_age_hours: int = 24) -> int:
''', (f'-{max_age_hours} hours',))
return cursor.rowcount
-
-# =============================================================================
-# GSM Cleanup Functions
-# =============================================================================
-
-def cleanup_old_gsm_signals(max_age_days: int = 60) -> int:
- """
- Remove old GSM signal observations (60-day archive).
-
- Args:
- max_age_days: Maximum age in days (default: 60)
-
- Returns:
- Number of deleted entries
- """
- with get_db() as conn:
- cursor = conn.execute('''
- DELETE FROM gsm_signals
- WHERE timestamp < datetime('now', ?)
- ''', (f'-{max_age_days} days',))
- return cursor.rowcount
-
-
-def cleanup_old_gsm_tmsi_log(max_age_hours: int = 24) -> int:
- """
- Remove old TMSI log entries (24-hour buffer for crowd density).
-
- Args:
- max_age_hours: Maximum age in hours (default: 24)
-
- Returns:
- Number of deleted entries
- """
- with get_db() as conn:
- cursor = conn.execute('''
- DELETE FROM gsm_tmsi_log
- WHERE timestamp < datetime('now', ?)
- ''', (f'-{max_age_hours} hours',))
- return cursor.rowcount
-
-
-def cleanup_old_gsm_velocity_log(max_age_hours: int = 1) -> int:
- """
- Remove old velocity log entries (1-hour buffer for movement tracking).
-
- Args:
- max_age_hours: Maximum age in hours (default: 1)
-
- Returns:
- Number of deleted entries
- """
- with get_db() as conn:
- cursor = conn.execute('''
- DELETE FROM gsm_velocity_log
- WHERE timestamp < datetime('now', ?)
- ''', (f'-{max_age_hours} hours',))
- return cursor.rowcount
diff --git a/utils/dependencies.py b/utils/dependencies.py
index e6a1bee..a12eca7 100644
--- a/utils/dependencies.py
+++ b/utils/dependencies.py
@@ -444,38 +444,6 @@ TOOL_DEPENDENCIES = {
}
}
},
- 'gsm': {
- 'name': 'GSM Intelligence',
- 'tools': {
- 'grgsm_scanner': {
- 'required': True,
- 'description': 'gr-gsm scanner for finding GSM towers',
- 'install': {
- 'apt': 'Build gr-gsm from source: https://github.com/bkerler/gr-gsm',
- 'brew': 'brew install gr-gsm (may require manual build)',
- 'manual': 'https://github.com/bkerler/gr-gsm'
- }
- },
- 'grgsm_livemon': {
- 'required': True,
- 'description': 'gr-gsm live monitor for decoding GSM signals',
- 'install': {
- 'apt': 'Included with gr-gsm package',
- 'brew': 'Included with gr-gsm',
- 'manual': 'Included with gr-gsm'
- }
- },
- 'tshark': {
- 'required': True,
- 'description': 'Wireshark CLI for parsing GSM packets',
- 'install': {
- 'apt': 'sudo apt-get install tshark',
- 'brew': 'brew install wireshark',
- 'manual': 'https://www.wireshark.org/download.html'
- }
- }
- }
- }
}
diff --git a/utils/gsm_geocoding.py b/utils/gsm_geocoding.py
deleted file mode 100644
index feaf164..0000000
--- a/utils/gsm_geocoding.py
+++ /dev/null
@@ -1,226 +0,0 @@
-"""GSM Cell Tower Geocoding Service.
-
-Provides hybrid cache-first geocoding with async API fallback for cell towers.
-"""
-
-from __future__ import annotations
-
-import logging
-import queue
-from typing import Any
-
-import requests
-
-import config
-from utils.database import get_db
-
-logger = logging.getLogger('intercept.gsm_geocoding')
-
-# Queue for pending geocoding requests
-_geocoding_queue = queue.Queue(maxsize=100)
-
-
-def lookup_cell_coordinates(mcc: int, mnc: int, lac: int, cid: int) -> dict[str, Any] | None:
- """
- Lookup cell tower coordinates with cache-first strategy.
-
- Strategy:
- 1. Check gsm_cells table (cache) - fast synchronous lookup
- 2. If not found, return None (caller decides whether to use API)
-
- Args:
- mcc: Mobile Country Code
- mnc: Mobile Network Code
- lac: Location Area Code
- cid: Cell ID
-
- Returns:
- dict with keys: lat, lon, source='cache', azimuth (optional),
- range_meters (optional), operator (optional), radio (optional)
- Returns None if not found in cache.
- """
- try:
- with get_db() as conn:
- result = conn.execute('''
- SELECT lat, lon, azimuth, range_meters, operator, radio
- FROM gsm_cells
- WHERE mcc = ? AND mnc = ? AND lac = ? AND cid = ?
- ''', (mcc, mnc, lac, cid)).fetchone()
-
- if result and result['lat'] is not None and result['lon'] is not None:
- return {
- 'lat': result['lat'],
- 'lon': result['lon'],
- 'source': 'cache',
- 'azimuth': result['azimuth'],
- 'range_meters': result['range_meters'],
- 'operator': result['operator'],
- 'radio': result['radio']
- }
-
- return None
-
- except Exception as e:
- logger.error(f"Error looking up coordinates from cache: {e}")
- return None
-
-
-def _get_api_key() -> str:
- """Get OpenCellID API key at runtime (env var first, then database)."""
- env_key = config.GSM_OPENCELLID_API_KEY
- if env_key:
- return env_key
- from utils.database import get_setting
- return get_setting('gsm.opencellid.api_key', '')
-
-
-def lookup_cell_from_api(mcc: int, mnc: int, lac: int, cid: int) -> dict[str, Any] | None:
- """
- Lookup cell tower from OpenCellID API and cache result.
-
- Args:
- mcc: Mobile Country Code
- mnc: Mobile Network Code
- lac: Location Area Code
- cid: Cell ID
-
- Returns:
- dict with keys: lat, lon, source='api', azimuth (optional),
- range_meters (optional), operator (optional), radio (optional)
- Returns None if API call fails or cell not found.
- """
- try:
- api_key = _get_api_key()
- if not api_key:
- logger.warning("OpenCellID API key not configured")
- return None
-
- api_url = config.GSM_OPENCELLID_API_URL
- params = {
- 'key': api_key,
- 'mcc': mcc,
- 'mnc': mnc,
- 'lac': lac,
- 'cellid': cid,
- 'format': 'json'
- }
-
- response = requests.get(api_url, params=params, timeout=10)
-
- if response.status_code == 200:
- cell_data = response.json()
- lat = cell_data.get('lat')
- lon = cell_data.get('lon')
-
- # Validate response has actual coordinates
- if lat is None or lon is None:
- logger.warning(
- f"OpenCellID API returned 200 but no coordinates for "
- f"MCC={mcc} MNC={mnc} LAC={lac} CID={cid}: {cell_data}"
- )
- return None
-
- # Cache the result
- with get_db() as conn:
- conn.execute('''
- INSERT OR REPLACE INTO gsm_cells
- (mcc, mnc, lac, cid, lat, lon, azimuth, range_meters, samples, radio, operator, last_verified)
- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
- ''', (
- mcc, mnc, lac, cid,
- lat, lon,
- cell_data.get('azimuth'),
- cell_data.get('range'),
- cell_data.get('samples'),
- cell_data.get('radio'),
- cell_data.get('operator')
- ))
- conn.commit()
-
- logger.info(f"Cached cell tower from API: MCC={mcc} MNC={mnc} LAC={lac} CID={cid} -> ({lat}, {lon})")
-
- return {
- 'lat': lat,
- 'lon': lon,
- 'source': 'api',
- 'azimuth': cell_data.get('azimuth'),
- 'range_meters': cell_data.get('range'),
- 'operator': cell_data.get('operator'),
- 'radio': cell_data.get('radio')
- }
- else:
- logger.warning(
- f"OpenCellID API returned {response.status_code} for "
- f"MCC={mcc} MNC={mnc} LAC={lac} CID={cid}: {response.text[:200]}"
- )
- return None
-
- except Exception as e:
- logger.error(f"Error calling OpenCellID API: {e}")
- return None
-
-
-def enrich_tower_data(tower_data: dict[str, Any]) -> dict[str, Any]:
- """
- Enrich tower data with coordinates using cache-first strategy.
-
- If coordinates found in cache, adds them immediately.
- If not found, marks as 'pending' and queues for background API lookup.
-
- Args:
- tower_data: Dictionary with keys mcc, mnc, lac, cid (and other tower data)
-
- Returns:
- Enriched tower_data dict with added fields:
- - lat, lon (if found in cache)
- - status='pending' (if needs API lookup)
- - source='cache' (if from cache)
- """
- mcc = tower_data.get('mcc')
- mnc = tower_data.get('mnc')
- lac = tower_data.get('lac')
- cid = tower_data.get('cid')
-
- # Validate required fields
- if not all([mcc is not None, mnc is not None, lac is not None, cid is not None]):
- logger.warning(f"Tower data missing required fields: {tower_data}")
- return tower_data
-
- # Try cache lookup
- coords = lookup_cell_coordinates(mcc, mnc, lac, cid)
-
- if coords:
- # Found in cache - add coordinates immediately
- tower_data['lat'] = coords['lat']
- tower_data['lon'] = coords['lon']
- tower_data['source'] = 'cache'
-
- # Add optional fields if available
- if coords.get('azimuth') is not None:
- tower_data['azimuth'] = coords['azimuth']
- if coords.get('range_meters') is not None:
- tower_data['range_meters'] = coords['range_meters']
- if coords.get('operator'):
- tower_data['operator'] = coords['operator']
- if coords.get('radio'):
- tower_data['radio'] = coords['radio']
-
- logger.debug(f"Cache hit for tower: MCC={mcc} MNC={mnc} LAC={lac} CID={cid}")
- else:
- # Not in cache - mark as pending and queue for API lookup
- tower_data['status'] = 'pending'
- tower_data['source'] = 'unknown'
-
- # Queue for background geocoding (non-blocking)
- try:
- _geocoding_queue.put_nowait(tower_data.copy())
- logger.debug(f"Queued tower for geocoding: MCC={mcc} MNC={mnc} LAC={lac} CID={cid}")
- except queue.Full:
- logger.warning("Geocoding queue full, dropping tower")
-
- return tower_data
-
-
-def get_geocoding_queue() -> queue.Queue:
- """Get the geocoding queue for the background worker."""
- return _geocoding_queue
diff --git a/utils/logging.py b/utils/logging.py
index addbabe..3d2cc6a 100644
--- a/utils/logging.py
+++ b/utils/logging.py
@@ -28,4 +28,3 @@ wifi_logger = get_logger('intercept.wifi')
bluetooth_logger = get_logger('intercept.bluetooth')
adsb_logger = get_logger('intercept.adsb')
satellite_logger = get_logger('intercept.satellite')
-gsm_spy_logger = get_logger('intercept.gsm_spy')