space.py 2.5 KB

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