# -*- 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, EquipmentInstructionsResponse 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