674 lines
26 KiB
Python
Executable File
674 lines
26 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
PirateWeatherADV — Weather script for mesh network auto-responders.
|
|
Uses the Pirate Weather API for data and OpenStreetMap Nominatim for geocoding.
|
|
|
|
TWO SEPARATE TRIGGERS:
|
|
|
|
weather / w — Current conditions only (single message)
|
|
Output example:
|
|
Weather for Peterborough, Ontario, Canada: Misty.
|
|
Currently 40°F / 4°C (feels like 32°F / 0°C).
|
|
Today: High 47°F / 8°C, Low 38°F / 3°C.
|
|
Humidity: 96%, Wind: 9 mph.
|
|
|
|
forecast / f — 7-day forecast (split into as many messages as needed)
|
|
Output example (message 1 of 2):
|
|
7-Day Forecast - Peterborough, Ontario, Canada:
|
|
Mon: Misty. Hi 40°F/4°C Lo 33°F/1°C 💧85%
|
|
Tue: Cloudy. Hi 47°F/8°C Lo 38°F/3°C 💧40%
|
|
Wed: Partly Cloudy. Hi 52°F/11°C Lo 41°F/5°C 💧15%
|
|
Output example (message 2 of 2):
|
|
Thu: Clear. Hi 55°F/13°C Lo 43°F/6°C
|
|
Fri: Rain. Hi 48°F/9°C Lo 39°F/4°C 💧90%
|
|
Sat: Cloudy. Hi 44°F/7°C Lo 36°F/2°C 💧50%
|
|
Sun: Clear. Hi 50°F/10°C Lo 38°F/3°C 💧10%
|
|
|
|
GPS MODE (no location typed):
|
|
Both triggers fall back to GPS coordinates in this order:
|
|
1. FROM_LAT / FROM_LON — GPS sent by the requesting node
|
|
2. LOCAL_LAT / LOCAL_LON — local node GPS (also accepts MM_LAT / MM_LON)
|
|
GPS mode output uses emoji format with a reverse-geocoded city name.
|
|
GPS mode example output:
|
|
📍 Peterborough, CA
|
|
🌡️ 40°F / 4°C (feels like 32°F / 0°C)
|
|
📊 Forecast: Misty
|
|
↕️ High: 47°F / 8°C Low: 38°F / 3°C
|
|
💧 Humidity: 96% 💨 Wind: 9 mph
|
|
|
|
INPUT FORMATS (spaces after commas are fine, all produce identical results):
|
|
weather peterborough,ontario,canada
|
|
weather peterborough, ontario, canada
|
|
w peterborough, ontario, canada
|
|
forecast peterborough, ontario, canada
|
|
f peterborough, ontario, canada
|
|
|
|
MODE DETECTION ORDER (how the script decides weather vs forecast):
|
|
1. TRIGGER env var — MeshMonitor sets this to the matched trigger pattern.
|
|
Most reliable source; checked first.
|
|
2. MESSAGE env var — Full original message typed by the user. Fallback if
|
|
TRIGGER alone doesn't identify the mode.
|
|
3. PARAM_mode — Explicit override env var ('forecast' or 'weather').
|
|
Useful for Timed Events.
|
|
4. Leading keyword — Keyword at the start of CLI arg or PARAM_location
|
|
(e.g. "forecast peterborough,ontario,canada").
|
|
5. Default — 'weather' if none of the above match.
|
|
|
|
All weather data is sourced exclusively from Pirate Weather (pirateweather.net).
|
|
Location names are resolved via OpenStreetMap Nominatim (free, no API key needed).
|
|
|
|
Requirements:
|
|
- Python 3.6+
|
|
- PIRATE_WEATHER_API_KEY environment variable (get free key at https://pirateweather.net/)
|
|
|
|
Setup:
|
|
1. Get a free API key from https://pirateweather.net/
|
|
2. Add to docker-compose.yaml environment variables:
|
|
- PIRATE_WEATHER_API_KEY=your_api_key_here
|
|
- LOCAL_LAT=your_latitude # optional GPS fallback for local node
|
|
- LOCAL_LON=your_longitude # (MM_LAT / MM_LON also accepted as aliases)
|
|
3. Ensure volume mapping in docker-compose.yaml:
|
|
- ./scripts:/data/scripts
|
|
4. Copy PirateWeatherADV.py to scripts/ directory
|
|
5. Make executable: chmod +x scripts/PirateWeatherADV.py
|
|
|
|
MeshMonitor Auto Responder Configuration:
|
|
Navigate to Settings → Automation → Auto Responder → Add Trigger.
|
|
Create TWO trigger entries, each pointing to the same script:
|
|
|
|
Trigger 1 — Current conditions (weather / w):
|
|
Trigger field: weather, weather {location:.+}, w {location:.+}
|
|
Response type: Script
|
|
Script path: /data/scripts/PirateWeatherADV.py
|
|
|
|
Trigger 2 — 7-day forecast (forecast / f):
|
|
Trigger field: forecast, forecast {location:.+}, f {location:.+}
|
|
Response type: Script
|
|
Script path: /data/scripts/PirateWeatherADV.py
|
|
|
|
The comma-separated patterns in each trigger field allow one entry to handle
|
|
both the bare keyword (GPS mode) and the keyword + location variants.
|
|
MeshMonitor sets the TRIGGER env var to the matched pattern, which the script
|
|
uses to reliably determine weather vs forecast mode regardless of whether
|
|
PARAM_location contains the keyword.
|
|
|
|
Local testing:
|
|
Current conditions by location:
|
|
TEST_MODE=true PARAM_location="peterborough,ontario,canada" PIRATE_WEATHER_API_KEY=your_key python3 PirateWeatherADV.py
|
|
7-day forecast by location:
|
|
TEST_MODE=true PARAM_location="forecast peterborough,ontario,canada" PIRATE_WEATHER_API_KEY=your_key python3 PirateWeatherADV.py
|
|
GPS mode — current conditions:
|
|
TEST_MODE=true FROM_LAT=43.55 FROM_LON=-78.49 PIRATE_WEATHER_API_KEY=your_key python3 PirateWeatherADV.py
|
|
GPS mode — forecast (simulate forecast trigger):
|
|
TEST_MODE=true TRIGGER=forecast FROM_LAT=43.55 FROM_LON=-78.49 PIRATE_WEATHER_API_KEY=your_key python3 PirateWeatherADV.py
|
|
CLI argument — current conditions (Timed Events):
|
|
TEST_MODE=true PIRATE_WEATHER_API_KEY=your_key python3 PirateWeatherADV.py peterborough,ontario,canada
|
|
CLI argument — forecast (Timed Events):
|
|
TEST_MODE=true PIRATE_WEATHER_API_KEY=your_key python3 PirateWeatherADV.py forecast peterborough,ontario,canada
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
import urllib.request
|
|
import urllib.parse
|
|
from typing import Optional, Tuple, Dict, List, Any
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Configuration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
PIRATE_WEATHER_API_KEY = os.environ.get('PIRATE_WEATHER_API_KEY', '')
|
|
TEST_MODE = os.environ.get('TEST_MODE', 'false').lower() == 'true'
|
|
|
|
# MeshMonitor enforces a 200-char max per message and a 10-second script timeout.
|
|
# Network timeouts: geocode 3s + API 4s + reverse geocode 3s = 10s worst case.
|
|
MSG_LIMIT = 200
|
|
|
|
# Day-of-week abbreviations indexed by Python's tm_wday (0=Mon … 6=Sun)
|
|
DAY_NAMES = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun']
|
|
|
|
# Trigger keywords that identify forecast mode
|
|
FORECAST_KEYWORDS = ('forecast', 'f')
|
|
# Trigger keywords that identify weather (current conditions) mode
|
|
WEATHER_KEYWORDS = ('weather', 'w')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Unit conversion
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def to_c(f: float) -> float:
|
|
"""Convert Fahrenheit to Celsius."""
|
|
return (f - 32) * 5 / 9
|
|
|
|
|
|
def fmt_temp(f: float) -> str:
|
|
"""Format a temperature as XX°F / XX°C."""
|
|
return f'{f:.0f}°F / {to_c(f):.0f}°C'
|
|
|
|
|
|
def fmt_temp_compact(f: float) -> str:
|
|
"""Compact temperature format for forecast lines: XX°F/XX°C."""
|
|
return f'{f:.0f}°F/{to_c(f):.0f}°C'
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Coordinate helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _parse_coords(lat_str: str, lon_str: str, label: str) -> Optional[Tuple[float, float]]:
|
|
"""Parse and range-validate a lat/lon pair. Returns (lat, lon) or None."""
|
|
try:
|
|
lat = float(lat_str)
|
|
lon = float(lon_str)
|
|
if not (-90 <= lat <= 90) or not (-180 <= lon <= 180):
|
|
print(f'Coordinates out of valid range for {label}: {lat}, {lon}', file=sys.stderr)
|
|
return None
|
|
return (lat, lon)
|
|
except ValueError:
|
|
print(f'Invalid coordinate values for {label}: {lat_str}, {lon_str}', file=sys.stderr)
|
|
return None
|
|
|
|
|
|
def geocode_location(location: str) -> Optional[Tuple[float, float]]:
|
|
"""
|
|
Forward-geocode a location string to (lat, lon) using Nominatim.
|
|
Returns None if the location cannot be found or the request fails.
|
|
"""
|
|
try:
|
|
params = {'q': location, 'format': 'json', 'limit': 1}
|
|
url = 'https://nominatim.openstreetmap.org/search?' + urllib.parse.urlencode(params)
|
|
req = urllib.request.Request(url, headers={'User-Agent': 'PirateWeatherADV/1.0'})
|
|
with urllib.request.urlopen(req, timeout=3) as resp:
|
|
data = json.loads(resp.read().decode('utf-8'))
|
|
if data:
|
|
return (float(data[0]['lat']), float(data[0]['lon']))
|
|
except Exception as e:
|
|
print(f'Geocoding error for "{location}": {e}', file=sys.stderr)
|
|
return None
|
|
|
|
|
|
def reverse_geocode_city(lat: float, lon: float) -> str:
|
|
"""
|
|
Reverse-geocode (lat, lon) to a city name using Nominatim.
|
|
Returns a raw coordinate string if the lookup fails.
|
|
"""
|
|
try:
|
|
params = {'lat': lat, 'lon': lon, 'format': 'json', 'zoom': 10}
|
|
url = 'https://nominatim.openstreetmap.org/reverse?' + urllib.parse.urlencode(params)
|
|
req = urllib.request.Request(url, headers={'User-Agent': 'PirateWeatherADV/1.0'})
|
|
with urllib.request.urlopen(req, timeout=3) as resp:
|
|
data = json.loads(resp.read().decode('utf-8'))
|
|
|
|
address = data.get('address') or {}
|
|
city = (
|
|
address.get('city')
|
|
or address.get('town')
|
|
or address.get('village')
|
|
or address.get('hamlet')
|
|
or address.get('county')
|
|
)
|
|
state = address.get('state', '')
|
|
country_code = address.get('country_code', '').upper()
|
|
|
|
if city:
|
|
if state and country_code == 'US':
|
|
return f'{city}, {state}'
|
|
elif country_code:
|
|
return f'{city}, {country_code}'
|
|
else:
|
|
return city
|
|
except Exception as e:
|
|
print(f'Reverse geocode error: {e}', file=sys.stderr)
|
|
|
|
return f'{lat:.4f}, {lon:.4f}'
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Label formatter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _fmt_label(location: str) -> str:
|
|
"""
|
|
Build a clean display label from a raw location string.
|
|
Splits on commas, capitalises each part carefully so:
|
|
- Abbreviations (ON, NY, CA) are preserved in upper case
|
|
- Apostrophes (St. John's) are not mangled by str.title()
|
|
- Empty parts from double/trailing commas are skipped
|
|
"""
|
|
def _fmt_part(part: str) -> str:
|
|
words = part.strip().split()
|
|
out = []
|
|
for w in words:
|
|
if not w:
|
|
continue
|
|
if w.isupper() and len(w) <= 3:
|
|
out.append(w)
|
|
else:
|
|
out.append(w[0].upper() + w[1:])
|
|
return ' '.join(out)
|
|
|
|
parts = [p.strip() for p in location.split(',') if p.strip()]
|
|
return ', '.join(_fmt_part(p) for p in parts)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Input resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def resolve_input() -> Tuple[Optional[Tuple[float, float]], Optional[str], str, str]:
|
|
"""
|
|
Determine coordinates, display label, output mode (weather/forecast),
|
|
and input mode (location/gps).
|
|
|
|
Returns:
|
|
(coords, label, output_mode, input_mode)
|
|
|
|
coords — (lat, lon) or None if nothing could be resolved
|
|
label — display name string, or None if GPS mode
|
|
output_mode — 'weather' (current conditions) or 'forecast' (7-day)
|
|
input_mode — 'location' or 'gps'
|
|
|
|
Resolution order for location:
|
|
1. CLI arguments (e.g. Timed Events)
|
|
2. PARAM_location env var (trigger pipeline)
|
|
|
|
Resolution order for coordinates when no location string given:
|
|
3. FROM_LAT / FROM_LON — requesting node GPS
|
|
4. LOCAL_LAT / LOCAL_LON or MM_LAT / MM_LON — local node GPS
|
|
|
|
Output mode is determined by:
|
|
- The leading keyword in the location string (forecast/f vs weather/w)
|
|
- PARAM_mode env var as a fallback override ('forecast' or 'f')
|
|
- Defaults to 'weather' if no keyword is found
|
|
"""
|
|
# ---------------------------------------------------------------------------
|
|
# Determine output_mode from multiple sources, in priority order:
|
|
#
|
|
# 1. TRIGGER env var — MeshMonitor sets this to the pattern that matched
|
|
# e.g. "forecast {location:.+}" or "forecast" — most reliable source.
|
|
# 2. MESSAGE env var — full original message, e.g. "forecast peterborough..."
|
|
# Used as fallback when TRIGGER doesn't contain the keyword.
|
|
# 3. PARAM_mode env var — explicit override, useful for Timed Events.
|
|
# 4. Leading keyword in the CLI arg or PARAM_location string.
|
|
# 5. Default: 'weather'
|
|
# ---------------------------------------------------------------------------
|
|
|
|
output_mode = 'weather' # default
|
|
|
|
# 1. Check TRIGGER env var (most reliable — set by MeshMonitor)
|
|
trigger = os.environ.get('TRIGGER', '').strip().lower()
|
|
if any(trigger == kw or trigger.startswith(kw + ' ') or trigger.startswith(kw + ',')
|
|
for kw in FORECAST_KEYWORDS):
|
|
output_mode = 'forecast'
|
|
elif any(trigger == kw or trigger.startswith(kw + ' ') or trigger.startswith(kw + ',')
|
|
for kw in WEATHER_KEYWORDS):
|
|
output_mode = 'weather'
|
|
|
|
# 2. Check full MESSAGE env var as fallback
|
|
if output_mode == 'weather': # only if not already determined
|
|
message = os.environ.get('MESSAGE', '').strip().lower()
|
|
if any(message == kw or message.startswith(kw + ' ') or message.startswith(kw + ',')
|
|
for kw in FORECAST_KEYWORDS):
|
|
output_mode = 'forecast'
|
|
|
|
# 3. PARAM_mode explicit override
|
|
param_mode = os.environ.get('PARAM_mode', '').strip().lower()
|
|
if param_mode in ('forecast', 'f'):
|
|
output_mode = 'forecast'
|
|
elif param_mode in ('weather', 'w'):
|
|
output_mode = 'weather'
|
|
|
|
# Gather raw location string: CLI args first, then PARAM_location
|
|
location = ''
|
|
if len(sys.argv) > 1:
|
|
location = ' '.join(sys.argv[1:]).strip()
|
|
if not location:
|
|
location = os.environ.get('PARAM_location', '').strip()
|
|
|
|
# 4. Detect and strip leading keyword from the location string itself
|
|
# (handles CLI args and direct PARAM_location with keyword prefix)
|
|
loc_lower = location.lower()
|
|
stripped = False
|
|
for kw in FORECAST_KEYWORDS:
|
|
if loc_lower == kw or loc_lower.startswith(kw + ' ') or loc_lower.startswith(kw + ','):
|
|
output_mode = 'forecast'
|
|
location = location[len(kw):].strip()
|
|
stripped = True
|
|
break
|
|
if not stripped:
|
|
for kw in WEATHER_KEYWORDS:
|
|
if loc_lower == kw or loc_lower.startswith(kw + ' ') or loc_lower.startswith(kw + ','):
|
|
output_mode = 'weather'
|
|
location = location[len(kw):].strip()
|
|
break
|
|
|
|
# If we have a location string, geocode it
|
|
if location and location.lower() not in ('help', 'h', '?'):
|
|
# Normalise: strip spaces around commas, filter empty parts
|
|
parts = [p.strip() for p in location.split(',') if p.strip()]
|
|
location_normalised = ','.join(parts)
|
|
coords = geocode_location(location_normalised)
|
|
label = _fmt_label(location)
|
|
if coords:
|
|
return (coords, label, output_mode, 'location')
|
|
else:
|
|
return (None, label, output_mode, 'location')
|
|
|
|
# No location string — fall back to GPS coordinates
|
|
from_lat = os.environ.get('FROM_LAT', '').strip()
|
|
from_lon = os.environ.get('FROM_LON', '').strip()
|
|
if from_lat and from_lon:
|
|
coords = _parse_coords(from_lat, from_lon, 'requesting node')
|
|
if coords:
|
|
return (coords, None, output_mode, 'gps')
|
|
|
|
local_lat = (os.environ.get('LOCAL_LAT', '') or os.environ.get('MM_LAT', '')).strip()
|
|
local_lon = (os.environ.get('LOCAL_LON', '') or os.environ.get('MM_LON', '')).strip()
|
|
if local_lat and local_lon:
|
|
coords = _parse_coords(local_lat, local_lon, 'local node')
|
|
if coords:
|
|
return (coords, None, output_mode, 'gps')
|
|
|
|
return (None, None, output_mode, 'gps')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Weather fetch
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_weather(lat: float, lon: float) -> Dict[str, Any]:
|
|
"""
|
|
Fetch current conditions and 7-day daily forecast from Pirate Weather.
|
|
|
|
Returns {'data': {...}} on success or {'error': '...'} on failure.
|
|
"""
|
|
if not PIRATE_WEATHER_API_KEY:
|
|
return {'error': 'PIRATE_WEATHER_API_KEY not set. Get a free key at pirateweather.net.'}
|
|
|
|
try:
|
|
url = f'https://api.pirateweather.net/forecast/{PIRATE_WEATHER_API_KEY}/{lat},{lon}'
|
|
req = urllib.request.Request(url)
|
|
with urllib.request.urlopen(req, timeout=4) as resp:
|
|
data = json.loads(resp.read().decode('utf-8'))
|
|
|
|
current = data.get('currently') or {}
|
|
daily_periods = (data.get('daily') or {}).get('data') or []
|
|
today = daily_periods[0] if daily_periods else {}
|
|
|
|
temp = current.get('temperature')
|
|
feels_like = current.get('apparentTemperature')
|
|
summary = current.get('summary') or 'N/A'
|
|
humidity = current.get('humidity') or 0
|
|
wind_speed = current.get('windSpeed') or 0
|
|
high_temp = today.get('temperatureHigh')
|
|
low_temp = today.get('temperatureLow')
|
|
|
|
if temp is None or feels_like is None:
|
|
return {'error': 'Pirate Weather returned incomplete data (missing temperature).'}
|
|
|
|
# Timezone offset in seconds (offset field is a float, e.g. -5.0 or 5.5)
|
|
tz_offset_sec = int((data.get('offset') or 0) * 3600)
|
|
|
|
# Build 7-day forecast list from all available daily periods.
|
|
# Use 8 entries so we always have a full 7-day lookahead regardless
|
|
# of what day of the week today is (index 0 = today, so 8 covers today
|
|
# plus 7 future days, and build_forecast_messages caps output at 7 days).
|
|
forecast = []
|
|
for day in daily_periods[:8]:
|
|
day_time = day.get('time')
|
|
day_high = day.get('temperatureHigh')
|
|
day_low = day.get('temperatureLow')
|
|
day_sum = day.get('summary') or 'N/A'
|
|
day_precip = day.get('precipProbability') or 0
|
|
|
|
if day_time is None or day_high is None or day_low is None:
|
|
continue
|
|
|
|
forecast.append({
|
|
'time': day_time,
|
|
'summary': day_sum,
|
|
'high': day_high,
|
|
'low': day_low,
|
|
'precip_prob': day_precip,
|
|
})
|
|
|
|
return {'data': {
|
|
'temp': temp,
|
|
'feels_like': feels_like,
|
|
'summary': summary,
|
|
'humidity': humidity,
|
|
'wind_speed': wind_speed,
|
|
'high_temp': high_temp,
|
|
'low_temp': low_temp,
|
|
'forecast': forecast,
|
|
'tz_offset_sec': tz_offset_sec,
|
|
}}
|
|
|
|
except urllib.error.HTTPError as e:
|
|
return {'error': f'Weather API error: {e.code} {e.reason}'}
|
|
except urllib.error.URLError as e:
|
|
return {'error': f'Network error: {e.reason}'}
|
|
except json.JSONDecodeError:
|
|
return {'error': 'Invalid response from Pirate Weather API'}
|
|
except Exception as e:
|
|
# Sanitise exception message — strip API key if it appears
|
|
err_msg = str(e).replace(PIRATE_WEATHER_API_KEY, '***') if PIRATE_WEATHER_API_KEY else str(e)
|
|
return {'error': f'Unexpected error: {err_msg}'}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Response formatters
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def format_weather(label: str, d: dict) -> str:
|
|
"""
|
|
Format current conditions as a single message string.
|
|
|
|
Example:
|
|
Weather for Peterborough, Ontario, Canada: Misty.
|
|
Currently 40°F / 4°C (feels like 32°F / 0°C).
|
|
Today: High 47°F / 8°C, Low 38°F / 3°C.
|
|
Humidity: 96%, Wind: 9 mph.
|
|
"""
|
|
msg = (
|
|
f'Weather for {label}: {d["summary"]}. '
|
|
f'Currently {fmt_temp(d["temp"])} (feels like {fmt_temp(d["feels_like"])}). '
|
|
)
|
|
if d['high_temp'] is not None and d['low_temp'] is not None:
|
|
msg += f'Today: High {fmt_temp(d["high_temp"])}, Low {fmt_temp(d["low_temp"])}. '
|
|
msg += f'Humidity: {d["humidity"]*100:.0f}%, Wind: {d["wind_speed"]:.0f} mph.'
|
|
return msg
|
|
|
|
|
|
def format_weather_gps(lat: float, lon: float, d: dict) -> str:
|
|
"""
|
|
Format current conditions in emoji style for GPS mode.
|
|
|
|
Example:
|
|
📍 Peterborough, CA
|
|
🌡️ 40°F / 4°C (feels like 32°F / 0°C)
|
|
📊 Forecast: Misty
|
|
↕️ High: 47°F / 8°C Low: 38°F / 3°C
|
|
💧 Humidity: 96% 💨 Wind: 9 mph
|
|
"""
|
|
city = reverse_geocode_city(lat, lon)
|
|
lines = [
|
|
f'📍 {city}',
|
|
f'🌡️ {fmt_temp(d["temp"])} (feels like {fmt_temp(d["feels_like"])})',
|
|
f'📊 Forecast: {d["summary"]}',
|
|
]
|
|
if d['high_temp'] is not None and d['low_temp'] is not None:
|
|
lines.append(f'↕️ High: {fmt_temp(d["high_temp"])} Low: {fmt_temp(d["low_temp"])}')
|
|
lines.append(f'💧 Humidity: {d["humidity"]*100:.0f}% 💨 Wind: {d["wind_speed"]:.0f} mph')
|
|
return '\n'.join(lines)
|
|
|
|
|
|
def build_forecast_messages(label: str, d: dict) -> List[str]:
|
|
"""
|
|
Build the 7-day forecast as a list of messages, each within MSG_LIMIT bytes.
|
|
|
|
The first message starts with a header line. Subsequent day lines are packed
|
|
into messages greedily. If a day line starts a new message, that message
|
|
begins directly with the day line (no repeated header).
|
|
|
|
Returns a list of message strings (may be empty if no forecast data).
|
|
"""
|
|
forecast = d.get('forecast', [])
|
|
if not forecast:
|
|
return []
|
|
|
|
tz_offset = d.get('tz_offset_sec', 0)
|
|
header = f'7-Day Forecast - {label}:'
|
|
|
|
# Build all day lines first — cap at 7 days regardless of how many
|
|
# entries were returned (we fetch 8 to guarantee 7 future days exist)
|
|
day_lines = []
|
|
for day in forecast[:7]:
|
|
try:
|
|
local_ts = day['time'] + tz_offset
|
|
day_name = DAY_NAMES[time.gmtime(local_ts).tm_wday]
|
|
except Exception:
|
|
day_name = '???'
|
|
|
|
high = day['high']
|
|
low = day['low']
|
|
summ = day['summary']
|
|
precip = day['precip_prob']
|
|
|
|
hi_str = fmt_temp_compact(high)
|
|
lo_str = fmt_temp_compact(low)
|
|
precip_str = f' 💧{precip*100:.0f}%' if precip else ''
|
|
|
|
line = f'{day_name}: {summ}. Hi {hi_str} Lo {lo_str}{precip_str}'
|
|
# Safety: if a single line exceeds MSG_LIMIT (e.g. very long API summary),
|
|
# truncate it so MeshMonitor never silently drops characters.
|
|
while len(line.encode('utf-8')) > MSG_LIMIT:
|
|
line = line[:-1]
|
|
day_lines.append(line)
|
|
|
|
if not day_lines:
|
|
return []
|
|
|
|
# Pack day lines into messages greedily, respecting MSG_LIMIT bytes
|
|
messages = []
|
|
current = header # first message always starts with the header
|
|
|
|
for line in day_lines:
|
|
candidate = current + '\n' + line if current else line
|
|
|
|
if len(candidate.encode('utf-8')) <= MSG_LIMIT:
|
|
current = candidate
|
|
else:
|
|
# Current message is full — save it and start a new one
|
|
if current:
|
|
messages.append(current)
|
|
# New message starts with just the day line (no repeated header)
|
|
current = line
|
|
|
|
# Don't forget the last message
|
|
if current:
|
|
messages.append(current)
|
|
|
|
return messages
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Output helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def emit(messages: List[str]) -> None:
|
|
"""
|
|
Print the correct MeshMonitor JSON output format and flush stdout.
|
|
|
|
Single message → {"response": "..."}
|
|
Multiple messages → {"responses": ["msg1", "msg2", ...]}
|
|
"""
|
|
if not messages:
|
|
print(json.dumps({'response': 'No data to display.'}))
|
|
elif len(messages) == 1:
|
|
print(json.dumps({'response': messages[0]}))
|
|
else:
|
|
print(json.dumps({'responses': messages}))
|
|
sys.stdout.flush()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Entry point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main():
|
|
try:
|
|
coords, label, output_mode, input_mode = resolve_input()
|
|
|
|
if TEST_MODE:
|
|
trigger = os.environ.get('TRIGGER', '(none)')
|
|
print(
|
|
f'TRIGGER={trigger!r} | output_mode={output_mode} | '
|
|
f'input_mode={input_mode} | coords={coords} | label={label}',
|
|
file=sys.stderr
|
|
)
|
|
|
|
# Handle no-location / no-GPS case
|
|
if coords is None:
|
|
if input_mode == 'location' and label:
|
|
emit([f'Could not find location: "{label}". '
|
|
'Try e.g. "weather peterborough,ontario,canada".'])
|
|
else:
|
|
emit([
|
|
'No location or GPS data available. '
|
|
'Options: (1) Send "weather city,province,country". '
|
|
'(2) Use CLI arg in Timed Event: "peterborough,ontario,canada". '
|
|
'(3) Ensure node has GPS (FROM_LAT/FROM_LON). '
|
|
'(4) Set LOCAL_LAT/LOCAL_LON in docker-compose.'
|
|
])
|
|
return
|
|
|
|
lat, lon = coords
|
|
|
|
result = get_weather(lat, lon)
|
|
if 'error' in result:
|
|
emit([f'Error: {result["error"]}'])
|
|
return
|
|
|
|
d = result['data']
|
|
|
|
if output_mode == 'forecast':
|
|
# Use typed label or reverse-geocoded city name for GPS mode
|
|
fc_label = label if label else reverse_geocode_city(lat, lon)
|
|
messages = build_forecast_messages(fc_label, d)
|
|
if not messages:
|
|
emit(['No forecast data available.'])
|
|
else:
|
|
emit(messages)
|
|
|
|
if TEST_MODE:
|
|
for i, msg in enumerate(messages, 1):
|
|
print(f'\n--- FORECAST MSG {i} ---\n{msg}', file=sys.stderr)
|
|
print('--- END TEST ---\n', file=sys.stderr)
|
|
|
|
else: # output_mode == 'weather'
|
|
if input_mode == 'location':
|
|
msg = format_weather(label, d)
|
|
else:
|
|
msg = format_weather_gps(lat, lon, d)
|
|
|
|
emit([msg])
|
|
|
|
if TEST_MODE:
|
|
print(f'\n--- WEATHER MSG ---\n{msg}\n--- END TEST ---\n', file=sys.stderr)
|
|
|
|
except Exception as e:
|
|
emit([f'Error: {str(e)}'])
|
|
if TEST_MODE:
|
|
import traceback
|
|
traceback.print_exc(file=sys.stderr)
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|