import asyncio from dataclasses import dataclass from datetime import datetime, timezone from uuid import uuid4 import os from aiocaldav import DAVClient, Calendar from .n2yo import n2yo_api, Compass from .satellites import read_satellites, Downlink from .settings import settings @dataclass class Pass: name: str norad: int start_utc: int start_az: float start_compass: Compass end_utc: int end_az: float end_compass: Compass max_el: float downlinks: list[Downlink] def as_ical(self) -> str: now = datetime.now().astimezone(timezone.utc) start = datetime.fromtimestamp(self.start_utc, timezone.utc) end = datetime.fromtimestamp(self.end_utc, timezone.utc) summary = f"{self.name} @ {self.max_el:.0f}deg" if self.max_el > settings.observer.good_elevation: summary += " [!]" description = [] description.append( f"Azimuth: {self.start_az} ({self.start_compass}) to {self.end_az} ({self.end_compass})" ) description.append("") description.append("Downlinks:") for downlink in self.downlinks: description.append(f"- {downlink.proto} @ {downlink.freq} MHz") description = "\\n".join(description) return ( "BEGIN:VCALENDAR\n" "VERSION:2.0\n" "PRODID:-//leafbla.de//sat-kalender.py//EN\n" "BEGIN:VEVENT\n" f"UID:sat-kalender-{uuid4()}\n" f"DTSTAMP:{now.year:04}{now.month:02}{now.day:02}T{now.hour:02}{now.minute:02}{now.second:02}Z\n" f"DTSTART:{start.year:04}{start.month:02}{start.day:02}T{start.hour:02}{start.minute:02}{start.second:02}Z\n" f"DTEND:{end.year:04}{end.month:02}{end.day:02}T{end.hour:02}{end.minute:02}{end.second:02}Z\n" f"SUMMARY:{summary}\n" f"DESCRIPTION:{description}\n" "END:VEVENT\n" "END:VCALENDAR\n" ) async def main(): satellites = read_satellites() passes = list() print(f"Fetching data for {', '.join(v.name for v in satellites.values())}") async with n2yo_api(settings.n2yo_api_key) as api: for norad, info in satellites.items(): radio_passes = await api.get_radio_passes( norad, settings.observer.latitude, settings.observer.longitude, settings.observer.altitude, 1, settings.observer.min_elevation, ) print(f"{info.name}: {len(radio_passes.passes)} passes") for radio_pass in radio_passes.passes: passes.append( Pass( info.name, norad, radio_pass.start_utc, radio_pass.start_az, radio_pass.start_az_compass, radio_pass.end_utc, radio_pass.end_az, radio_pass.end_az_compass, radio_pass.max_el, info.downlink, ) ) print("Adding events to calendar") dav_client = DAVClient( settings.caldav.uri, username=settings.caldav.username, password=settings.caldav.password, ) cal = Calendar(client=dav_client, url=settings.caldav.uri) await asyncio.gather(*[cal.add_event(p.as_ical()) for p in passes]) print(f"Done :3") if __name__ == "__main__": asyncio.run(main())