Implement

sat_kalender/__main__.py

@@ -0,0 +1,110 @@
+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())

sat_kalender/n2yo.py

@@ -0,0 +1,95 @@
+from contextlib import asynccontextmanager
+from typing import Literal
+
+import aiohttp
+from pydantic import BaseModel, Field
+
+API_URL = "https://api.n2yo.com/rest/v1/satellite"
+
+Compass = Literal[
+    "N",
+    "NNE",
+    "NE",
+    "ENE",
+    "E",
+    "ESE",
+    "SE",
+    "SSE",
+    "S",
+    "SSW",
+    "SW",
+    "WSW",
+    "W",
+    "WNW",
+    "NW",
+    "NNW",
+]
+
+
+class PassInfo(BaseModel):
+    sat_id: int = Field(alias="satid")
+    sat_name: str = Field(alias="satname")
+    transaction_count: int = Field(alias="transactionscount")
+    pass_count: int = Field(alias="passescount")
+
+
+class RadioPass(BaseModel):
+    start_az: float = Field(alias="startAz")
+    start_az_compass: Compass = Field(alias="startAzCompass")
+    start_utc: int = Field(alias="startUTC")
+
+    max_az: float = Field(alias="maxAz")
+    max_az_compass: Compass = Field(alias="maxAzCompass")
+    max_el: float = Field(alias="maxEl")
+    max_utc: int = Field(alias="maxUTC")
+
+    end_az: float = Field(alias="endAz")
+    end_az_compass: Compass = Field(alias="endAzCompass")
+    end_utc: int = Field(alias="endUTC")
+
+
+class RadioPasses(BaseModel):
+    info: PassInfo
+    passes: list[RadioPass] | None
+
+
+class N2YO:
+    def __init__(self, api_key: str):
+        self.api_key = api_key
+        self.client = aiohttp.ClientSession()
+
+    async def get_radio_passes(
+        self,
+        norad: int,
+        latitude: float,
+        longitude: float,
+        altitude: float,
+        days: int,
+        min_elevation: int,
+    ):
+        url = "/".join(
+            str(s)
+            for s in [
+                API_URL,
+                "radiopasses",
+                norad,
+                latitude,
+                longitude,
+                altitude,
+                days,
+                min_elevation,
+            ]
+        )
+
+        async with self.client.get(url, params={"apiKey": self.api_key}) as request:
+            response = await request.json()
+
+        return RadioPasses.parse_obj(response)
+
+@asynccontextmanager
+async def n2yo_api(key: str):
+    api = N2YO(key)
+    try:
+        yield api
+    finally:
+        await api.client.close()

sat_kalender/satellites.py

@@ -0,0 +1,23 @@
+import tomllib
+
+from pydantic import BaseModel
+
+from .settings import settings
+
+class Downlink(BaseModel):
+    proto: str
+    freq: float
+
+class Satellite(BaseModel):
+    name: str
+    downlink: list[Downlink]
+
+
+def read_satellites() -> dict[int, Satellite]:
+    with open(settings.satellites_file, "rb") as f:
+        data = tomllib.load(f)
+
+    return {
+        int(k): Satellite.parse_obj(v)
+        for k, v in data.items()
+    }

sat_kalender/settings.py

@@ -0,0 +1,25 @@
+from pydantic import BaseSettings, BaseModel
+
+class Observer(BaseModel):
+    latitude: float
+    longitude: float
+    altitude: float
+    min_elevation: int = 15
+    good_elevation: int = 50
+
+class CalDav(BaseModel):
+    uri: str
+    username: str
+    password: str
+
+class Settings(BaseSettings):
+    n2yo_api_key: str
+    satellites_file: str = "satellites.toml"
+    observer: Observer
+    caldav: CalDav
+    
+    class Config:
+        env_file = ".env"
+        env_nested_delimiter = '_'
+
+settings = Settings()