Content
I have a WeatherFlow Tempest weather station set up in Minneapolis, and I wanted to have its observations sent to APRS so that others could see our weather. I had set up a process to get data off of the weather station receiver, and I just needed to format it into a proper APRS packet, then send it to APRS-IS/CWOP.
This was intended to be a custom script without any other hardware or software dependencies. I wanted it to run on a server and publish the data every five minutes. This is a list of the data I wanted to send:
- Temperature
- Humidity
- Pressure
- Wind Speed
- Wind Direction
- Wind Gusts
- Rain Amounts
- Station Position
- A short station comment ("Tempest with Custom Script")
APRS data shown on aprs.fi
This is what a properly formatted post should look like:
MYCALL>APRS,TCPIP*:!DDMM.MMN/DDDMM.MMW_220/008g012t076r000p000P016h73b10113Tempest with Custom Script
Here are the files and folders that I created on the server:
tempest-aprs/
docker-compose.yml
.env
app/
Dockerfile
requirements.txt
tempest_aprs.py
state/
This was the docker-compose.yml file:
services:
tempest-aprs:
build:
context: ./app
dockerfile: Dockerfile
container_name: tempest-aprs
restart: unless-stopped
env_file:
- .env
environment:
- PYTHONUNBUFFERED=1
volumes:
- ./state:/state
command: python -u /app/tempest_aprs.py --loop
This was the .env file contents:
TEMPEST_STATION_ID=YOUR_TEMPEST_STATION_ID
TEMPEST_TOKEN=YOUR_TEMPEST_TOKEN
APRS_ENABLED=true
APRS_CALLSIGN=MYCALLSIGN
APRS_PASSCODE=YOUR_APRS_IS_PASSCODE
APRS_SERVER=cwop.aprs.net
APRS_PORT=14580
APRS_LAT=YOUR_LATITUDE
APRS_LON=YOUR_LONGITUDE
APRS_MIN_INTERVAL_SECONDS=300
APRS_STATE_FILE=YOUR_STATE_DIRECTORY/tempest-aprs-last-sent
APRS_RAIN_STATE_FILE=YOUR_STATE_DIRECTORY/tempest-rain-state.json
APRS_PRESSURE_OFFSET_HPA=0
APRS_COMMENT=Tempest with Custom Script
TZ=YOUR_TIMEZONE
POLL_INTERVAL_SECONDS=60
This is the Dockerfile content:
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY tempest_aprs.py .
CMD ["python", "-u", "/app/tempest_aprs.py", "--loop"]
Requirements.txt contained just this:
requests==2.32.4tempest_aprs.py contained much more code. This is the heart of the script.
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import datetime as dt
import json
import os
import socket
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo
import requests
TRUE_VALUES = {"1", "true", "yes", "y", "on"}
@dataclass(frozen=True)
class Config:
tempest_station_id: str
tempest_token: str
tempest_url: str
aprs_enabled: bool
aprs_callsign: str
aprs_passcode: str
aprs_server: str
aprs_port: int
aprs_lat: float
aprs_lon: float
aprs_comment: str
aprs_dest: str
aprs_path: str
aprs_min_interval_seconds: int
aprs_state_file: Path
aprs_rain_state_file: Path
aprs_pressure_offset_hpa: float
aprs_timeout_seconds: float
poll_interval_seconds: int
timezone_name: str
def env_str(name: str, default: str | None = None) -> str:
value = os.environ.get(name)
if value is None or value.strip() == "":
if default is None:
raise RuntimeError(f"Missing required environment variable: {name}")
return default
return value.strip()
def env_int(name: str, default: int) -> int:
value = os.environ.get(name)
if value is None or value.strip() == "":
return default
return int(value.strip())
def env_float(name: str, default: float) -> float:
value = os.environ.get(name)
if value is None or value.strip() == "":
return default
return float(value.strip())
def env_bool(name: str, default: bool = False) -> bool:
value = os.environ.get(name)
if value is None or value.strip() == "":
return default
return value.strip().lower() in TRUE_VALUES
def load_config() -> Config:
station_id = env_str("TEMPEST_STATION_ID")
token = env_str("TEMPEST_TOKEN")
tempest_url = (
f"https://swd.weatherflow.com/swd/rest/observations/station/"
f"{station_id}?token={token}"
)
return Config(
tempest_station_id=station_id,
tempest_token=token,
tempest_url=tempest_url,
aprs_enabled=env_bool("APRS_ENABLED", False),
aprs_callsign=env_str("APRS_CALLSIGN").upper(),
aprs_passcode=env_str("APRS_PASSCODE", ""),
aprs_server=env_str("APRS_SERVER", "cwop.aprs.net"),
aprs_port=env_int("APRS_PORT", 14580),
aprs_lat=env_float("APRS_LAT", 0.0),
aprs_lon=env_float("APRS_LON", 0.0),
aprs_comment=env_str("APRS_COMMENT", "Tempest with Custom Script"),
aprs_dest=env_str("APRS_DEST", "APRS"),
aprs_path=env_str("APRS_PATH", "TCPIP*"),
aprs_min_interval_seconds=env_int("APRS_MIN_INTERVAL_SECONDS", 300),
aprs_state_file=Path(env_str("APRS_STATE_FILE", "YOUR_STATE_DIRECTORY/tempest-aprs-last-sent")),
aprs_rain_state_file=Path(env_str("APRS_RAIN_STATE_FILE", "YOUR_STATE_DIRECTORY/tempest-rain-state.json")),
aprs_pressure_offset_hpa=env_float("APRS_PRESSURE_OFFSET_HPA", 0.0),
aprs_timeout_seconds=env_float("APRS_TIMEOUT_SECONDS", 10.0),
poll_interval_seconds=env_int("POLL_INTERVAL_SECONDS", 60),
timezone_name=env_str("TZ", "UTC"),
)
def log(message: str) -> None:
timestamp = dt.datetime.now().isoformat(timespec="seconds")
print(f"{timestamp} {message}", flush=True)
def get_float(mapping: dict[str, Any], *keys: str) -> float | None:
for key in keys:
if key not in mapping:
continue
value = mapping.get(key)
if value is None or value == "":
continue
try:
return float(value)
except (TypeError, ValueError):
continue
return None
def get_int(mapping: dict[str, Any], *keys: str) -> int | None:
value = get_float(mapping, *keys)
if value is None:
return None
return int(round(value))
def c_to_f(value_c: float | None) -> float | None:
if value_c is None:
return None
return (value_c * 9.0 / 5.0) + 32.0
def mps_to_mph(value_mps: float | None) -> float | None:
if value_mps is None:
return None
return value_mps * 2.2369362921
def mm_to_hundredths_in(value_mm: float | None) -> int | None:
if value_mm is None:
return None
return int(round(value_mm * 0.03937007874 * 100.0))
def fmt_3(value: float | int | None) -> str:
if value is None:
return "..."
try:
rounded = int(round(float(value)))
except (TypeError, ValueError):
return "..."
if rounded < 0:
return "..."
return f"{min(rounded, 999):03d}"
def fmt_temp_f(value_f: float | None) -> str:
if value_f is None:
return "..."
try:
rounded = int(round(float(value_f)))
except (TypeError, ValueError):
return "..."
if rounded < 0:
return f"-{min(abs(rounded), 99):02d}"
return f"{min(rounded, 999):03d}"
def fmt_humidity(value: float | int | None) -> str:
if value is None:
return ".."
try:
rounded = int(round(float(value)))
except (TypeError, ValueError):
return ".."
rounded = max(0, min(100, rounded))
# APRS convention: h00 means 100% relative humidity.
if rounded == 100:
return "00"
return f"{rounded:02d}"
def fmt_barometer_hpa(value_hpa: float | None) -> str:
if value_hpa is None:
return "....."
try:
tenths = int(round(float(value_hpa) * 10.0))
except (TypeError, ValueError):
return "....."
if tenths < 0:
return "....."
return f"{min(tenths, 99999):05d}"
def fmt_lat_aprs(decimal_degrees: float) -> str:
hemisphere = "N" if decimal_degrees >= 0 else "S"
absolute = abs(decimal_degrees)
degrees = int(absolute)
minutes = (absolute - degrees) * 60.0
return f"{degrees:02d}{minutes:05.2f}{hemisphere}"
def fmt_lon_aprs(decimal_degrees: float) -> str:
hemisphere = "E" if decimal_degrees >= 0 else "W"
absolute = abs(decimal_degrees)
degrees = int(absolute)
minutes = (absolute - degrees) * 60.0
return f"{degrees:03d}{minutes:05.2f}{hemisphere}"
def sanitize_comment(value: str) -> str:
return value.replace("\r", " ").replace("\n", " ").strip()[:40]
def load_json_file(path: Path, default: dict[str, Any]) -> dict[str, Any]:
try:
return json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError:
return default
except (OSError, json.JSONDecodeError):
return default
def write_json_file(path: Path, data: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
tmp_path = path.with_suffix(path.suffix + ".tmp")
tmp_path.write_text(json.dumps(data, indent=2, sort_keys=True), encoding="utf-8")
tmp_path.replace(path)
def fetch_tempest_observation(config: Config) -> dict[str, Any]:
log("Fetching Tempest data")
response = requests.get(config.tempest_url, timeout=30)
response.raise_for_status()
payload = response.json()
if "obs" not in payload or not payload["obs"]:
raise RuntimeError(f"Tempest response did not contain observations: {payload}")
obs = payload["obs"][0]
if not isinstance(obs, dict):
raise RuntimeError(f"Tempest observation was not an object: {obs}")
return obs
def observation_timestamp(obs: dict[str, Any]) -> float:
value = get_float(obs, "timestamp", "epoch", "time")
if value is None:
return time.time()
return value
def midnight_timestamp(local_timezone_name: str, timestamp: float) -> float:
timezone = ZoneInfo(local_timezone_name)
local = dt.datetime.fromtimestamp(timestamp, timezone)
midnight = local.replace(hour=0, minute=0, second=0, microsecond=0)
return midnight.timestamp()
def update_rain_state(obs: dict[str, Any], config: Config) -> dict[str, int | None]:
"""
The code accepts several possible Tempest field names because WeatherFlow
payloads differ between REST, WebSocket, and UDP-style structures.
APRS expects:
r = rain in the last hour, hundredths of an inch
p = rain in the last 24 hours, hundredths of an inch
P = rain since local midnight, hundredths of an inch
"""
now_ts = observation_timestamp(obs)
one_hour_ago = now_ts - 3600
one_day_ago = now_ts - 86400
midnight_ts = midnight_timestamp(config.timezone_name, now_ts)
state = load_json_file(config.aprs_rain_state_file, {"samples": []})
samples = state.get("samples", [])
if not isinstance(samples, list):
samples = []
# Possible per-observation / per-report rain increment fields, in mm.
rain_increment_mm = get_float(
obs,
"rain_last_minute_mm",
"rain_accumulated_mm",
"rain_accumulated",
"precip",
"precipitation",
"rain",
)
last_observation_timestamp = state.get("last_observation_timestamp")
already_seen = False
try:
already_seen = (
last_observation_timestamp is not None
and float(last_observation_timestamp) >= now_ts
)
except (TypeError, ValueError):
already_seen = False
if rain_increment_mm is not None and rain_increment_mm > 0 and not already_seen:
samples.append(
{
"timestamp": now_ts,
"rain_mm": rain_increment_mm,
}
)
pruned_samples: list[dict[str, float]] = []
for sample in samples:
try:
sample_ts = float(sample["timestamp"])
sample_rain_mm = float(sample["rain_mm"])
except (KeyError, TypeError, ValueError):
continue
if sample_ts >= one_day_ago:
pruned_samples.append(
{
"timestamp": sample_ts,
"rain_mm": sample_rain_mm,
}
)
rolling_1h_mm = sum(
sample["rain_mm"]
for sample in pruned_samples
if sample["timestamp"] >= one_hour_ago
)
rolling_24h_mm = sum(sample["rain_mm"] for sample in pruned_samples)
rolling_today_mm = sum(
sample["rain_mm"]
for sample in pruned_samples
if sample["timestamp"] >= midnight_ts
)
direct_1h_mm = get_float(
obs,
"precip_accum_last_1hr",
"precip_accum_last_1h",
"rain_last_hour_mm",
"rain_1h_mm",
)
direct_24h_mm = get_float(
obs,
"precip_accum_last_24hr",
"precip_accum_last_24h",
"rain_last_24h_mm",
"rain_24h_mm",
)
direct_today_mm = get_float(
obs,
"precip_accum_local_day",
"local_day_rain_accumulation",
"local_daily_rain_mm",
"local_daily_rain_final_mm",
"rain_today_mm",
"rain_since_midnight_mm",
)
rain_1h_mm = direct_1h_mm if direct_1h_mm is not None else rolling_1h_mm
rain_24h_mm = direct_24h_mm if direct_24h_mm is not None else rolling_24h_mm
rain_today_mm = direct_today_mm if direct_today_mm is not None else rolling_today_mm
state = {
"last_observation_timestamp": now_ts,
"samples": pruned_samples,
"updated_at": time.time(),
}
write_json_file(config.aprs_rain_state_file, state)
return {
"rain_1h_hundredths": mm_to_hundredths_in(rain_1h_mm),
"rain_24h_hundredths": mm_to_hundredths_in(rain_24h_mm),
"rain_today_hundredths": mm_to_hundredths_in(rain_today_mm),
}
def build_aprs_weather_frame(
obs: dict[str, Any],
rain_totals: dict[str, int | None],
config: Config,
) -> str:
temperature_c = get_float(
obs,
"air_temperature",
"air_temperature_c",
"temperature",
"temperature_c",
)
temperature_f = get_float(obs, "air_temperature_f", "temperature_f")
if temperature_f is None:
temperature_f = c_to_f(temperature_c)
humidity = get_float(obs, "relative_humidity", "humidity", "humidity_pct")
pressure_hpa = get_float(
obs,
"sea_level_pressure",
"sea_level_pressure_mb",
"sea_level_pressure_hpa",
"barometric_pressure",
"barometric_pressure_mb",
"pressure",
"pressure_mb",
)
if pressure_hpa is not None:
pressure_hpa += config.aprs_pressure_offset_hpa
wind_dir = get_int(obs, "wind_direction", "wind_direction_degrees", "wind_dir")
wind_avg_mps = get_float(obs, "wind_avg", "wind_avg_mps", "wind_speed_mps")
wind_gust_mps = get_float(obs, "wind_gust", "wind_gust_mps")
wind_avg_mph = get_float(obs, "wind_avg_mph", "wind_speed_mph")
if wind_avg_mph is None:
wind_avg_mph = mps_to_mph(wind_avg_mps)
wind_gust_mph = get_float(obs, "wind_gust_mph")
if wind_gust_mph is None:
wind_gust_mph = mps_to_mph(wind_gust_mps)
lat = fmt_lat_aprs(config.aprs_lat)
lon = fmt_lon_aprs(config.aprs_lon)
rain_1h = rain_totals.get("rain_1h_hundredths")
rain_24h = rain_totals.get("rain_24h_hundredths")
rain_today = rain_totals.get("rain_today_hundredths")
info = (
f"!{lat}/{lon}_"
f"{fmt_3(wind_dir)}/{fmt_3(wind_avg_mph)}"
f"g{fmt_3(wind_gust_mph)}"
f"t{fmt_temp_f(temperature_f)}"
f"r{fmt_3(rain_1h)}"
f"p{fmt_3(rain_24h)}"
f"P{fmt_3(rain_today)}"
f"h{fmt_humidity(humidity)}"
f"b{fmt_barometer_hpa(pressure_hpa)}"
f"{sanitize_comment(config.aprs_comment)}"
)
return f"{config.aprs_callsign}>{config.aprs_dest},{config.aprs_path}:{info}"
def last_aprs_sent_time(config: Config) -> float | None:
try:
return float(config.aprs_state_file.read_text(encoding="utf-8").strip())
except FileNotFoundError:
return None
except (OSError, ValueError):
return None
def write_last_aprs_sent_time(config: Config, timestamp: float) -> None:
config.aprs_state_file.parent.mkdir(parents=True, exist_ok=True)
config.aprs_state_file.write_text(str(timestamp), encoding="utf-8")
def should_send_now(config: Config) -> bool:
if not config.aprs_enabled:
return False
previous = last_aprs_sent_time(config)
if previous is None:
return True
return (time.time() - previous) >= config.aprs_min_interval_seconds
def send_aprs_frame(frame: str, config: Config) -> None:
if not config.aprs_passcode:
raise RuntimeError("APRS_ENABLED=true but APRS_PASSCODE is not set")
login_line = (
f"user {config.aprs_callsign} "
f"pass {config.aprs_passcode} "
f"vers TempestAPRS 1.0 "
f"filter m/1\r\n"
)
packet_line = f"{frame}\r\n"
with socket.create_connection(
(config.aprs_server, config.aprs_port),
timeout=config.aprs_timeout_seconds,
) as sock:
sock.settimeout(config.aprs_timeout_seconds)
sock.sendall(login_line.encode("ascii", errors="ignore"))
try:
response = sock.recv(512).decode("ascii", errors="replace").strip()
if response:
log(f"APRS-IS login response: {response}")
except socket.timeout:
log("APRS-IS login response timed out; continuing")
sock.sendall(packet_line.encode("ascii", errors="ignore"))
def run_once(config: Config, dry_run: bool = False, show_obs: bool = False) -> None:
obs = fetch_tempest_observation(config)
if show_obs:
print(json.dumps(obs, indent=2, sort_keys=True), flush=True)
rain_totals = update_rain_state(obs, config)
frame = build_aprs_weather_frame(obs, rain_totals, config)
if dry_run:
log("Dry run; not sending APRS packet")
print(frame, flush=True)
return
if not config.aprs_enabled:
log("APRS disabled; not sending APRS packet")
print(frame, flush=True)
return
if not should_send_now(config):
log("APRS upload skipped; rate limit window has not elapsed")
return
send_aprs_frame(frame, config)
write_last_aprs_sent_time(config, time.time())
log(f"APRS weather packet sent: {frame}")
def run_loop(config: Config) -> None:
log("Tempest APRS service started")
log(
"Configuration: "
f"callsign={config.aprs_callsign} "
f"server={config.aprs_server}:{config.aprs_port} "
f"poll={config.poll_interval_seconds}s "
f"aprs_interval={config.aprs_min_interval_seconds}s "
f"enabled={config.aprs_enabled}"
)
while True:
try:
run_once(config)
except Exception as exc:
log(f"Run failed: {exc}")
time.sleep(config.poll_interval_seconds)
def main() -> int:
parser = argparse.ArgumentParser(description="Publish WeatherFlow Tempest data to APRS/CWOP.")
parser.add_argument("--once", action="store_true", help="Run one collection pass and exit.")
parser.add_argument("--loop", action="store_true", help="Run forever.")
parser.add_argument("--dry-run", action="store_true", help="Fetch Tempest data and print an APRS frame without sending.")
parser.add_argument("--show-obs", action="store_true", help="Print the raw Tempest observation JSON.")
args = parser.parse_args()
config = load_config()
if args.dry_run:
run_once(config, dry_run=True, show_obs=args.show_obs)
return 0
if args.once:
run_once(config, dry_run=False, show_obs=args.show_obs)
return 0
run_loop(config)
return 0
if __name__ == "__main__":
raise SystemExit(main())I was exceptionally excited when the system returned the proper content and posted successfully to APRS!
