rpotter6298
5b562338ab
Everything that talks to the energy backend: the Sigenergy Modbus client, its register map, the plant store, and the polling daemon. Sits at the base of the Astrape dependency chain, because the plant store is what satisfies gibil's PlantHistoryPort -- the oracle cannot forecast without observed plant history. EnvLoader lives here for the same reason: it is the only place all three repos can see it without a fourth repo for 37 lines. EnvLoader no longer resolves env/ from __file__. Where a package is installed says nothing about where the deployment keeps its secrets; it now uses the working directory, or ASTRAPE_ENV_DIR. Extracted from a single repository where the forecasting engine, the database plumbing, the hardware client, the ingest daemons and the web UI all imported each other freely. Verified that every module imports with gibil resolved from its own installed checkout. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
46 lines
1.6 KiB
Python
46 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
from os import environ
|
|
from pathlib import Path
|
|
|
|
|
|
class EnvLoader:
|
|
"""Loads Astrape env files without overriding process environment.
|
|
|
|
The directory is resolved against the working directory, or
|
|
ASTRAPE_ENV_DIR if set. Deliberately not derived from __file__: where a
|
|
package happens to be installed says nothing about where the deployment
|
|
keeps its secrets, and tying the two breaks the moment the package moves.
|
|
Every unit sets WorkingDirectory=/srv/astrape, so the default is correct
|
|
for the daemons and for a shell started in the project root.
|
|
"""
|
|
|
|
def __init__(self, env_dir: Path | None = None) -> None:
|
|
if env_dir is None:
|
|
env_dir = Path(environ.get("ASTRAPE_ENV_DIR", "env"))
|
|
self.env_dir = env_dir
|
|
|
|
def load(self) -> None:
|
|
if not self.env_dir.exists():
|
|
return
|
|
|
|
for path in sorted(self.env_dir.glob("*.env")):
|
|
self._load_file(path)
|
|
|
|
def _load_file(self, path: Path) -> None:
|
|
for raw_line in path.read_text(encoding="utf-8").splitlines():
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
|
|
key, value = line.split("=", 1)
|
|
key = key.strip()
|
|
value = self._clean_value(value.strip())
|
|
if key and key not in environ:
|
|
environ[key] = value
|
|
|
|
def _clean_value(self, value: str) -> str:
|
|
if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
|
|
return value[1:-1]
|
|
return value
|