rpotter6298
c8e3016fd6
- Implement `sigen_daemon.py` to poll Sigenergy plant metrics and store snapshots. - Create `web_daemon.py` for serving a web interface with various endpoints. - Add debug scripts: - `debug_duplicates.py` to find duplicate target times in forecast data. - `debug_energy_forecast.py` to print baseline energy forecast curves. - `debug_oracle_evaluations.py` to run the oracle evaluator. - `debug_sigen.py` to inspect stored Sigenergy plant snapshots. - `debug_weather.py` to trace resolved truth data. - `modbus_test.py` for exploring Sigenergy plants or inverters over Modbus TCP. - Introduce `oracle_evaluator.py` for evaluating stored oracle predictions against actuals. - Add TCN training scripts in `tcn` directory for training usage sequence models.
143 lines
4.8 KiB
Python
143 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from importlib import import_module, reload
|
|
from os import environ
|
|
from pathlib import Path
|
|
import json
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
from gibil.classes.env_loader import EnvLoader
|
|
|
|
EnvLoader().load()
|
|
|
|
HOST = environ.get("ASTRAPE_WEB_HOST", "0.0.0.0")
|
|
PORT = int(environ.get("ASTRAPE_WEB_PORT", "8080"))
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
|
WATCHED_PATHS = [
|
|
PROJECT_ROOT / "gibil" / "classes" / "webui.py",
|
|
PROJECT_ROOT / "gibil" / "classes" / "weather" / "display.py",
|
|
PROJECT_ROOT / "gibil" / "classes" / "oracle" / "display.py",
|
|
PROJECT_ROOT / "gibil" / "classes" / "oracle" / "quality_display.py",
|
|
PROJECT_ROOT / "gibil" / "classes" / "weather" / "store.py",
|
|
PROJECT_ROOT / "gibil" / "classes" / "oracle" / "store.py",
|
|
PROJECT_ROOT / "gibil" / "classes" / "oracle" / "builder.py",
|
|
PROJECT_ROOT / "gibil" / "classes" / "oracle" / "config.py",
|
|
PROJECT_ROOT / "gibil" / "classes" / "sigen" / "store.py",
|
|
]
|
|
|
|
|
|
class AstrapeWebHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None:
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path
|
|
|
|
if path in {"/", "/oracle"}:
|
|
self._send_html(self._webui().render_page("oracle"))
|
|
return
|
|
|
|
if path == "/weather":
|
|
self._send_html(self._webui().render_page("weather"))
|
|
return
|
|
|
|
if path == "/quality":
|
|
self._send_html(self._webui().render_page("quality"))
|
|
return
|
|
|
|
if path == "/api/weather":
|
|
self._send_json_text(self._webui().weather_payload())
|
|
return
|
|
|
|
if path == "/api/oracle":
|
|
self._send_json_text(self._webui().oracle_payload())
|
|
return
|
|
|
|
if path == "/api/oracle-quality":
|
|
params = parse_qs(parsed.query)
|
|
lookback_hours = self._float_param(params, "lookback_hours", 168)
|
|
self._send_json_text(
|
|
self._webui().oracle_quality_payload(
|
|
lookback_hours=lookback_hours
|
|
)
|
|
)
|
|
return
|
|
|
|
if path == "/api/ui-version":
|
|
self._send_json_text(json.dumps({"version": self._ui_version()}))
|
|
return
|
|
|
|
self.send_error(404)
|
|
|
|
def log_message(self, format: str, *args: object) -> None:
|
|
print(f"{self.address_string()} - {format % args}")
|
|
|
|
def _webui(self):
|
|
weather_store_module = import_module("gibil.classes.weather.store")
|
|
sigen_store_module = import_module("gibil.classes.sigen.store")
|
|
oracle_store_module = import_module("gibil.classes.oracle.store")
|
|
oracle_builder_module = import_module("gibil.classes.oracle.builder")
|
|
oracle_display_module = import_module("gibil.classes.oracle.display")
|
|
oracle_quality_display_module = import_module(
|
|
"gibil.classes.oracle.quality_display"
|
|
)
|
|
weather_display_module = import_module("gibil.classes.weather.display")
|
|
webui_module = import_module("gibil.classes.webui")
|
|
reload(weather_store_module)
|
|
reload(sigen_store_module)
|
|
reload(oracle_store_module)
|
|
reload(oracle_builder_module)
|
|
reload(oracle_display_module)
|
|
reload(oracle_quality_display_module)
|
|
reload(weather_display_module)
|
|
reload(webui_module)
|
|
return webui_module.WebUI()
|
|
|
|
def _float_param(
|
|
self,
|
|
params: dict[str, list[str]],
|
|
key: str,
|
|
default: float,
|
|
) -> float:
|
|
values = params.get(key)
|
|
if not values:
|
|
return default
|
|
try:
|
|
return float(values[0])
|
|
except ValueError:
|
|
return default
|
|
|
|
def _ui_version(self) -> str:
|
|
mtimes = [
|
|
str(path.stat().st_mtime_ns)
|
|
for path in WATCHED_PATHS
|
|
if path.exists()
|
|
]
|
|
return ".".join(mtimes)
|
|
|
|
def _send_html(self, body: str) -> None:
|
|
encoded = body.encode("utf-8")
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(encoded)))
|
|
self.end_headers()
|
|
self.wfile.write(encoded)
|
|
|
|
def _send_json_text(self, body: str) -> None:
|
|
encoded = body.encode("utf-8")
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.send_header("Content-Length", str(len(encoded)))
|
|
self.end_headers()
|
|
self.wfile.write(encoded)
|
|
|
|
|
|
def main() -> None:
|
|
server = ThreadingHTTPServer((HOST, PORT), AstrapeWebHandler)
|
|
print(f"Astrape web UI listening on http://{HOST}:{PORT}")
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|