space.py 2.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980
  1. # -*- coding: utf-8 -*-
  2. from typing import List
  3. from fastapi import APIRouter, Depends, HTTPException, Query
  4. from sqlalchemy.orm import Session
  5. from app.api.dependencies.db import get_db
  6. from app.controllers.equipment.fcu.q_learning import QLearningCommandBuilder
  7. from app.crud.space.weight import get_weights_by_space, get_weights_by_vav, create_weight, update_weight
  8. from app.models.domain.space import SpaceControlResponse
  9. from app.schemas.sapce_weight import SpaceWeight, SpaceWeightCreate, SpaceWeightUpdate
  10. from app.services.transfer import Season
  11. router = APIRouter()
  12. @router.get('/control', response_model=SpaceControlResponse)
  13. async def get_space_command(
  14. project_id: str = Query(..., max_length=50, regex='^Pj', alias='projectId'),
  15. space_id: str = Query(..., max_length=50, regex='^Sp', alias='roomId'),
  16. timestamp: str = Query(None, min_length=14, max_length=14, alias='time'),
  17. method: int = Query(3),
  18. ):
  19. if project_id == 'Pj1101080259':
  20. flag = 0
  21. else:
  22. flag = 1
  23. response = {
  24. 'projectId': project_id,
  25. 'roomId': space_id,
  26. 'flag': flag,
  27. 'time': timestamp,
  28. 'method': method,
  29. }
  30. return response
  31. @router.get('/test')
  32. async def get_test_result(current: float, pre: float, target: float):
  33. builder = QLearningCommandBuilder(Season('Cooling'))
  34. command = await builder.get_command(current, pre, target)
  35. return command
  36. @router.post('/weight', response_model=SpaceWeight)
  37. async def create_space_weight(weight: SpaceWeightCreate, db: Session = Depends(get_db)):
  38. return create_weight(db=db, weight=weight)
  39. @router.get('/weight/{space_id}', response_model=List[SpaceWeight])
  40. async def read_weight_by_space(space_id: str, db: Session = Depends(get_db)):
  41. db_weights = get_weights_by_space(db, space_id=space_id)
  42. return db_weights
  43. @router.get('/weight/vav/{vav_id}', response_model=List[SpaceWeight])
  44. async def read_weight_by_vav(vav_id: str, db: Session = Depends(get_db)):
  45. db_weights = get_weights_by_vav(db, vav_id=vav_id)
  46. return db_weights
  47. @router.patch('/weight/{space_id}/{vav_id}', response_model=SpaceWeight)
  48. async def update_weight_by_space(
  49. space_id: str,
  50. vav_id: str,
  51. weight_in: SpaceWeightUpdate,
  52. db: Session = Depends(get_db)
  53. ):
  54. weights = get_weights_by_space(db, space_id=space_id)
  55. new_weight = None
  56. for weight in weights:
  57. if weight.vav_box_id == vav_id:
  58. new_weight = update_weight(db, db_weight=weight, weight_in=weight_in)
  59. break
  60. if not new_weight:
  61. raise HTTPException(status_code=404, detail='Wight not found')
  62. return new_weight