fast api

backend/app/api/alerts.py

@@ -0,0 +1,18 @@
+from fastapi import APIRouter, HTTPException, Query
+
+from app.schemas.alert import AlertOut
+from app.services import alert_service, equipment_service
+
+router = APIRouter(prefix="/alerts", tags=["Алерты"])
+
+
+@router.get("", response_model=list[AlertOut], summary="Список последних алертов")
+def list_alerts(limit: int = Query(50, ge=1, le=500)):
+    return alert_service.list_alerts(limit=limit)
+
+
+@router.get("/{equipment_id}", response_model=list[AlertOut], summary="Алерты по оборудованию")
+def alerts_by_equipment(equipment_id: int, limit: int = Query(50, ge=1, le=500)):
+    if equipment_service.get_equipment(equipment_id) is None:
+        raise HTTPException(status_code=404, detail="Оборудование не найдено")
+    return alert_service.list_alerts(equipment_id, limit)

backend/app/api/equipment.py

@@ -0,0 +1,41 @@
+from fastapi import APIRouter, HTTPException, Query
+
+from app.schemas.equipment import EquipmentOut, ForecastOut, ReadingOut, StatusOut
+from app.services import equipment_service
+
+router = APIRouter(prefix="/equipment", tags=["Оборудование"])
+
+
+def _or_404(value):
+    if value is None:
+        raise HTTPException(status_code=404, detail="Оборудование не найдено")
+    return value
+
+
+@router.get("", response_model=list[EquipmentOut], summary="Список всех единиц оборудования")
+def list_equipment():
+    return equipment_service.list_equipment()
+
+
+@router.get("/{equipment_id}", response_model=EquipmentOut, summary="Информация об одной единице")
+def get_equipment(equipment_id: int):
+    return _or_404(equipment_service.get_equipment(equipment_id))
+
+
+@router.get("/{equipment_id}/status", response_model=StatusOut, summary="Текущее состояние")
+def get_status(equipment_id: int):
+    return _or_404(equipment_service.get_status(equipment_id))
+
+
+@router.get("/{equipment_id}/history", response_model=list[ReadingOut],
+            summary="История показаний датчиков")
+def get_history(equipment_id: int,
+                from_cycle: int | None = Query(None, ge=0),
+                to_cycle: int | None = Query(None, ge=0),
+                limit: int = Query(500, ge=1, le=5000)):
+    return _or_404(equipment_service.get_history(equipment_id, from_cycle, to_cycle, limit))
+
+
+@router.get("/{equipment_id}/forecast", response_model=ForecastOut, summary="Прогноз деградации")
+def get_forecast(equipment_id: int):
+    return _or_404(equipment_service.get_forecast(equipment_id))

backend/app/api/predict.py

@@ -0,0 +1,15 @@
+from fastapi import APIRouter, HTTPException
+
+from app.schemas.equipment import StatusOut
+from app.schemas.predict import PredictIn
+from app.services import predict_service
+
+router = APIRouter(prefix="/predict", tags=["Прогнозирование"])
+
+
+@router.post("", response_model=StatusOut, summary="Ручной запрос инференса")
+def predict(body: PredictIn):
+    result = predict_service.run_prediction(body.equipment_id)
+    if result is None:
+        raise HTTPException(status_code=404, detail="Оборудование не найдено")
+    return result

backend/app/core/config.py

@@ -0,0 +1,7 @@
+import os
+
+CORS_ORIGINS = os.getenv("CORS_ORIGINS", "http://localhost:5173").split(",")
+
+RUL_WARNING = 75
+RUL_CRITICAL = 30
+RUL_CLIP = 125

backend/app/schemas/alert.py

@@ -0,0 +1,13 @@
+from datetime import datetime
+from typing import Literal
+
+from pydantic import BaseModel
+
+
+class AlertOut(BaseModel):
+    id: int
+    equipment_id: int
+    cycle: int
+    triggered_at: datetime
+    severity: Literal["info", "warning", "critical"]
+    message: str

backend/app/schemas/equipment.py

