1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495 |
- # -*- 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_v1, get_vav_control_v2
- from app.models.domain.equipment import EquipmentControlResponse, EquipmentControlRequest
- 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')
- ):
- 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_v1(project_id, equip_id)
- output = {
- 'SupplyAirFlowSet': vav.supply_air_flow_set,
- 'VirtualRealtimeTemperature': vav.virtual_realtime_temperature,
- 'TargetTemperatureSet': vav.virtual_target_temperature
- }
- 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):
- 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_v1(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
|