114 lines
3.1 KiB
Python
114 lines
3.1 KiB
Python
#!/usr/bin/python
|
|
# -*- coding: utf-8 -*-
|
|
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request
|
|
from fastapi.responses import JSONResponse, HTMLResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from typing import List
|
|
import uvicorn
|
|
import time
|
|
from pathlib import Path
|
|
from json_extensions import load_from_json_file, save_to_json_file
|
|
|
|
|
|
Base_Log = Path('./log')
|
|
|
|
Activities_File = Path('activities.json')
|
|
|
|
Base_Log.mkdir(exist_ok=True, parents=True)
|
|
app = FastAPI()
|
|
|
|
# Mount static files like CSS/JS/images
|
|
app.mount("/static", StaticFiles(directory="static"), name="static")
|
|
|
|
# Set up templates directory
|
|
templates = Jinja2Templates(directory="templates")
|
|
|
|
# Dummy shared state
|
|
current_activity = "No activity yet"
|
|
if Activities_File.is_file():
|
|
activities = load_from_json_file(Activities_File)
|
|
else:
|
|
activities = [
|
|
'Dormire',
|
|
'Guida Auto',
|
|
'Work',
|
|
'Cesso',
|
|
'Giardino',
|
|
'Cena/Pranzo',
|
|
'Cucina',
|
|
'Allenamento',
|
|
'Messa',
|
|
'Duolinguo',
|
|
'Online',
|
|
'SelfStudy',
|
|
]
|
|
save_to_json_file(Activities_File, activities)
|
|
# for i in range(99):
|
|
# activities.append(f'Act {i}')
|
|
connections: List[WebSocket] = []
|
|
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def get_homepage(request: Request, u: str = 'Nobody'):
|
|
return templates.TemplateResponse("index2.html", {"request": request, 'user': u})
|
|
|
|
|
|
@app.get("/api/init")
|
|
async def get_initial_data():
|
|
return JSONResponse(content={
|
|
"current": current_activity,
|
|
"activities": activities
|
|
})
|
|
|
|
|
|
@app.websocket("/ws/activity")
|
|
async def activity_websocket(websocket: WebSocket, u: str = 'Unk'):
|
|
await websocket.accept()
|
|
connections.append(websocket)
|
|
try:
|
|
while True:
|
|
data = await websocket.receive_json()
|
|
print(data)
|
|
if data.get("action") == "select":
|
|
now = time.time()
|
|
data['user'] = u
|
|
data['epoch'] = now
|
|
dump_data(data)
|
|
global current_activity
|
|
current_activity = data["activity"]
|
|
if current_activity not in activities:
|
|
activities.append(current_activity)
|
|
save_to_json_file(Activities_File, activities)
|
|
await broadcast({
|
|
"type": "activity_update",
|
|
"activity": current_activity
|
|
})
|
|
except WebSocketDisconnect:
|
|
connections.remove(websocket)
|
|
|
|
|
|
def dump_data(dt: dict):
|
|
user = dt.get('user', 'Ukn')
|
|
epoch = dt.get('epoch', 0.0)
|
|
human_epoch = time.strftime('%Y_%m_%d_%H%M%S', time.localtime(epoch))
|
|
outfile = Base_Log / f'Act_{user}_{int(1000.0 * epoch)}_{human_epoch}.json'
|
|
save_to_json_file(outfile, dt)
|
|
|
|
|
|
async def broadcast(message: dict):
|
|
for conn in connections:
|
|
try:
|
|
await conn.send_json(message)
|
|
except Exception:
|
|
connections.remove(conn)
|
|
|
|
|
|
def main():
|
|
pass
|
|
|
|
|
|
if __name__ == '__main__':
|
|
uvicorn.run(app, host="127.0.0.1", port=8000)
|