@@ -0,0 +1,44 @@
+from typing import Literal
+
+from pydantic import BaseModel
+
+Status = Literal["normal", "warning", "critical"]
+
+
+class EquipmentOut(BaseModel):
+    id: int
+    name: str
+    type: str
+    unit_number: int
+    current_cycle: int
+    status: Status
+    predicted_rul: float
+
+
+class StatusOut(BaseModel):
+    equipment_id: int
+    status: Status
+    current_cycle: int
+    predicted_rul: float
+    defect_probability: float
+    residual: float
+
+
+class ReadingOut(BaseModel):
+    cycle: int
+    sensors: dict[str, float]
+
+
+class ForecastPoint(BaseModel):
+    cycle: int
+    rul: float
+    rul_low: float
+    rul_high: float
+
+
+class ForecastOut(BaseModel):
+    equipment_id: int
+    current_cycle: int
+    predicted_rul: float
+    predicted_failure_cycle: int
+    points: list[ForecastPoint]

backend/app/schemas/predict.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel, Field
+
+
+class PredictIn(BaseModel):
+    equipment_id: int = Field(ge=1)

backend/app/services/alert_service.py

@@ -0,0 +1,6 @@
+from app.services.mock_data import ALERTS
+
+
+def list_alerts(equipment_id: int | None = None, limit: int = 50) -> list[dict]:
+    items = [a for a in ALERTS if equipment_id is None or a["equipment_id"] == equipment_id]
+    return items[:limit]

backend/app/services/equipment_service.py

@@ -0,0 +1,48 @@
+import math
+
+from app.services.mock_data import EQUIPMENT, READINGS
+
+
+def list_equipment() -> list[dict]:
+    return list(EQUIPMENT.values())
+
+
+def get_equipment(equipment_id: int) -> dict | None:
+    return EQUIPMENT.get(equipment_id)
+
+
+def get_status(equipment_id: int) -> dict | None:
+    e = EQUIPMENT.get(equipment_id)
+    if e is None:
+        return None
+    rul = e["predicted_rul"]
+    return {
+        "equipment_id": equipment_id, "status": e["status"], "current_cycle": e["current_cycle"],
+        "predicted_rul": rul, "residual": e["residual"],
+        "defect_probability": round(1 / (1 + math.exp((rul - 50) / 12)), 3),
+    }
+
+
+def get_history(equipment_id: int, from_cycle: int | None, to_cycle: int | None,
+                limit: int) -> list[dict] | None:
+    if equipment_id not in EQUIPMENT:
+        return None
+    rows = [r for r in READINGS[equipment_id]
+            if (from_cycle is None or r["cycle"] >= from_cycle)
+            and (to_cycle is None or r["cycle"] <= to_cycle)]
+    return rows[-limit:]
+
+
+def get_forecast(equipment_id: int) -> dict | None:
+    e = EQUIPMENT.get(equipment_id)
+    if e is None:
+        return None
+    rul, cycle = e["predicted_rul"], e["current_cycle"]
+    horizon = round(rul)
+    points = []
+    for k in range(horizon + 1):
+        value, spread = rul - k, 2 + 0.12 * k
+        points.append({"cycle": cycle + k, "rul": round(value, 1),
+                       "rul_low": round(max(0.0, value - spread), 1), "rul_high": round(value + spread, 1)})
+    return {"equipment_id": equipment_id, "current_cycle": cycle, "predicted_rul": rul,
+            "predicted_failure_cycle": cycle + horizon, "points": points}

backend/app/services/mock_data.py

