env_regulation_api/app/controllers/equipment/ahu/basic.py

# -*- coding: utf-8 -*-

from typing import Dict, List, Optional
import numpy as np
from httpx import AsyncClient
from loguru import logger

from app.controllers.events import ahu_supply_air_temp_set_dict
from app.schemas.equipment import AHU
from app.schemas.system import ACAT
from app.services.platform import DataPlatformService, InfoCode
from app.services.transfer import ReviewService, EquipmentInfoService
from app.services.weather import WeatherService


class AHUController:

    def __init__(self, equipment: Optional[AHU] = None, system: Optional[ACAT] = None):
        super(AHUController, self).__init__()
        self._equipment = equipment
        self._system = system
        self._supply_air_temperature_command_dict = ahu_supply_air_temp_set_dict.get('dataframe')

    async def build_freq_set(self, supply_air_temperature_set_duration: List, hot_rate: float) -> float:
        extent = 5
        if np.isnan(self._equipment.fan_freq_set)\
                or np.isnan(self._system.supply_static_press)\
                or np.isnan(self._system.supply_static_press_set):
            temp_freq_set = 80.0
        else:
            pre_fan_freq_set = self._equipment.fan_freq_set

            if self._system.supply_static_press < self._system.supply_static_press_set - extent:
                temp_freq_set = pre_fan_freq_set + 2
            elif self._system.supply_static_press > self._system.supply_static_press_set + extent:
                temp_freq_set = pre_fan_freq_set - 2
            else:
                temp_freq_set = pre_fan_freq_set

        temperature_value_list = np.array([item['value'] for item in supply_air_temperature_set_duration])
        if temperature_value_list.size > 0\
                and np.all(temperature_value_list == temperature_value_list[0])\
                and temperature_value_list[0] <= 18.0\
                and hot_rate >= 0.5:
            freq_set = 100.0
        else:
            freq_set = min(temp_freq_set, 90.0)

        return freq_set

    async def build_supply_air_temperature_set(self, outdoor_temperature: float) -> float:
        command_df = self._supply_air_temperature_command_dict
        temp_command = command_df.loc[command_df['outdoor_temperature'] == outdoor_temperature]['temperature_set']
        if len(temp_command) > 0:
            command = float(temp_command)
        else:
            command = 20.0

        return command


@logger.catch()
async def get_freq_controlled(project_id: str, equipment_id: str) -> None:
    async with AsyncClient() as client:
        platform = DataPlatformService(client, project_id)
        review_service = ReviewService(client, project_id)
        # equip_service = EquipmentInfoService(client, project_id)

        fan_freq_set = await platform.get_realtime_fan_freq_set(equipment_id)
        # system = await equip_service.get_system_by_equipment(equipment_id)
        system = ['Sy1101050030eab54ad66b084a1cb1b919950b263280']
        press = await platform.get_realtime_supply_static_press(system[0])
        press_set = await platform.get_realtime_supply_static_press_set(system[0])
        hot_rate = await review_service.get_fill_rate()
        supply_air_temperature_set = await platform.get_duration(
            InfoCode.supply_air_temperature_set,
            equipment_id,
            30 * 60
        )

    equip_params = {
        'id': equipment_id,
        'fan_freq_set': fan_freq_set
    }
    ahu = AHU(**equip_params)

    system_params = {
        'id': system[0],
        'supply_static_press': press,
        'supply_static_press_set': press_set
    }
    acat_system = ACAT(**system_params)

    ahu_controller = AHUController(ahu, acat_system)
    new_freq_set = await ahu_controller.build_freq_set(supply_air_temperature_set, hot_rate)
    logger.debug(f'freq set: {new_freq_set}')

    async with AsyncClient() as client:
        platform = DataPlatformService(client, project_id)
        await platform.set_code_value(equipment_id, InfoCode.fan_freq_set, new_freq_set)


@logger.catch()
async def get_supply_air_temperature_controlled(project_id: str, equipment_id: str) -> None:
    async with AsyncClient() as client:
        weather_service = WeatherService(client)
        realtime_weather = await weather_service.get_realtime_weather(project_id)
        outdoor_temperature = realtime_weather.get('temperature')

    ahu_controller = AHUController()
    new_supply_air_temperature_set = await ahu_controller.build_supply_air_temperature_set(outdoor_temperature)
    logger.debug(f'supply air temperature set: {new_supply_air_temperature_set}')

    async with AsyncClient() as client:
        platform = DataPlatformService(client, project_id)
        await platform.set_code_value(equipment_id, InfoCode.supply_air_temperature_set, new_supply_air_temperature_set)