vidane/weather-or-not
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: weather-or-not scaffold

Browse Source
This commit is contained in:
vidane
2026-04-29 04:46:01 +00:00
commit 86e6499895
13 changed files with 1353 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
.gitignore vendored Normal file
View File

@@ -0,0 +1,21 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# Virtual environments
.venv/
venv/
# IDE
.idea/
.vscode/
*.swp
# Data
.history.json
# Distribution
dist/
build/
*.egg-info/

89
README.md Normal file
View File

@@ -0,0 +1,89 @@
# 🌤️ Weather or Not
A weather-based activity decision engine that tells you whether to ride your motorcycle or mow the lawn today.
## Features
- **Motorcycle ride check** — evaluates temperature, precipitation, wind, humidity, and road conditions
- **Lawn mowing check** — evaluates weather conditions AND tracks when you last mowed (with configurable cooldown)
- **Extensible** — add new activities by implementing `BaseActivity`
- **CLI tool** — run from terminal or schedule via cron
- **Notifications** — sends results via Mattermost or Hermes
## Quick Start
```bash
# Install
pip install -e .
# Run a check (uses config.yaml for location)
weather-or-not
# With verbose output
weather-or-not -v
# Override location
weather-or-not --lat 48.8566 --lon 2.3522 # Paris
# Record that you performed an activity
weather-or-not record mow_lawn
weather-or-not record motorcycle_ride
# Send notification
weather-or-not notify send
```
## Configuration
Copy `config.yaml` and customize:
```yaml
location:
latitude: 40.7128 # Your latitude
longitude: -74.0060 # Your longitude
activities:
motorcycle_ride:
enabled: true
min_temp_c: 10
max_temp_c: 35
max_precipitation_mm: 0
max_wind_speed_kmh: 30
mow_lawn:
enabled: true
cooldown_days: 3
max_precipitation_today_mm: 0
max_precipitation_past_24h_mm: 5
```
## Adding New Activities
Create a new file in `src/weather_or_not/` that extends `BaseActivity`:
```python
from weather_or_not.activities import BaseActivity, RuleResult, Verdict
from weather_or_not.weather import WeatherForecast
class MyActivity(BaseActivity):
name = "my_activity"
description = "Check if it's good for my activity"
async def check_rules(self, weather: WeatherForecast) -> list[RuleResult]:
return [
RuleResult(
rule_name="My Rule",
passed=weather.current_temp_c > 15,
message="Temperature is above 15°C",
),
]
```
Then register it in `cli.py`'s `run_check()` function.
## Data Storage
Activity history is stored in `~/.local/share/weather-or-not/history.json`.
## License
MIT

43
config.yaml Normal file
View File

@@ -0,0 +1,43 @@
# Weather or Not - Configuration
# Location for weather data (Open-Meteo)
location:
latitude: 40.7128 # Change to your latitude
longitude: -74.0060 # Change to your longitude
label: "Home" # Display name
# Activity thresholds
activities:
motorcycle_ride:
enabled: true
min_temp_c: 10
max_temp_c: 35
max_precipitation_mm: 0 # No rain allowed
max_wind_speed_kmh: 30
max_humidity_pct: 90
no_ice: true # No freezing conditions
no_wet_roads: true # No rain or snow in past 2 hours
notes: "Requires dry roads, no precipitation, comfortable temperature"
mow_lawn:
enabled: true
# Cooldown: minimum days between mowing suggestions
cooldown_days: 3
max_precipitation_today_mm: 0 # No rain expected today
max_precipitation_past_24h_mm: 5 # Ground should be mostly dry
max_wind_speed_kmh: 25
min_temp_c: 15
notes: "Checks that you haven't mowed too recently and ground is dry"
# Notification settings
notifications:
mattermost:
enabled: true
webhook_url: "" # Your Mattermost incoming webhook URL
channel: "#general" # Or your preferred channel
# For Mattermost home channel delivery via Hermes, use the hermes notify endpoint
hermes_notify: true # Set to true to send via Hermes Mattermost integration
# Output format
output:
verbose: true # Show detailed reasoning
colorize: true # Use colors in terminal output (requires colorama)

40
pyproject.toml Normal file
View File

@@ -0,0 +1,40 @@
[build-system]
requires = ["setuptools>=68.0", "wheel"]
build-backend = "setuptools.backends._legacy:_Backend"
[project]
name = "weather-or-not"
version = "0.1.0"
description = "Weather-based activity decision engine"
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.11"
dependencies = [
"httpx>=0.27",
"pydantic>=2.0",
"pydantic-settings>=2.0",
"pyyaml>=6.0",
"python-dateutil>=2.9",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.23",
"pytest-httpx>=0.30",
]
[project.scripts]
weather-or-not = "weather_or_not.cli:main"
[tool.setuptools.packages.find]
where = ["."]
namespaces = false
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
[tool.ruff]
target-version = "py311"
line-length = 100

