Flatten to a single package directory

The checkout is now astrape/astrape-ems, not astrape/astrape-ems/<package>. The package
contents move to the repository root and pyproject maps the package name to
'.' via package-dir, with the subpackages listed explicitly.

The systemd units move out to /srv/astrape/deploy/systemd, outside every
repository, because they describe this deployment rather than any one
component.

Verified both install modes: editable, which is how the deployment runs, and
a plain wheel install, which is what the container build does.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
rpotter6298
2026-08-30 15:05:15 +02:00
parent 91ab976bee
commit d5f3c0f7ac
15 changed files with 6 additions and 27 deletions
View File
+95
View File
@@ -0,0 +1,95 @@
from __future__ import annotations
import argparse
from dataclasses import dataclass
from os import environ
from sys import stderr
from time import sleep
from astrape_ems.classes.env import EnvLoader
from astrape_ems.classes.client import SigenPlantClient
from astrape_ems.classes.store import SigenStore
@dataclass(frozen=True)
class SigenDaemonConfig:
poll_seconds: float
@classmethod
def from_env(cls) -> "SigenDaemonConfig":
return cls(
poll_seconds=float(environ.get("SIGEN_POLL_SECONDS", "5")),
)
class SigenDaemon:
"""Polls Sigenergy plant metrics and stores normalized snapshots."""
def __init__(
self,
config: SigenDaemonConfig,
plant_client: SigenPlantClient,
sigen_store: SigenStore,
) -> None:
self.config = config
self.plant_client = plant_client
self.sigen_store = sigen_store
@classmethod
def from_env(cls) -> "SigenDaemon":
return cls(
config=SigenDaemonConfig.from_env(),
plant_client=SigenPlantClient.from_env(),
sigen_store=SigenStore.from_env(),
)
def initialize(self) -> None:
self.sigen_store.initialize()
def run_once(self) -> int:
snapshot = self.plant_client.fetch_snapshot()
return self.sigen_store.save_snapshot(snapshot)
def run_forever(self) -> None:
self.initialize()
while True:
try:
saved_count = self.run_once()
print(f"stored_sigen_plant_snapshots={saved_count}", flush=True)
except Exception as error:
print(f"sigen_poll_error={error}", file=stderr, flush=True)
sleep(self.config.poll_seconds)
def main() -> None:
try:
EnvLoader().load()
daemon = SigenDaemon.from_env()
args = parse_args()
if args.once:
daemon.initialize()
saved_count = daemon.run_once()
print(f"stored_sigen_plant_snapshots={saved_count}", flush=True)
return
daemon.run_forever()
except Exception as error:
print(f"sigen_daemon_startup_error={error}", file=stderr)
raise SystemExit(1) from error
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Poll Sigenergy plant metrics into Astrape's database."
)
parser.add_argument(
"--once",
action="store_true",
help="Initialize storage, save one snapshot, and exit.",
)
return parser.parse_args()
if __name__ == "__main__":
main()