45 lines
1.1 KiB
Python
45 lines
1.1 KiB
Python
import csv
|
|
import datetime
|
|
|
|
from typing import Annotated
|
|
|
|
import uvicorn
|
|
|
|
from fastapi import FastAPI, Form
|
|
from pydantic import BaseModel
|
|
|
|
from .config import settings
|
|
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
class Reading(BaseModel):
|
|
temperature: float
|
|
humidity: float
|
|
age: int
|
|
device: str
|
|
|
|
|
|
@app.post("/upload", response_model=Reading)
|
|
def add_reading(
|
|
temp: Annotated[float, Form()],
|
|
humidity: Annotated[float, Form()],
|
|
age: Annotated[int, Form()],
|
|
device: Annotated[str, Form()],
|
|
):
|
|
current_datetime = datetime.datetime.now()
|
|
deducted_datetime = current_datetime - datetime.timedelta(milliseconds=age)
|
|
append_to_csv(settings.FILE_PATH, temp, humidity, deducted_datetime, device)
|
|
return Reading(temperature=temp, humidity=humidity, age=age, device=device)
|
|
|
|
|
|
def append_to_csv(filename, temperature, humidity, datetime_var, device):
|
|
with open(filename, "a", newline="") as csvfile:
|
|
writer = csv.writer(csvfile)
|
|
writer.writerow([temperature, humidity, datetime_var, device])
|
|
|
|
|
|
def init():
|
|
uvicorn.run(app, host=settings.HOST, port=settings.PORT)
|