@@ -0,0 +1,62 @@
+import random
+from datetime import UTC, datetime, timedelta
+
+from app.core.config import RUL_CLIP, RUL_CRITICAL, RUL_WARNING
+
+SENSORS = {
+    "s2": (641.82, 0.50, 1), "s3": (1589.70, 6.13, 1), "s4": (1400.60, 9.00, 1),
+    "s7": (554.36, 0.89, -1), "s8": (2388.06, 0.07, 1), "s9": (9046.19, 22.08, 1),
+    "s11": (47.47, 0.27, 1), "s12": (521.66, 0.74, -1), "s13": (2388.02, 0.07, 1),
+    "s14": (8138.62, 19.08, 1), "s15": (8.4195, 0.04, 1), "s17": (392.0, 1.55, 1),
+    "s20": (39.06, 0.18, -1), "s21": (23.419, 0.11, -1),
+}
+
+_rnd = random.Random(42)
+
+EQUIPMENT: dict[int, dict] = {}
+READINGS: dict[int, list[dict]] = {}
+ALERTS: list[dict] = []
+
+
+def status_from_rul(rul: float) -> str:
+    if rul <= RUL_CRITICAL:
+        return "critical"
+    if rul <= RUL_WARNING:
+        return "warning"
+    return "normal"
+
+
+def noisy_rul(true_rul: float) -> float:
+    return round(min(max(min(true_rul, RUL_CLIP) + _rnd.gauss(0, 4), 0), RUL_CLIP), 1)
+
+
+def _build() -> None:
+    now = datetime.now(UTC)
+    for i in range(1, 13):
+        life = _rnd.randint(130, 300)                  
+        cut = int(life * _rnd.uniform(0.35, 0.97))     
+        READINGS[i] = [
+            {"cycle": c, "sensors": {
+                code: round(base + trend * std * 3 * (c / life) ** 2 + _rnd.gauss(0, std * 0.4), 4)
+                for code, (base, std, trend) in SENSORS.items()}}
+            for c in range(1, cut + 1)
+        ]
+        predicted = noisy_rul(life - cut)
+        EQUIPMENT[i] = {
+            "id": i, "name": f"ТРДД-{i:03d}", "type": "Турбовентиляторный двигатель",
+            "unit_number": i, "current_cycle": cut, "predicted_rul": predicted,
+            "status": status_from_rul(predicted), "residual": round(_rnd.gauss(0, 0.5), 3),
+            "true_rul": life - cut,
+        }
+        for threshold, severity, text in (
+            (RUL_WARNING, "warning", "Остаточный ресурс ниже 75 циклов — запланируйте обслуживание"),
+            (RUL_CRITICAL, "critical", "Остаточный ресурс ниже 30 циклов — требуется срочная проверка"),
+        ):
+            if life - threshold <= cut:
+                ALERTS.append({"id": len(ALERTS) + 1, "equipment_id": i, "cycle": life - threshold,
+                               "triggered_at": now - timedelta(seconds=30 * (cut - (life - threshold))),
+                               "severity": severity, "message": text})
+    ALERTS.sort(key=lambda a: a["triggered_at"], reverse=True)
+
+
+_build()

backend/app/services/predict_service.py

@@ -0,0 +1,11 @@
+from app.services import equipment_service
+from app.services.mock_data import EQUIPMENT, noisy_rul, status_from_rul
+
+
+def run_prediction(equipment_id: int) -> dict | None:
+    e = EQUIPMENT.get(equipment_id)
+    if e is None:
+        return None
+    e["predicted_rul"] = noisy_rul(e["true_rul"])
+    e["status"] = status_from_rul(e["predicted_rul"])
+    return equipment_service.get_status(equipment_id)

backend/main.py

@@ -0,0 +1,13 @@
+from app.api import alerts, equipment, predict
+from app.core.config import CORS_ORIGINS
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+app = FastAPI(title="OFF-Tech: оценка состояния оборудования", version="0.1.0")
+
+app.add_middleware(CORSMiddleware, allow_origins=CORS_ORIGINS,
+                   allow_methods=["*"], allow_headers=["*"])
+
+app.include_router(equipment.router)
+app.include_router(alerts.router)
+app.include_router(predict.router)

backend/requirements.txt

@@ -8,4 +8,4 @@ httpx==0.27.2
 alembic==1.20.0
 psycopg[binary]==3.3.6
 alembic==1.20.0
-python-dotenv==1.0.1
+python-dotenv==1.0.1