# -*- coding: utf-8 -*- from typing import List from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session from app.api.dependencies.db import get_db from app.controllers.equipment.fcu.q_learning import QLearningCommandBuilder from app.crud.space.weight import get_weights_by_space, get_weights_by_vav, create_weight, update_weight from app.models.domain.space import SpaceControlResponse from app.schemas.sapce_weight import SpaceWeight, SpaceWeightCreate, SpaceWeightUpdate from app.services.transfer import Season router = APIRouter() @router.get('/control', response_model=SpaceControlResponse) async def get_space_command( project_id: str = Query(..., max_length=50, regex='^Pj', alias='projectId'), space_id: str = Query(..., max_length=50, regex='^Sp', alias='roomId'), timestamp: str = Query(None, min_length=14, max_length=14, alias='time'), method: int = Query(3), ): response = { 'projectId': project_id, 'roomId': space_id, 'flag': 1, 'time': timestamp, 'method': method, } return response @router.get('/test') async def get_test_result(current: float, pre: float, target: float): builder = QLearningCommandBuilder(Season('Cooling')) command = await builder.get_command(current, pre, target) return command @router.post('/weight', response_model=SpaceWeight) async def create_space_weight(weight: SpaceWeightCreate, db: Session = Depends(get_db)): return create_weight(db=db, weight=weight) @router.get('/weight/{space_id}', response_model=List[SpaceWeight]) async def read_weight_by_space(space_id: str, db: Session = Depends(get_db)): db_weights = get_weights_by_space(db, space_id=space_id) return db_weights @router.get('/weight/vav/{vav_id}', response_model=List[SpaceWeight]) async def read_weight_by_vav(vav_id: str, db: Session = Depends(get_db)): db_weights = get_weights_by_vav(db, vav_id=vav_id) return db_weights @router.put('/weight/{space_id}/{vav_id}', response_model=SpaceWeight) async def update_weight_by_space( space_id: str, vav_id: str, weight_in: SpaceWeightUpdate, db: Session = Depends(get_db) ): weights = get_weights_by_space(db, space_id=space_id) new_weight = None for weight in weights: if weight.vav_box_id == vav_id: new_weight = update_weight(db, db_weight=weight, weight_in=weight_in) break if not new_weight: raise HTTPException(status_code=404, detail='Wight not found') return new_weight