3
src/weather_or_not/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""Weather or Not - Weather-based activity decision engine."""
__version__ = "0.1.0"

221
src/weather_or_not/activities.py Normal file
View File

@@ -0,0 +1,221 @@
"""Base activity rule engine and registry."""
from __future__ import annotations
import asyncio
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, date
from enum import Enum
from pathlib import Path
from typing import Any
from .weather import WeatherForecast
class Verdict(str, Enum):
GO = "GO"
NO_GO = "NO-GO"
CAUTION = "CAUTION"
@dataclass
class RuleResult:
"""Result of a single rule check."""
rule_name: str
passed: bool
message: str
@dataclass
class ActivityResult:
"""Result of an activity evaluation."""
activity_name: str
verdict: Verdict
rules: list[RuleResult] = field(default_factory=list)
reasoning: list[str] = field(default_factory=list)
extra_info: dict[str, Any] = field(default_factory=dict)
@property
def is_go(self) -> bool:
return self.verdict == Verdict.GO
@property
def is_no_go(self) -> bool:
return self.verdict == Verdict.NO_GO
def format(self, verbose: bool = True, colorize: bool = True) -> str:
"""Format the result for display."""
emoji = ""
if colorize:
if self.verdict == Verdict.GO:
emoji = "✅"
elif self.verdict == Verdict.NO_GO:
emoji = "❌"
else:
emoji = "⚠️"
else:
if self.verdict == Verdict.GO:
emoji = "[GO]"
elif self.verdict == Verdict.NO_GO:
emoji = "[NO-GO]"
else:
emoji = "[CAUTION]"
lines = [f" {emoji} {self.activity_name}: {self.verdict.value}"]
if verbose:
for rule in self.rules:
status = "✓" if rule.passed else "✗"
lines.append(f" {status} {rule.rule_name}: {rule.message}")
for reason in self.reasoning:
lines.append(f" • {reason}")
return "\n".join(lines)
class BaseActivity(ABC):
"""Base class for activity evaluators."""
name: str = "base"
description: str = ""
def __init__(self, config: dict[str, Any], data_store: DataStore):
self.config = config
self.data_store = data_store
async def evaluate(self, weather: WeatherForecast) -> ActivityResult:
"""Evaluate whether the activity should be done today."""
rules = await self.check_rules(weather)
verdict = self._determine_verdict(rules)
reasoning = self._build_reasoning(rules)
# Update last check time
self.data_store.record_activity_check(self.name, weather)
return ActivityResult(
activity_name=self.name,
verdict=verdict,
rules=rules,
reasoning=reasoning,
)
@abstractmethod
async def check_rules(self, weather: WeatherForecast) -> list[RuleResult]:
"""Check all rules for this activity."""
...
def _determine_verdict(self, rules: list[RuleResult]) -> Verdict:
"""Determine verdict based on rule results."""
failed = [r for r in rules if not r.passed]
if not failed:
return Verdict.GO
# If any critical rule failed, it's a hard NO-GO
critical_failures = [
r for r in failed
if "no rain" in r.rule_name.lower() or "precip" in r.rule_name.lower()
]
if critical_failures:
return Verdict.NO_GO
return Verdict.CAUTION
def _build_reasoning(self, rules: list[RuleResult]) -> list[str]:
"""Build human-readable reasoning."""
reasons = []
for rule in rules:
if not rule.passed:
reasons.append(f"{rule.rule_name}: {rule.message}")
return reasons
class DataStore:
"""Persistent storage for activity history (JSON file-based)."""
def __init__(self, data_dir: Path | None = None):
if data_dir is None:
data_dir = Path.home() / ".local" / "share" / "weather-or-not"
self.data_dir = data_dir
self.data_dir.mkdir(parents=True, exist_ok=True)
self._data_file = self.data_dir / "history.json"
self._data = self._load()
def _load(self) -> dict:
if self._data_file.exists():
import json
try:
with open(self._data_file) as f:
return json.load(f)
except (json.JSONDecodeError, IOError):
return {}
return {}
def _save(self):
import json
with open(self._data_file, "w") as f:
json.dump(self._data, f, indent=2)
def record_activity_check(self, activity_name: str, weather: WeatherForecast):
"""Record that an activity was checked at the current time."""
from datetime import datetime, timezone
now = datetime.now(timezone.utc).isoformat()
if "checks" not in self._data:
self._data["checks"] = {}
if activity_name not in self._data["checks"]:
self._data["checks"][activity_name] = []
self._data["checks"][activity_name].append({
"timestamp": now,
"latitude": weather.latitude,
"longitude": weather.longitude,
"temp_c": weather.current_temp_c,
})
self._save()
def last_activity_date(self, activity_name: str) -> date | None:
"""Get the last date an activity was performed (recorded by user)."""
if "performed" not in self._data:
self._data["performed"] = {}
records = self._data["performed"].get(activity_name, [])
if not records:
return None
# Parse dates and return the most recent
from datetime import datetime
dates = []
for record in records:
try:
d = date.fromisoformat(str(record["date"]))
dates.append(d)
except (ValueError, TypeError):
continue
return max(dates) if dates else None
def record_activity_performed(self, activity_name: str, activity_date: date | None = None):
"""Record that an activity was actually performed."""
if activity_date is None:
activity_date = date.today()
if "performed" not in self._data:
self._data["performed"] = {}
if activity_name not in self._data["performed"]:
self._data["performed"][activity_name] = []
self._data["performed"][activity_name].append({
"date": activity_date.isoformat(),
"timestamp": datetime.now().isoformat(),
})
self._save()
def days_since_activity(self, activity_name: str) -> int:
"""Calculate days since an activity was last performed."""
last = self.last_activity_date(activity_name)
if last is None:
return 999 # Never done
return (date.today() - last).days
def get_activity_stats(self, activity_name: str) -> dict:
"""Get statistics for an activity."""
performed = self._data.get("performed", {}).get(activity_name, [])
checks = self._data.get("checks", {}).get(activity_name, [])
return {
"times_performed": len(performed),
"times_checked": len(checks),
"last_performed": performed[-1]["date"] if performed else None,
"last_checked": checks[-1]["timestamp"] if checks else None,
}

171
src/weather_or_not/cli.py Normal file
View File

@@ -0,0 +1,171 @@
"""CLI entry point for Weather or Not."""
from __future__ import annotations
import argparse
import asyncio
import sys
from datetime import date
from pathlib import Path
from typing import Any
import yaml
from .weather import fetch_weather
from .activities import DataStore
from .motorcycle import MotorcycleActivity
from .lawn import LawnActivity
from .notify import NotificationService, NotificationConfig
def load_config(config_path: str = "config.yaml") -> dict[str, Any]:
"""Load configuration from YAML file."""
path = Path(config_path)
if not path.exists():
# Try home directory
path = Path.home() / ".config" / "weather-or-not" / "config.yaml"
if not path.exists():
print(f"Warning: No config file found at {config_path}")
print("Using defaults. Create a config.yaml to customize.")
return {}
with open(path) as f:
return yaml.safe_load(f) or {}
async def run_check(
latitude: float,
longitude: float,
config: dict[str, Any],
data_store: DataStore,
verbose: bool = True,
colorize: bool = True,
) -> list[str]:
"""Run a full weather check and return formatted results."""
# Fetch weather
print("Fetching weather data...")
weather = await fetch_weather(latitude, longitude)
print(f" Got data ({weather.generationtime_ms:.1f}ms generation time)")
# Build activity evaluators
activities_config = config.get("activities", {})
activities: list[Any] = []
# Motorcycle
if activities_config.get("motorcycle_ride", {}).get("enabled", True):
mc_config = activities_config.get("motorcycle_ride", {})
activities.append(MotorcycleActivity(mc_config, data_store))
# Lawn
if activities_config.get("mow_lawn", {}).get("enabled", True):
lawn_config = activities_config.get("mow_lawn", {})
activities.append(LawnActivity(lawn_config, data_store))
# Evaluate all activities
results = []
for activity in activities:
print(f"Evaluating {activity.name}...")
result = await activity.evaluate(weather)
results.append(result)
return results
def main():
"""Main CLI entry point."""
parser = argparse.ArgumentParser(
description="Weather or Not — Should you ride or mow?",
)
parser.add_argument(
"-c", "--config",
default="config.yaml",
help="Path to config file (default: config.yaml)",
)
parser.add_argument(
"--verbose", "-v",
action="store_true",
help="Show detailed reasoning",
)
parser.add_argument(
"--no-color",
action="store_true",
help="Disable colored output",
)
parser.add_argument(
"notify",
nargs="?",
choices=["send"],
help="Send notification (default: just display)",
)
parser.add_argument(
"--lat",
type=float,
help="Override latitude from config",
)
parser.add_argument(
"--lon",
type=float,
help="Override longitude from config",
)
parser.add_argument(
"record",
nargs="*",
help="Record activity as performed: motorcycle_ride, mow_lawn",
)
args = parser.parse_args()
# Load config
config = load_config(args.config)
# Get location
location = config.get("location", {})
lat = args.lat or location.get("latitude", 40.7128)
lon = args.lon or location.get("longitude", -74.0060)
# Output settings
output_config = config.get("output", {})
verbose = args.verbose or output_config.get("verbose", True)
colorize = not args.no_color and output_config.get("colorize", True)
# Data store
data_store = DataStore()
# Handle record command
if args.record:
for activity_name in args.record:
if activity_name in ("motorcycle_ride", "mow_lawn"):
data_store.record_activity_performed(activity_name)
print(f"✓ Recorded: {activity_name} performed today ({date.today()})")
else:
print(f"Unknown activity: {activity_name}")
print(f"Available: motorcycle_ride, mow_lawn")
sys.exit(1)
return
# Run check
try:
results = asyncio.run(
run_check(lat, lon, config, data_store, verbose, colorize)
)
except Exception as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
# Display results
notify_config = NotificationService(
NotificationConfig(**config.get("notifications", {}))
)
print(notify_config.format_for_terminal(results))
# Send notification if requested
if args.notify == "send":
asyncio.run(
notify_config.send(results)
)
print("\nNotification sent.")
if __name__ == "__main__":
main()

144
src/weather_or_not/lawn.py Normal file
View File

@@ -0,0 +1,144 @@
"""Lawn mowing activity evaluator."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import date
from typing import Any
from .activities import (
BaseActivity,
DataStore,
RuleResult,
Verdict,
)
from .weather import WeatherForecast, weather_code_description
class LawnActivity(BaseActivity):
"""Evaluate whether it's a good day to mow the lawn."""
name = "mow_lawn"
description = "Check if conditions are good for mowing the lawn"
async def check_rules(self, weather: WeatherForecast) -> list[RuleResult]:
rules: list[RuleResult] = []
activity_config = self.config
# 1. Cooldown check — have we mowed recently?
cooldown_days = activity_config.get("cooldown_days", 3)
days_since = self.data_store.days_since_activity("mow_lawn")
if days_since < cooldown_days:
rules.append(RuleResult(
rule_name="Cooldown period",
passed=False,
message=(
f"Last mowed {days_since} day(s) ago. "
f"Wait {cooldown_days - days_since} more day(s) "
f"(recommended minimum: {cooldown_days} days between mows)"
),
))
else:
rules.append(RuleResult(
rule_name="Cooldown period",
passed=True,
message=f"Good — last mowed {days_since} days ago (minimum: {cooldown_days})",
))
# 2. Today's precipitation
precip_today = weather.precip_today_mm()
max_precip = activity_config.get("max_precipitation_today_mm", 0)
if precip_today > max_precip:
rules.append(RuleResult(
rule_name="Rain today",
passed=False,
message=f"{precip_today}mm rain expected today (max: {max_precip}mm)",
))
else:
rules.append(RuleResult(
rule_name="Rain today",
passed=True,
message="No rain expected today",
))
# 3. Past 24h precipitation (ground dryness)
precip_24h = weather.precip_past_24h_mm()
max_24h = activity_config.get("max_precipitation_past_24h_mm", 5)
if precip_24h > max_24h:
rules.append(RuleResult(
rule_name="Ground moisture",
passed=False,
message=(
f"{precip_24h}mm in past 24h — ground likely too wet "
f"(max: {max_24h}mm)"
),
))
else:
rules.append(RuleResult(
rule_name="Ground moisture",
passed=True,
message=f"Ground is dry ({precip_24h}mm in past 24h)",
))
# 4. Wind speed
wind = weather.current_wind_speed
max_wind = activity_config.get("max_wind_speed_kmh", 25)
if wind > max_wind:
rules.append(RuleResult(
rule_name="Wind speed",
passed=False,
message=f"{wind} km/h exceeds safe limit of {max_wind} km/h",
))
else:
rules.append(RuleResult(
rule_name="Wind speed",
passed=True,
message=f"{wind} km/h is safe for mowing",
))
# 5. Temperature
temp = weather.current_temp_c
min_temp = activity_config.get("min_temp_c", 15)
if temp < min_temp:
rules.append(RuleResult(
rule_name="Temperature",
passed=False,
message=f"{temp}°C is below minimum {min_temp}°C for comfortable mowing",
))
else:
rules.append(RuleResult(
rule_name="Temperature",
passed=True,
message=f"{temp}°C is comfortable",
))
# 6. Weather code
code = weather.current_weather_code
if code in {0, 1, 2}: # Clear, mainly clear, partly cloudy
rules.append(RuleResult(
rule_name="Weather conditions",
passed=True,
message=weather_code_description(code),
))
else:
rules.append(RuleResult(
rule_name="Weather conditions",
passed=False,
message=weather_code_description(code),
))
return rules
def format_result(self, result) -> str:
"""Override to add extra lawn-specific info."""
lines = [result.format(verbose=True, colorize=True)]
stats = self.data_store.get_activity_stats("mow_lawn")
lines.append("")
lines.append(f" 📊 Stats: performed {stats['times_performed']} times, "
f"checked {stats['times_checked']} times")
if stats["last_performed"]:
lines.append(f" 📅 Last mowed: {stats['last_performed']}")
return "\n".join(lines)

