From 86e6499895e684ffa36482271c24fcdc0422e518 Mon Sep 17 00:00:00 2001 From: vidane Date: Wed, 29 Apr 2026 04:46:01 +0000 Subject: [PATCH] Initial commit: weather-or-not scaffold --- .gitignore | 21 +++ README.md | 89 +++++++++++++ config.yaml | 43 ++++++ pyproject.toml | 40 ++++++ src/weather_or_not/__init__.py | 3 + src/weather_or_not/activities.py | 221 +++++++++++++++++++++++++++++++ src/weather_or_not/cli.py | 171 ++++++++++++++++++++++++ src/weather_or_not/lawn.py | 144 ++++++++++++++++++++ src/weather_or_not/motorcycle.py | 152 +++++++++++++++++++++ src/weather_or_not/notify.py | 105 +++++++++++++++ src/weather_or_not/weather.py | 180 +++++++++++++++++++++++++ tests/__init__.py | 0 tests/test_activities.py | 184 +++++++++++++++++++++++++ 13 files changed, 1353 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 config.yaml create mode 100644 pyproject.toml create mode 100644 src/weather_or_not/__init__.py create mode 100644 src/weather_or_not/activities.py create mode 100644 src/weather_or_not/cli.py create mode 100644 src/weather_or_not/lawn.py create mode 100644 src/weather_or_not/motorcycle.py create mode 100644 src/weather_or_not/notify.py create mode 100644 src/weather_or_not/weather.py create mode 100644 tests/__init__.py create mode 100644 tests/test_activities.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..69617a8 --- /dev/null +++ b/.gitignore @@ -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/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..d9dfc0d --- /dev/null +++ b/README.md @@ -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 diff --git a/config.yaml b/config.yaml new file mode 100644 index 0000000..eb36484 --- /dev/null +++ b/config.yaml @@ -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) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..507e802 --- /dev/null +++ b/pyproject.toml @@ -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 diff --git a/src/weather_or_not/__init__.py b/src/weather_or_not/__init__.py new file mode 100644 index 0000000..0f90f88 --- /dev/null +++ b/src/weather_or_not/__init__.py @@ -0,0 +1,3 @@ +"""Weather or Not - Weather-based activity decision engine.""" + +__version__ = "0.1.0" diff --git a/src/weather_or_not/activities.py b/src/weather_or_not/activities.py new file mode 100644 index 0000000..682705c --- /dev/null +++ b/src/weather_or_not/activities.py @@ -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, + } diff --git a/src/weather_or_not/cli.py b/src/weather_or_not/cli.py new file mode 100644 index 0000000..a2a25a6 --- /dev/null +++ b/src/weather_or_not/cli.py @@ -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() diff --git a/src/weather_or_not/lawn.py b/src/weather_or_not/lawn.py new file mode 100644 index 0000000..3c5bd18 --- /dev/null +++ b/src/weather_or_not/lawn.py @@ -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) diff --git a/src/weather_or_not/motorcycle.py b/src/weather_or_not/motorcycle.py new file mode 100644 index 0000000..8d89814 --- /dev/null +++ b/src/weather_or_not/motorcycle.py @@ -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" + ) diff --git a/src/weather_or_not/notify.py b/src/weather_or_not/notify.py new file mode 100644 index 0000000..7949260 --- /dev/null +++ b/src/weather_or_not/notify.py @@ -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) diff --git a/src/weather_or_not/weather.py b/src/weather_or_not/weather.py new file mode 100644 index 0000000..0f92764 --- /dev/null +++ b/src/weather_or_not/weather.py @@ -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})") diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_activities.py b/tests/test_activities.py new file mode 100644 index 0000000..10c02c2 --- /dev/null +++ b/tests/test_activities.py @@ -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