| """ |
| TrackerScheduler: time-based tracking plan with sunrise/sunset resolution. |
| |
| Loads a JSON tracking plan (timeline of events) and resolves the active |
| event for any given timestamp. Supports literal times ("10:30") and |
| astronomical events ("sunrise", "sunset") via the ``astral`` library. |
| |
| Adapted from the tracker repo's AsyncSolarTrackerScheduler — made synchronous |
| and integrated with Baseline's site config. |
| |
| Plan JSON format |
| ---------------- |
| :: |
| |
| { |
| "timeline": [ |
| {"start": "sunrise", "mode": "tracking", "angle": null}, |
| {"start": "10:00", "mode": "antiTracking", "angle": 15}, |
| {"start": "16:00", "mode": "tracking", "angle": null}, |
| {"start": "sunset", "mode": "fixed_angle", "angle": 180} |
| ] |
| } |
| |
| Modes |
| ----- |
| - ``tracking`` — follow the sun (astronomical tracking) |
| - ``antiTracking`` — offset from astronomical position by ``angle`` degrees |
| - ``fixed_angle`` — hold a fixed tilt angle |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import logging |
| from datetime import datetime, date |
| from pathlib import Path |
| from typing import Optional |
| from zoneinfo import ZoneInfo |
|
|
| from astral import LocationInfo |
| from astral.sun import sun |
|
|
| from config.settings import SITE_LATITUDE, SITE_LONGITUDE |
|
|
| logger = logging.getLogger(__name__) |
|
|
| |
| _SITE_TZ = ZoneInfo("Asia/Jerusalem") |
|
|
|
|
| class TrackerScheduler: |
| """Load a JSON tracking plan and resolve events by time. |
| |
| Parameters |
| ---------- |
| plan_file : Path or str, optional |
| Path to the JSON plan file. Either this or ``plan_data`` must be given. |
| plan_data : dict, optional |
| Pre-loaded plan dict (must contain ``"timeline"`` key). |
| latitude, longitude : float |
| Site coordinates for sunrise/sunset calculation. |
| timezone : str |
| IANA timezone name. |
| """ |
|
|
| def __init__( |
| self, |
| plan_file: Optional[Path | str] = None, |
| plan_data: Optional[dict] = None, |
| latitude: float = SITE_LATITUDE, |
| longitude: float = SITE_LONGITUDE, |
| timezone: str = "Asia/Jerusalem", |
| ): |
| self.latitude = latitude |
| self.longitude = longitude |
| self.tz = ZoneInfo(timezone) |
| self.location = LocationInfo( |
| timezone=timezone, |
| latitude=latitude, |
| longitude=longitude, |
| ) |
| self.timeline: list[dict] = [] |
|
|
| if plan_data is not None: |
| self.timeline = plan_data.get("timeline", []) |
| elif plan_file is not None: |
| self._load_plan(Path(plan_file)) |
|
|
| def _load_plan(self, path: Path) -> None: |
| with open(path) as f: |
| data = json.load(f) |
| self.timeline = data.get("timeline", []) |
| logger.info("Loaded plan %s: %d events", path.name, len(self.timeline)) |
|
|
| def _resolve_time(self, event_start: str, ref_date: date) -> datetime: |
| """Resolve an event start string to a timezone-aware datetime.""" |
| if event_start in ("sunrise", "sunset"): |
| s = sun(self.location.observer, date=ref_date, tzinfo=self.tz) |
| return s[event_start] |
| |
| t = datetime.strptime(event_start, "%H:%M").time() |
| return datetime.combine(ref_date, t, tzinfo=self.tz) |
|
|
| def get_event(self, current_time: Optional[datetime] = None) -> Optional[dict]: |
| """Return the active event for the given time. |
| |
| Walks the timeline in reverse and returns the first event |
| whose start time is <= current_time. If no event matches, |
| returns the last event in the timeline (wrap-around). |
| |
| Returns |
| ------- |
| dict with keys: ``start``, ``mode``, ``angle`` (may be None). |
| None if the timeline is empty. |
| """ |
| if not self.timeline: |
| return None |
|
|
| now = current_time or datetime.now(self.tz) |
| ref_date = now.date() if hasattr(now, "date") else now |
|
|
| for event in reversed(self.timeline): |
| try: |
| event_dt = self._resolve_time(event["start"], ref_date) |
| if now >= event_dt: |
| return event |
| except Exception as exc: |
| logger.warning("Failed to resolve event %s: %s", event, exc) |
| continue |
|
|
| |
| return self.timeline[-1] |
|
|
| def get_all_events(self, ref_date: Optional[date] = None) -> list[dict]: |
| """Return all events with resolved timestamps for a given date. |
| |
| Useful for display / debugging. |
| """ |
| today = ref_date or date.today() |
| result = [] |
| for event in self.timeline: |
| try: |
| dt = self._resolve_time(event["start"], today) |
| result.append({ |
| "start_raw": event["start"], |
| "start_resolved": dt.isoformat(), |
| "mode": event.get("mode"), |
| "angle": event.get("angle"), |
| }) |
| except Exception as exc: |
| result.append({ |
| "start_raw": event["start"], |
| "error": str(exc), |
| }) |
| return result |
|
|
|
|
| |
| |
| |
|
|
| PLAN_LIBRARY = { |
| "night-east": { |
| "description": "Track sun during day, face east at night", |
| "timeline": [ |
| {"start": "sunrise", "mode": "tracking", "angle": None}, |
| {"start": "sunset", "mode": "fixed_angle", "angle": 180}, |
| ], |
| }, |
| "day-max": { |
| "description": "Fixed east-facing during day (max morning light), track at night", |
| "timeline": [ |
| {"start": "sunrise", "mode": "fixed_angle", "angle": 180}, |
| {"start": "sunset", "mode": "tracking", "angle": None}, |
| ], |
| }, |
| "day-mid": { |
| "description": "Fixed mid position during day, track at night", |
| "timeline": [ |
| {"start": "sunrise", "mode": "fixed_angle", "angle": 90}, |
| {"start": "sunset", "mode": "tracking", "angle": None}, |
| ], |
| }, |
| "full-tracking": { |
| "description": "Full astronomical tracking 24/7 (default)", |
| "timeline": [ |
| {"start": "sunrise", "mode": "tracking", "angle": None}, |
| ], |
| }, |
| "shading-midday": { |
| "description": "Track morning/evening, anti-track during midday heat", |
| "timeline": [ |
| {"start": "sunrise", "mode": "tracking", "angle": None}, |
| {"start": "10:00", "mode": "antiTracking", "angle": 15}, |
| {"start": "16:00", "mode": "tracking", "angle": None}, |
| ], |
| }, |
| } |
|
|
|
|
| def get_plan(name: str) -> dict: |
| """Look up a built-in plan by name.""" |
| if name not in PLAN_LIBRARY: |
| raise KeyError(f"Unknown plan: {name}. Available: {list(PLAN_LIBRARY.keys())}") |
| return PLAN_LIBRARY[name] |
|
|
