47 lines
1.2 KiB
Python
47 lines
1.2 KiB
Python
from fastapi import FastAPI
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel
|
|
from .anomaly import AnomalyEngine
|
|
from .models import TelemetryEvent
|
|
from .adapters.telemetry_gateway import parse_telemetry
|
|
import threading
|
|
|
|
|
|
app = FastAPI(title="SolarPulse Viz Real MVP")
|
|
|
|
# Initialize anomaly engine once
|
|
anomaly_engine = AnomalyEngine()
|
|
|
|
_lock = threading.Lock()
|
|
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"status": "SolarPulse Viz Real MVP is running"}
|
|
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.post("/telemetry")
|
|
async def telemetry(payload: TelemetryEvent):
|
|
# Compute a lightweight anomaly score for the incoming payload
|
|
with _lock:
|
|
score = anomaly_engine.score({
|
|
"v_current": payload.v_current,
|
|
"temp": payload.temp,
|
|
# Use actual vibration value if provided (default 0.0)
|
|
"vibration": payload.vibration,
|
|
})
|
|
|
|
# In a real system we'd persist to a DB and route to dashboards/alerts
|
|
return {
|
|
"site_id": payload.site_id,
|
|
"asset_id": payload.asset_id,
|
|
"timestamp": payload.timestamp,
|
|
"anomaly_score": score,
|
|
"status": payload.status or "OK",
|
|
}
|