152
src/weather_or_not/motorcycle.py Normal file
View File

@@ -0,0 +1,152 @@
"""Motorcycle ride activity evaluator."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from .activities import (
BaseActivity,
DataStore,
RuleResult,
Verdict,
)
from .weather import WeatherForecast, weather_code_description
@dataclass
class MotorcycleConfig:
"""Configuration for motorcycle ride evaluation."""
enabled: bool = True
min_temp_c: float = 10
max_temp_c: float = 35
max_precipitation_mm: float = 0
max_wind_speed_kmh: float = 30
max_humidity_pct: float = 90
no_ice: bool = True
no_wet_roads: bool = True
class MotorcycleActivity(BaseActivity):
"""Evaluate whether it's safe/pleasant to ride a motorcycle."""
name = "motorcycle_ride"
description = "Check if conditions are good for motorcycle commuting"
async def check_rules(self, weather: WeatherForecast) -> list[RuleResult]:
rules: list[RuleResult] = []
# 1. Temperature check
temp = weather.current_temp_c
if temp < self.config.get("min_temp_c", 10):
rules.append(RuleResult(
rule_name="Temperature too low",
passed=False,
message=f"{temp}°C is below minimum {self.config.get('min_temp_c', 10)}°C",
))
elif temp > self.config.get("max_temp_c", 35):
rules.append(RuleResult(
rule_name="Temperature too high",
passed=False,
message=f"{temp}°C exceeds maximum {self.config.get('max_temp_c', 35)}°C",
))
else:
rules.append(RuleResult(
rule_name="Temperature",
passed=True,
message=f"{temp}°C is comfortable for riding",
))
# 2. Precipitation check
precip_today = weather.precip_today_mm()
if precip_today > self.config.get("max_precipitation_mm", 0):
rules.append(RuleResult(
rule_name="Precipitation expected",
passed=False,
message=f"{precip_today}mm rain expected today (max: {self.config.get('max_precipitation_mm', 0)}mm)",
))
else:
rules.append(RuleResult(
rule_name="Precipitation",
passed=True,
message="No rain expected",
))
# 3. Wind speed check
wind = weather.current_wind_speed
max_wind = self.config.get("max_wind_speed_kmh", 30)
if wind > max_wind:
rules.append(RuleResult(
rule_name="Wind too strong",
passed=False,
message=f"{wind} km/h exceeds safe limit of {max_wind} km/h",
))
else:
rules.append(RuleResult(
rule_name="Wind speed",
passed=True,
message=f"{wind} km/h is safe for riding",
))
# 4. Humidity check
humidity = weather.current_humidity
max_hum = self.config.get("max_humidity_pct", 90)
if humidity > max_hum:
rules.append(RuleResult(
rule_name="Humidity too high",
passed=False,
message=f"{humidity}% humidity (max: {max_hum}%)",
))
else:
rules.append(RuleResult(
rule_name="Humidity",
passed=True,
message=f"{humidity}% is comfortable",
))
# 5. Weather code check (snow, ice, thunderstorm)
code = weather.current_weather_code
dangerous_codes = {45, 48, 66, 67, 71, 73, 75, 77, 85, 86, 95, 96, 99}
if code in dangerous_codes:
rules.append(RuleResult(
rule_name="Dangerous weather",
passed=False,
message=weather_code_description(code),
))
else:
rules.append(RuleResult(
rule_name="Weather conditions",
passed=True,
message=weather_code_description(code),
))
# 6. Road conditions (wet roads in past 2 hours)
if self.config.get("no_wet_roads", True):
precip_2h = weather.precip_next_2h_mm()
if precip_2h > 0:
rules.append(RuleResult(
rule_name="Road conditions",
passed=False,
message=f"Rain expected in next 2h ({precip_2h}mm) — wet roads",
))
else:
rules.append(RuleResult(
rule_name="Road conditions",
passed=True,
message="Roads expected to stay dry",
))
return rules
def get_summary(self, weather: WeatherForecast) -> str:
"""Get a quick summary of current conditions."""
temp = weather.current_temp_c
wind = weather.current_wind_speed
precip = weather.precip_today_mm()
desc = weather_code_description(weather.current_weather_code)
return (
f"Current: {temp}°C, {desc}, "
f"Wind: {wind} km/h, "
f"Rain today: {precip}mm"
)

