from __future__ import annotations from datetime import datetime, timedelta, timezone from os import environ from astrape_ems.classes.env import EnvLoader from astrape_dashboard.classes.sample_data import WeatherSampleData from astrape_oracle.classes.weather.store import WeatherStore, WeatherStoreConfigurationError from astrape_dashboard.classes.displays.weather import WeatherDisplay from astrape_dashboard.classes.displays.station import WeatherStationDisplay from astrape_dashboard.classes.displays.oracle import OracleDisplay from astrape_dashboard.classes.displays.oracle_quality import OracleQualityDisplay class WebUI: """Composes Astrape web modules into a small control panel.""" def __init__(self) -> None: self.weather_display = WeatherDisplay() self.weather_station_display = WeatherStationDisplay() self.oracle_display = OracleDisplay() self.oracle_quality_display = OracleQualityDisplay() def render_page(self, page: str = "oracle") -> str: current_page = ( page if page in {"oracle", "weather", "weather_station", "quality"} else "oracle" ) return f""" Astrape

Astrape

Gibil control panel

{self._nav(current_page)}
{self._page_body(current_page)}
""" def _nav(self, current_page: str) -> str: pages = [ ("oracle", "/oracle", "Oracle"), ("weather", "/weather", "Weather"), ("weather_station", "/weather-station", "Station"), ("quality", "/quality", "Quality"), ] links = [ f'{label}' for key, href, label in pages ] return f"" def _page_body(self, page: str) -> str: if page == "weather": return self.weather_display.render() if page == "weather_station": return self.weather_station_display.render() if page == "quality": return self.oracle_quality_display.render() return self.oracle_display.render() def weather_payload(self) -> str: EnvLoader().load() if environ.get("ASTRAPE_WEB_SAMPLE_DATA") == "1": return self.weather_display.data_payload(WeatherSampleData().build()) try: dataset = WeatherStore.from_env().load_display_dataset() except WeatherStoreConfigurationError: dataset = None return self.weather_display.data_payload(dataset) def weather_station_payload(self, lookback_hours: float = 168) -> str: EnvLoader().load() try: hours = min(2160, max(1, float(lookback_hours))) except (TypeError, ValueError): hours = 168 try: now = datetime.now(timezone.utc) dataset = WeatherStore.from_env().load_station_display_dataset( start_at=now - timedelta(hours=hours), end_at=now, ) except WeatherStoreConfigurationError: dataset = None return self.weather_station_display.data_payload(dataset) def oracle_payload(self) -> str: EnvLoader().load() return self.oracle_display.data_payload() def oracle_quality_payload(self, lookback_hours: float = 168) -> str: EnvLoader().load() return self.oracle_quality_display.data_payload(lookback_hours=lookback_hours)