mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-05-21 23:24:44 +02:00
K7ZVX
549ae4bdfb
- Environmental feed system with tick-based adapters - NWS Active Alerts: polls api.weather.gov, zone-based filtering - NOAA SWPC: Kp, SFI, R/S/G scales, band assessment, alert detection - Tropospheric ducting: Open-Meteo GFS refractivity profile, duct classification - !alerts command for active weather warnings - !solar / !hf commands for RF propagation (HF + UHF ducting) - Alert engine integration: severe weather, R3+ blackout, ducting events - LLM context injection for weather/propagation queries - Dashboard RF Propagation card with HF + UHF ducting display - EnvironmentalConfig with per-feed toggles in config.yaml
49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
"""Alerts command handler."""
|
|
|
|
import time
|
|
from datetime import datetime
|
|
|
|
from .base import CommandContext, CommandHandler
|
|
|
|
|
|
class AlertsCommand(CommandHandler):
|
|
"""Active weather alerts for mesh area."""
|
|
|
|
name = "alerts"
|
|
description = "Active weather alerts for mesh area"
|
|
usage = "!alerts"
|
|
|
|
def __init__(self, env_store):
|
|
self._env_store = env_store
|
|
|
|
async def execute(self, args: str, context: CommandContext) -> str:
|
|
"""Execute the alerts command."""
|
|
if not self._env_store:
|
|
return "Environmental feeds not enabled."
|
|
|
|
zones = self._env_store._mesh_zones
|
|
alerts = self._env_store.get_for_zones(zones)
|
|
|
|
if not alerts:
|
|
alerts = self._env_store.get_active(source="nws")
|
|
|
|
if not alerts:
|
|
return "No active weather alerts for the mesh area."
|
|
|
|
lines = [f"Active Alerts ({len(alerts)}):"]
|
|
for a in alerts[:5]:
|
|
# Format expiry time
|
|
expires = a.get("expires", 0)
|
|
if expires:
|
|
try:
|
|
dt = datetime.fromtimestamp(expires)
|
|
expires_str = dt.strftime("%b %d %H:%MZ")
|
|
except Exception:
|
|
expires_str = "Unknown"
|
|
else:
|
|
expires_str = "Unknown"
|
|
|
|
lines.append(f"* {a['event_type']} -- {a.get('area_desc', '')[:60]}")
|
|
lines.append(f" Until {expires_str}")
|
|
|
|
return "\n".join(lines)
|