105
src/weather_or_not/notify.py Normal file
View File

@@ -0,0 +1,105 @@
"""Notification handlers for Weather or Not."""
from __future__ import annotations
import httpx
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from .activities import ActivityResult, Verdict
@dataclass
class NotificationConfig:
"""Notification configuration."""
mattermost_enabled: bool = True
mattermost_webhook_url: str = ""
mattermost_channel: str = "#general"
hermes_notify: bool = True # Use Hermes Mattermost integration
class NotificationService:
"""Send notifications via Mattermost or Hermes."""
def __init__(self, config: NotificationConfig | None = None):
self.config = config or NotificationConfig()
async def send(self, results: list[ActivityResult]) -> None:
"""Send notification for all activity results."""
if self.config.hermes_notify:
# Hermes integration: write to a file that the gateway
# can pick up and send via Mattermost
await self._notify_hermes(results)
elif self.config.mattermost_enabled and self.config.mattermost_webhook_url:
await self._send_mattermost_webhook(results)
async def _notify_hermes(self, results: list[ActivityResult]) -> None:
"""Send via Hermes notification system."""
# Write notification to a temp file that Hermes can process
# This is the simplest integration with Hermes' Mattermost
message = self._format_message(results)
# Write to a well-known location for Hermes to pick up
output_dir = Path.home() / ".hermes" / "cron" / "output"
output_dir.mkdir(parents=True, exist_ok=True)
output_file = output_dir / "weather-or-not-latest.md"
with open(output_file, "w") as f:
f.write(message)
def _format_message(self, results: list[ActivityResult]) -> str:
"""Format results into a human-readable message."""
lines = ["## 🌤️ Weather or Not — Daily Check"]
lines.append("")
for result in results:
if result.verdict == Verdict.GO:
emoji = "✅"
elif result.verdict == Verdict.NO_GO:
emoji = "❌"
else:
emoji = "⚠️"
lines.append(f"{emoji} **{result.activity_name}**: {result.verdict.value}")
if result.reasoning:
lines.append("")
lines.append("*Reasoning:*")
for reason in result.reasoning:
lines.append(f" • {reason}")
lines.append("")
lines.append("---")
lines.append("*Generated by Weather or Not*")
return "\n".join(lines)
async def _send_mattermost_webhook(self, results: list[ActivityResult]) -> None:
"""Send a webhook to Mattermost."""
message = self._format_message(results)
payload = {
"channel": self.config.mattermost_channel,
"text": message,
"username": "Weather or Not",
"icon_emoji": ":partly_sunny:",
}
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.post(
self.config.mattermost_webhook_url,
json=payload,
)
resp.raise_for_status()
def format_for_terminal(self, results: list[ActivityResult]) -> str:
"""Format results for terminal display."""
lines = ["\n🌤️ **Weather or Not — Daily Check**\n"]
for result in results:
lines.append(result.format(verbose=True, colorize=True))
lines.append("")
return "\n".join(lines)

