File size: 3,458 Bytes
95c00ee | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | """Versioned platform APIs: buildings, telemetry, points."""
from __future__ import annotations
from typing import Optional
from fastapi import APIRouter, HTTPException, Query
from backend.knowledge.hvac_guide_catalog import catalog_all, catalog_record
from backend.services.canonical_telemetry_service import latest_points
from database.models import Building, Equipment, Point
from database.session import SessionLocal
router = APIRouter(prefix="/api/v1", tags=["v1"])
@router.get("/guide-catalog")
@router.get("/opportunities/catalog")
async def guide_catalog():
return {
"opportunities": catalog_all(),
"source_document": "150317hvacguide.pdf",
"energy_impact_note": "guide_savings_potential is GUIDE_POTENTIAL, not measured savings",
}
@router.get("/guide-catalog/{oid}")
@router.get("/opportunities/catalog/{oid}")
async def guide_catalog_one(oid: str):
item = catalog_record(oid)
if not item:
raise HTTPException(status_code=404, detail={"code": "UNKNOWN_OPPORTUNITY", "message": "Official catalog IDs are O1–O20."})
return item
@router.get("/buildings")
async def buildings():
db = SessionLocal()
try:
rows = db.query(Building).all()
return {
"buildings": [
{
"id": b.id,
"name": b.name,
"location": b.location,
"area_sqft": b.area_sqft,
"floors": b.floors,
"design_cooling_tonnage": b.design_cooling_tonnage,
}
for b in rows
]
}
finally:
db.close()
@router.get("/buildings/{building_id}")
async def building_one(building_id: str):
db = SessionLocal()
try:
b = db.query(Building).filter_by(id=building_id).first()
if not b:
raise HTTPException(status_code=404, detail={"code": "NOT_FOUND", "message": "Building not found."})
eq = db.query(Equipment).filter_by(building_id=building_id).all()
return {
"id": b.id,
"name": b.name,
"location": b.location,
"area_sqft": b.area_sqft,
"equipment": [{"id": e.id, "name": e.name, "type": e.type} for e in eq],
}
finally:
db.close()
@router.get("/equipment")
async def equipment(building_id: Optional[str] = None):
db = SessionLocal()
try:
q = db.query(Equipment)
if building_id:
q = q.filter_by(building_id=building_id)
rows = q.all()
return {"equipment": [{"id": e.id, "building_id": e.building_id, "name": e.name, "type": e.type} for e in rows]}
finally:
db.close()
@router.get("/telemetry")
async def telemetry(building_id: Optional[str] = None):
return {"points": latest_points(building_id)}
@router.get("/points")
async def points(equipment_id: Optional[str] = None):
db = SessionLocal()
try:
q = db.query(Point)
if equipment_id:
q = q.filter_by(equipment_id=equipment_id)
rows = q.limit(500).all()
return {
"points": [
{"id": p.id, "name": p.name, "equipment_id": p.equipment_id, "unit": p.unit, "category": p.category}
for p in rows
]
}
finally:
db.close()
|