123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119 |
- # -*- coding: utf-8 -*-
- from enum import Enum
- import numpy as np
- from fastapi import APIRouter, Depends, Query
- from loguru import logger
- from sqlalchemy.orm import Session
- from app.api.dependencies.db import get_db
- from app.controllers.equipment.fcu.basic import get_fcu_control_result
- from app.controllers.equipment.vav import get_vav_control_v2
- from app.models.domain.equipment import (
- EquipmentControlResponse,
- EquipmentControlRequest,
- )
- from app.schemas.equipment import PAU
- from app.utils.date import get_time_str
- class EquipmentName(str, Enum):
- FCU = "ACATFC"
- VAV = "ACATVA"
- router = APIRouter()
- @router.get("/control", response_model=EquipmentControlResponse)
- async def get_equipment_command_test(
- project_id: str = Query(..., max_length=50, regex="^Pj", alias="projectId"),
- equip_id: str = Query(..., max_length=50, regex="^Eq", alias="equipId"),
- equip_type: EquipmentName = Query(..., alias="equipType"),
- db: Session = Depends(get_db),
- ):
- if equip_type.value == EquipmentName.FCU:
- output = await get_fcu_control_result(project_id, equip_id)
- elif equip_type.value == EquipmentName.VAV:
- vav = await get_vav_control_v2(db, project_id, equip_id)
- output = {
- "SupplyAirFlowSet": vav.supply_air_flow_set,
- "VirtualRealtimeTemperature": vav.virtual_realtime_temperature
- if not np.isnan(vav.virtual_realtime_temperature)
- else None,
- "TargetTemperatureSet": vav.virtual_target_temperature
- if not np.isnan(vav.virtual_target_temperature)
- else None,
- }
- else:
- output = {}
- response = {
- "projectId": project_id,
- "equipId": equip_id,
- "time": get_time_str(),
- "output": output,
- }
- return response
- @router.post("/control", response_model=EquipmentControlResponse)
- async def get_equipment_command(
- equipment_control_info: EquipmentControlRequest, db: Session = Depends(get_db)
- ):
- if equipment_control_info.equipType == EquipmentName.FCU:
- output = await get_fcu_control_result(
- equipment_control_info.projectId, equipment_control_info.equipId
- )
- elif equipment_control_info.equipType == EquipmentName.VAV:
- vav = await get_vav_control_v2(
- db, equipment_control_info.projectId, equipment_control_info.equipId
- )
- output = {
- "SupplyAirFlowSet": vav.supply_air_flow_set,
- "VirtualRealtimeTemperature": vav.virtual_realtime_temperature
- if not np.isnan(vav.virtual_realtime_temperature)
- else None,
- "TargetTemperatureSet": vav.virtual_target_temperature
- if not np.isnan(vav.virtual_target_temperature)
- else None,
- }
- else:
- output = {}
- response = {
- "projectId": equipment_control_info.projectId,
- "equipId": equipment_control_info.equipId,
- "time": get_time_str(),
- "output": output,
- }
- logger.info(response)
- return response
- @router.post("/control/atva/v2", response_model=EquipmentControlResponse)
- async def get_fcu_command(
- equipment_control_info: EquipmentControlRequest, db: Session = Depends(get_db)
- ):
- vav = await get_vav_control_v2(
- db, equipment_control_info.projectId, equipment_control_info.equipId
- )
- output = {
- "SupplyAirFlowSet": vav.supply_air_flow_set,
- "VirtualRealtimeTemperature": vav.virtual_realtime_temperature
- if not np.isnan(vav.virtual_realtime_temperature)
- else None,
- "TargetTemperatureSet": vav.virtual_target_temperature
- if not np.isnan(vav.virtual_target_temperature)
- else None,
- }
- response = {
- "projectId": equipment_control_info.projectId,
- "equipId": equipment_control_info.equipId,
- "time": get_time_str(),
- "output": output,
- }
- logger.info(response)
- return response
|