180
src/weather_or_not/weather.py Normal file
View File

@@ -0,0 +1,180 @@
"""Weather data fetching from Open-Meteo API."""
from __future__ import annotations
import httpx
from datetime import datetime, timezone
from pydantic import BaseModel, Field
from typing import Optional
class WeatherForecast(BaseModel):
"""Parsed weather forecast data."""
latitude: float
longitude: float
generationtime_ms: float
utc_offset_seconds: int
timezone: str
timezone_abbreviation: str
elevation: float
current: dict
hourly: dict
daily: dict
@property
def current_temp_c(self) -> float:
return self.current.get("temperature_4m", 0)
@property
def current_wind_speed(self) -> float:
return self.current.get("wind_speed_10m", 0)
@property
def current_humidity(self) -> float:
return self.current.get("relative_humidity_2m", 100)
@property
def current_weather_code(self) -> int:
return self.current.get("weather_code", 0)
def precip_today_mm(self) -> float:
"""Total precipitation expected today."""
daily = self.daily
if "precipitation_sum" in daily:
precip_values = daily["precipitation_sum"]
if precip_values and len(precip_values) > 0:
return float(precip_values[0])
return 0.0
def precip_past_24h_mm(self) -> float:
"""Total precipitation in the past 24 hours (from hourly data)."""
hourly = self.hourly
if "precipitation" in hourly:
precip_values = hourly["precipitation"]
times = hourly.get("time", [])
now = datetime.now(timezone.utc)
total = 0.0
for i, (time_str, precip) in enumerate(zip(times, precip_values)):
try:
t = datetime.fromisoformat(time_str.replace("Z", "+00:00"))
if (now - t).total_seconds() <= 86400:
total += float(precip) if precip else 0
except (ValueError, TypeError):
continue
return total
return 0.0
def precip_next_2h_mm(self) -> float:
"""Precipitation expected in the next 2 hours."""
hourly = self.hourly
if "precipitation" in hourly:
precip_values = hourly["precipitation"]
times = hourly.get("time", [])
now = datetime.now(timezone.utc)
total = 0.0
for time_str, precip in zip(times, precip_values):
try:
t = datetime.fromisoformat(time_str.replace("Z", "+00:00"))
diff = (t - now).total_seconds()
if 0 <= diff <= 7200: # Next 2 hours
total += float(precip) if precip else 0
except (ValueError, TypeError):
continue
return total
return 0.0
def wind_speed_next_24h(self) -> list[float]:
"""Wind speeds for the next 24 hours."""
hourly = self.hourly
if "wind_speed_10m" in hourly:
return [float(w) for w in hourly["wind_speed_10m"][:24]]
return []
def temp_next_24h(self) -> list[float]:
"""Temperatures for the next 24 hours."""
hourly = self.hourly
if "temperature_2m" in hourly:
return [float(t) for t in hourly["temperature_2m"][:24]]
return []
async def fetch_weather(
latitude: float,
longitude: float,
*,
include_current: bool = True,
include_hourly: bool = True,
include_daily: bool = True,
) -> WeatherForecast:
"""Fetch weather forecast from Open-Meteo API.
Open-Meteo is free, no API key required.
"""
params: dict[str, str] = {}
if include_current:
params["current"] = (
"temperature_2m,relative_humidity_2m,precipitation,"
"weather_code,wind_speed_10m,wind_direction_10m,"
"is_day,pressure_msl"
)
if include_hourly:
params["hourly"] = (
"temperature_2m,relative_humidity_2m,precipitation,"
"wind_speed_10m,wind_direction_10m,weather_code,"
"apparent_temperature"
)
if include_daily:
params["daily"] = "precipitation_sum,temperature_2m_max,temperature_2m_min"
url = "https://api.open-meteo.com/v1/forecast"
params.update({
"latitude": str(latitude),
"longitude": str(longitude),
"timezone": "auto",
"forecast_days": 3,
})
async with httpx.AsyncClient(timeout=15.0) as client:
resp = await client.get(url, params=params)
resp.raise_for_status()
data = resp.json()
return WeatherForecast(**data)
def weather_code_description(code: int) -> str:
"""Human-readable description of a WMO weather code."""
codes = {
0: "Clear sky",
1: "Mainly clear",
2: "Partly cloudy",
3: "Overcast",
45: "Fog",
48: "Depositing rime fog",
51: "Light drizzle",
53: "Moderate drizzle",
55: "Dense drizzle",
56: "Light freezing drizzle",
57: "Dense freezing drizzle",
61: "Slight rain",
63: "Moderate rain",
65: "Heavy rain",
66: "Light freezing rain",
67: "Heavy freezing rain",
71: "Slight snowfall",
73: "Moderate snowfall",
75: "Heavy snowfall",
77: "Snow grains",
80: "Slight rain showers",
81: "Moderate rain showers",
82: "Violent rain showers",
85: "Slight snow showers",
86: "Heavy snow showers",
95: "Thunderstorm",
96: "Thunderstorm with slight hail",
99: "Thunderstorm with heavy hail",
}
return codes.get(code, f"Unknown ({code})")

