"""Analysis tools for DX spots - band activity, DX entities, and time-based trends."""
import re
from collections import Counter, defaultdict
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from .dx_cluster import DXSpot
# DXCC prefix to entity mapping (common prefixes)
DXCC_PREFIXES = {
'K': 'United States', 'W': 'United States', 'N': 'United States', 'AA-AL': 'United States',
'VE': 'Canada', 'VA': 'Canada', 'VO': 'Canada',
'G': 'England', 'M': 'England', '2E': 'England',
'DL': 'Germany', 'DA': 'Germany', 'DB': 'Germany', 'DC': 'Germany', 'DD': 'Germany',
'F': 'France',
'I': 'Italy',
'EA': 'Spain',
'OH': 'Finland',
'SM': 'Sweden',
'LA': 'Norway',
'OZ': 'Denmark',
'PA': 'Netherlands',
'ON': 'Belgium',
'HB': 'Switzerland', 'HB0': 'Liechtenstein',
'OE': 'Austria',
'OK': 'Czech Republic',
'OM': 'Slovakia',
'SP': 'Poland',
'HA': 'Hungary',
'YU': 'Serbia',
'9A': 'Croatia',
'S5': 'Slovenia',
'LY': 'Lithuania',
'YL': 'Latvia',
'ES': 'Estonia',
'UR': 'Ukraine', 'UT': 'Ukraine', 'UY': 'Ukraine', 'UZ': 'Ukraine',
'R': 'Russia', 'RA': 'Russia', 'RK': 'Russia', 'RN': 'Russia', 'RU': 'Russia', 'RW': 'Russia',
'JA': 'Japan', 'JE': 'Japan', 'JF': 'Japan', 'JH': 'Japan', 'JI': 'Japan', 'JJ': 'Japan', 'JK': 'Japan', 'JL': 'Japan', 'JM': 'Japan', 'JN': 'Japan', 'JO': 'Japan', 'JP': 'Japan', 'JQ': 'Japan', 'JR': 'Japan', 'JS': 'Japan',
'VK': 'Australia',
'ZL': 'New Zealand',
'PY': 'Brazil', 'PP': 'Brazil', 'PR': 'Brazil', 'PS': 'Brazil', 'PT': 'Brazil', 'PU': 'Brazil', 'PV': 'Brazil', 'PW': 'Brazil', 'PX': 'Brazil',
'LU': 'Argentina',
'CE': 'Chile',
'XE': 'Mexico',
'YV': 'Venezuela',
'ZS': 'South Africa',
'5B': 'Cyprus',
'9H': 'Malta',
'SV': 'Greece',
'TA': 'Turkey', 'TB': 'Turkey', 'TC': 'Turkey',
'A4': 'Oman',
'A6': 'UAE',
'A7': 'Qatar',
'A9': 'Bahrain',
'AP': 'Pakistan',
'BV': 'Taiwan',
'BY': 'China', 'BA': 'China', 'BD': 'China',
'DU': 'Philippines',
'E2': 'Thailand', 'HS': 'Thailand',
'9M': 'Malaysia',
'9V': 'Singapore',
'YB': 'Indonesia',
'VU': 'India',
'JT': 'Mongolia',
'P2': 'Papua New Guinea',
'T3': 'Kiribati',
'ZK': 'New Zealand',
'3D2': 'Fiji',
'KH6': 'Hawaii',
'KL': 'Alaska',
'KP2': 'Virgin Islands',
'KP4': 'Puerto Rico',
'VP2E': 'Anguilla',
'VP2M': 'Montserrat',
'VP2V': 'British Virgin Islands',
'VP5': 'Turks and Caicos',
'VP9': 'Bermuda',
'ZF': 'Cayman Islands',
'8P': 'Barbados',
'J3': 'Grenada',
'J6': 'St. Lucia',
'J7': 'Dominica',
'J8': 'St. Vincent',
'V3': 'Belize',
'V4': 'St. Kitts and Nevis',
'FG': 'Guadeloupe',
'FM': 'Martinique',
'TI': 'Costa Rica',
'YN': 'Nicaragua',
'HK': 'Colombia',
'HC': 'Ecuador',
'OA': 'Peru',
'CP': 'Bolivia',
'PZ': 'Suriname',
'P4': 'Aruba',
'PJ2': 'Curacao',
'PJ4': 'Bonaire',
}
@dataclass
class BandActivity:
"""Band activity statistics."""
band: str
spot_count: int
unique_dxcc: int
avg_spots_per_hour: float
@dataclass
class DXCCStats:
"""DXCC entity statistics."""
entity: str
spot_count: int
bands: List[str]
first_seen: datetime
last_seen: datetime
@dataclass
class TimeBasedTrend:
"""Time-based propagation trends."""
hour: int
spot_count: int
active_bands: List[str]
top_dxcc: List[str]
class SpotAnalyzer:
"""Analyzer for DX spots providing comprehensive statistics."""
def __init__(self, spots: List[DXSpot]):
"""Initialize analyzer with spots.
Args:
spots: List of DXSpot objects to analyze
"""
self.spots = spots
def _extract_prefix(self, callsign: str) -> str:
"""Extract DXCC prefix from callsign.
Args:
callsign: Amateur radio callsign
Returns:
DXCC prefix
"""
# Handle portable operations (/P, /M, etc.)
base_call = callsign.split('/')[0]
# Try to match known prefixes (longest first)
for prefix in sorted(DXCC_PREFIXES.keys(), key=len, reverse=True):
if base_call.startswith(prefix):
return prefix
# Fallback: extract numeric prefix pattern
match = re.match(r'^([A-Z0-9]+?)(\d)', base_call)
if match:
return match.group(1) + match.group(2)
return base_call[:2]
def _get_dxcc_entity(self, callsign: str) -> str:
"""Get DXCC entity name from callsign.
Args:
callsign: Amateur radio callsign
Returns:
DXCC entity name
"""
prefix = self._extract_prefix(callsign)
return DXCC_PREFIXES.get(prefix, f"Unknown ({prefix})")
def analyze_band_activity(self) -> List[Dict[str, Any]]:
"""Analyze band activity across all spots.
Returns:
List of band activity statistics
"""
band_data = defaultdict(lambda: {
'spots': [],
'dxcc_entities': set()
})
for spot in self.spots:
if spot.band:
band_data[spot.band]['spots'].append(spot)
entity = self._get_dxcc_entity(spot.dx_callsign)
band_data[spot.band]['dxcc_entities'].add(entity)
# Calculate statistics
results = []
for band, data in sorted(band_data.items(), key=lambda x: x[0]):
spot_count = len(data['spots'])
# Calculate time span
if data['spots']:
timestamps = [s.timestamp for s in data['spots']]
time_span = (max(timestamps) - min(timestamps)).total_seconds() / 3600
if time_span == 0:
time_span = 1 # Avoid division by zero
avg_per_hour = spot_count / time_span
else:
avg_per_hour = 0
results.append({
'band': band,
'spot_count': spot_count,
'unique_dxcc': len(data['dxcc_entities']),
'avg_spots_per_hour': round(avg_per_hour, 2),
'percentage': 0 # Will calculate after
})
# Calculate percentages
total_spots = sum(r['spot_count'] for r in results)
if total_spots > 0:
for result in results:
result['percentage'] = round(
(result['spot_count'] / total_spots) * 100, 1
)
# Sort by spot count
results.sort(key=lambda x: x['spot_count'], reverse=True)
return results
def analyze_dxcc_entities(self, top_n: int = 20) -> List[Dict[str, Any]]:
"""Analyze DX entity statistics.
Args:
top_n: Number of top entities to return
Returns:
List of DXCC entity statistics
"""
entity_data = defaultdict(lambda: {
'spots': [],
'bands': set(),
'timestamps': []
})
for spot in self.spots:
entity = self._get_dxcc_entity(spot.dx_callsign)
entity_data[entity]['spots'].append(spot)
if spot.band:
entity_data[entity]['bands'].add(spot.band)
entity_data[entity]['timestamps'].append(spot.timestamp)
# Build results
results = []
for entity, data in entity_data.items():
results.append({
'entity': entity,
'spot_count': len(data['spots']),
'bands': sorted(data['bands']),
'first_seen': min(data['timestamps']).isoformat(),
'last_seen': max(data['timestamps']).isoformat(),
})
# Sort by spot count and return top N
results.sort(key=lambda x: x['spot_count'], reverse=True)
return results[:top_n]
def analyze_time_trends(self) -> List[Dict[str, Any]]:
"""Analyze time-based propagation trends.
Returns:
List of hourly trend statistics
"""
hourly_data = defaultdict(lambda: {
'spots': [],
'bands': set(),
'entities': []
})
for spot in self.spots:
hour = spot.timestamp.hour
hourly_data[hour]['spots'].append(spot)
if spot.band:
hourly_data[hour]['bands'].add(spot.band)
hourly_data[hour]['entities'].append(
self._get_dxcc_entity(spot.dx_callsign)
)
# Build results
results = []
for hour in range(24):
data = hourly_data[hour]
spot_count = len(data['spots'])
# Get top 3 DX entities for this hour
if data['entities']:
entity_counts = Counter(data['entities'])
top_entities = [e for e, _ in entity_counts.most_common(3)]
else:
top_entities = []
results.append({
'hour': f"{hour:02d}:00 UTC",
'spot_count': spot_count,
'active_bands': sorted(data['bands']) if data['bands'] else [],
'top_dxcc': top_entities,
})
return results
def get_rare_dx_alerts(self, threshold: int = 3) -> List[Dict[str, Any]]:
"""Identify rare DX entities (low spot count).
Args:
threshold: Maximum spot count to consider rare
Returns:
List of rare DX alerts
"""
entity_counts = Counter()
entity_spots = defaultdict(list)
for spot in self.spots:
entity = self._get_dxcc_entity(spot.dx_callsign)
entity_counts[entity] += 1
entity_spots[entity].append(spot)
# Find rare entities
rare = []
for entity, count in entity_counts.items():
if count <= threshold and entity != "Unknown":
spots = entity_spots[entity]
rare.append({
'entity': entity,
'spot_count': count,
'callsigns': list(set(s.dx_callsign for s in spots)),
'frequencies': [s.frequency for s in spots],
'bands': list(set(s.band for s in spots if s.band)),
'last_seen': max(s.timestamp for s in spots).isoformat(),
})
# Sort by spot count (rarest first)
rare.sort(key=lambda x: x['spot_count'])
return rare
def generate_comprehensive_report(self) -> Dict[str, Any]:
"""Generate a comprehensive analysis report.
Returns:
Dictionary containing all analysis results
"""
if not self.spots:
return {
'error': 'No spots available for analysis',
'spot_count': 0
}
# Calculate time range
timestamps = [s.timestamp for s in self.spots]
time_range = {
'start': min(timestamps).isoformat(),
'end': max(timestamps).isoformat(),
'duration_hours': round(
(max(timestamps) - min(timestamps)).total_seconds() / 3600, 2
)
}
return {
'summary': {
'total_spots': len(self.spots),
'time_range': time_range,
'unique_dxcc_entities': len(set(
self._get_dxcc_entity(s.dx_callsign) for s in self.spots
)),
},
'band_activity': self.analyze_band_activity(),
'top_dxcc_entities': self.analyze_dxcc_entities(top_n=15),
'hourly_trends': self.analyze_time_trends(),
'rare_dx_alerts': self.get_rare_dx_alerts(threshold=3),
}
