83 lines
1.8 KiB
Python
83 lines
1.8 KiB
Python
from typing import Optional
|
|
from pydantic import BaseModel
|
|
|
|
from fastapi import FastAPI, Request, Body
|
|
|
|
from collections import defaultdict
|
|
|
|
from . import utils
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
mesures = defaultdict(list)
|
|
|
|
|
|
class Sonde(BaseModel):
|
|
identifiant: str
|
|
nom: str
|
|
|
|
|
|
sondes = {"test": Sonde(identifiant="test", nom="Testlocal")}
|
|
|
|
|
|
@app.get("/")
|
|
def read_root():
|
|
return {"msg": "Hello World"}
|
|
|
|
|
|
@app.post("/sonde/")
|
|
def post_sonde(sonde: Sonde):
|
|
print(sonde)
|
|
sondes[sonde.identifiant] = sonde
|
|
return list(sondes.values())
|
|
|
|
|
|
@app.get("/sonde/")
|
|
def list_sonde():
|
|
return [x for x in sondes.values()]
|
|
|
|
|
|
@app.post("/sonde/{idsonde}/")
|
|
def post_sonde_data(idsonde: str, body: dict = Body(...)):
|
|
if idsonde in sondes:
|
|
mesures_ = mesures[idsonde]
|
|
mesures_.append(body)
|
|
|
|
print(len(mesures_))
|
|
return {"count": len(mesures_)}
|
|
# print(dict(mesures))
|
|
|
|
|
|
@app.get("/sonde/{idsonde}/rapport")
|
|
def get_rapport(idsonde):
|
|
try:
|
|
last = mesures[idsonde][-1]
|
|
except IndexError:
|
|
pass
|
|
else:
|
|
last = utils.prepare(last)
|
|
for name, data in sorted(last["channels"].items()):
|
|
yield f'{name} {data["status"]}'
|
|
|
|
|
|
@app.get("/sonde/{idsonde}/historique")
|
|
def historique(idsonde):
|
|
print(len(mesures[idsonde]))
|
|
|
|
for previous, present in zip(mesures[idsonde], mesures[idsonde][1:]):
|
|
date = present["date"]
|
|
previous = utils.prepare(previous)
|
|
present = utils.prepare(present)
|
|
all_channels = sorted(set((*previous["channels"], *present["channels"])))
|
|
|
|
diff = utils.compare(all_channels, previous, present)
|
|
|
|
if diff:
|
|
for d in diff["changements"]:
|
|
line = date, *d
|
|
print(line)
|
|
yield " ".join(line)
|
|
else:
|
|
print(f"no change for {date}")
|