0
tests/__init__.py Normal file
View File

184
tests/test_activities.py Normal file
View File

@@ -0,0 +1,184 @@
"""Tests for activity evaluators."""
from __future__ import annotations
import pytest
from datetime import date, timedelta
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
from weather_or_not.activities import DataStore, Verdict
from weather_or_not.motorcycle import MotorcycleActivity
from weather_or_not.lawn import LawnActivity
from weather_or_not.weather import WeatherForecast
def _make_weather(
temp: float = 20,
wind: float = 10,
humidity: float = 50,
weather_code: int = 0,
precip_today: float = 0,
precip_24h: float = 0,
precip_2h: float = 0,
) -> WeatherForecast:
"""Create a minimal WeatherForecast for testing."""
return WeatherForecast(
latitude=40.7128,
longitude=-74.0060,
generationtime_ms=10.0,
utc_offset_seconds=-14400,
timezone="America/New_York",
timezone_abbreviation="EDT",
elevation=10.0,
current={
"temperature_4m": temp,
"wind_speed_10m": wind,
"relative_humidity_2m": humidity,
"weather_code": weather_code,
"precipitation": 0,
"is_day": 1,
"pressure_msl": 1013,
},
hourly={
"time": [
"2026-04-29T10:00:00",
"2026-04-29T11:00:00",
"2026-04-29T12:00:00",
],
"temperature_2m": [temp, temp + 1, temp + 2],
"relative_humidity_2m": [humidity, humidity - 5, humidity - 10],
"precipitation": [0, 0, 0],
"wind_speed_10m": [wind, wind - 1, wind - 2],
"weather_code": [0, 0, 0],
"apparent_temperature": [temp, temp + 1, temp + 2],
},
daily={
"time": ["2026-04-29"],
"precipitation_sum": [precip_today],
"temperature_2m_max": [temp + 5],
"temperature_2m_min": [temp - 5],
},
)
class TestDataStore:
def test_never_done(self, tmp_path):
store = DataStore(tmp_path)
assert store.days_since_activity("mow_lawn") == 999
def test_record_and_check(self, tmp_path):
store = DataStore(tmp_path)
store.record_activity_performed("mow_lawn", date(2026, 4, 25))
assert store.days_since_activity("mow_lawn") == 4 # Apr 29 - Apr 25
def test_stats(self, tmp_path):
store = DataStore(tmp_path)
store.record_activity_performed("mow_lawn", date(2026, 4, 25))
stats = store.get_activity_stats("mow_lawn")
assert stats["times_performed"] == 1
assert stats["last_performed"] == "2026-04-25"
def test_persistence(self, tmp_path):
"""Data persists across DataStore instances."""
store1 = DataStore(tmp_path)
store1.record_activity_performed("mow_lawn", date(2026, 4, 20))
del store1
store2 = DataStore(tmp_path)
assert store2.last_activity_date("mow_lawn") == date(2026, 4, 20)
class TestMotorcycleActivity:
def setup_method(self):
self.config = {
"min_temp_c": 10,
"max_temp_c": 35,
"max_precipitation_mm": 0,
"max_wind_speed_kmh": 30,
"max_humidity_pct": 90,
"no_ice": True,
"no_wet_roads": True,
}
self.data_store = DataStore(Path("/tmp/test-won-data"))
self.activity = MotorcycleActivity(self.config, self.data_store)
@pytest.mark.asyncio
async def test_good_conditions(self):
weather = _make_weather(temp=22, wind=10, humidity=50, weather_code=0)
result = await self.activity.evaluate(weather)
assert result.verdict == Verdict.GO
assert all(r.passed for r in result.rules)
@pytest.mark.asyncio
async def test_rain(self):
weather = _make_weather(precip_today=5)
result = await self.activity.evaluate(weather)
assert result.verdict == Verdict.NO_GO
assert any(not r.passed and "precip" in r.rule_name.lower()
for r in result.rules)
@pytest.mark.asyncio
async def test_wind_too_strong(self):
weather = _make_weather(wind=40)
result = await self.activity.evaluate(weather)
assert result.verdict == Verdict.NO_GO
assert any(not r.passed and "wind" in r.rule_name.lower()
for r in result.rules)
@pytest.mark.asyncio
async def test_temp_too_cold(self):
weather = _make_weather(temp=5)
result = await self.activity.evaluate(weather)
assert result.verdict == Verdict.NO_GO
@pytest.mark.asyncio
async def test_thunderstorm(self):
weather = _make_weather(weather_code=95)
result = await self.activity.evaluate(weather)
assert result.verdict == Verdict.NO_GO
class TestLawnActivity:
def setup_method(self):
self.config = {
"cooldown_days": 3,
"max_precipitation_today_mm": 0,
"max_precipitation_past_24h_mm": 5,
"max_wind_speed_kmh": 25,
"min_temp_c": 15,
}
self.data_store = DataStore(Path("/tmp/test-lawn-data"))
self.activity = LawnActivity(self.config, self.data_store)
@pytest.mark.asyncio
async def test_good_conditions(self):
# Ensure cooldown is met
self.data_store.record_activity_performed("mow_lawn", date(2026, 4, 20))
weather = _make_weather(temp=22, wind=10)
result = await self.activity.evaluate(weather)
assert result.verdict == Verdict.GO
@pytest.mark.asyncio
async def test_cooldown_not_met(self):
# Just mowed yesterday
self.data_store.record_activity_performed("mow_lawn", date(2026, 4, 28))
weather = _make_weather(temp=22, wind=10)
result = await self.activity.evaluate(weather)
assert result.verdict == Verdict.NO_GO
assert any(not r.passed and "cooldown" in r.rule_name.lower()
for r in result.rules)
@pytest.mark.asyncio
async def test_ground_wet(self):
self.data_store.record_activity_performed("mow_lawn", date(2026, 4, 20))
weather = _make_weather(precip_24h=10)
result = await self.activity.evaluate(weather)
assert result.verdict == Verdict.NO_GO
@pytest.mark.asyncio
async def test_rain_today(self):
self.data_store.record_activity_performed("mow_lawn", date(2026, 4, 20))
weather = _make_weather(precip_today=3)
result = await self.activity.evaluate(weather)
assert result.verdict == Verdict.NO_GO