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

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

from operator import attrgetter

import arrow
import numpy as np

from app.api.errors.iot import MissingIOTDataError
from app.models.domain.devices import ASHPRequest
from app.schemas.equipment import ASHP
from app.schemas.season import Season


class ASHPGroupController:
    def __init__(self, ashp_list: list[ASHP], outdoor_temp_list: list[float], season: Season):
        self._season = season
        self._ashp_list = ashp_list
        self._outdoor_temp_list = outdoor_temp_list
        self._interval = 0
        self._is_warning = False

    def initialization(self) -> tuple[float, float]:
        if self._season == Season.cooling:
            if self._outdoor_temp_list:
                outdoor_temp_avg = np.mean(self._outdoor_temp_list)
            else:
                outdoor_temp_avg = 22.0

            total_device_num = len(self._ashp_list)
            if outdoor_temp_avg <= 20.0:
                device_num = 1
                out_temp_set = 10.0
                interval = 45 * 60
            elif outdoor_temp_avg < 24.0:
                device_num = total_device_num // 3
                out_temp_set = 9.0
                interval = 0
            else:
                device_num = total_device_num // 3 + 1
                out_temp_set = 8.5
                interval = 0

            if self._outdoor_temp_list:
                if np.max(self._outdoor_temp_list) > 34.0:
                    device_num += 1
                    device_num = min(device_num, total_device_num - 1)
        elif self._season == Season.heating:
            if self._outdoor_temp_list:
                outdoor_temp_avg = np.mean(self._outdoor_temp_list)
            else:
                outdoor_temp_avg = -10.0

            total_device_num = len(self._ashp_list)
            if outdoor_temp_avg >= 0.0:
                device_num = 1
                out_temp_set = 45.0
                interval = 45 * 60
            elif outdoor_temp_avg > -15.0:
                device_num = total_device_num // 3
                out_temp_set = 45.0
                interval = 0
            else:
                device_num = total_device_num // 3 + 1
                out_temp_set = 50.0
                interval = 0

            if self._outdoor_temp_list:
                if np.min(self._outdoor_temp_list) < -15.0:
                    device_num += 1
                    device_num = min(device_num, total_device_num - 1)
        else:
            device_num = 0
            interval = 0
            out_temp_set = np.NAN

        self._interval = interval

        return device_num, out_temp_set

    def device_num_adjustment(self) -> tuple[float, float]:
        out_temp_15_avg = np.mean(
            [np.mean(device.out_temp_15min) for device in self._ashp_list if device.running_status])
        out_temp_set_15_avg = np.mean(
            [np.mean(device.out_temp_set_15min) for device in self._ashp_list if device.running_status])
        iplr_15_avg = np.mean([np.mean(device.iplr_15min) for device in self._ashp_list if device.running_status])

        out_temp_30_avg = np.mean(
            [np.mean(device.out_temp_30min) for device in self._ashp_list if device.running_status])
        out_temp_set_30_avg = np.mean(
            [np.mean(device.out_temp_set_30min) for device in self._ashp_list if device.running_status])
        iplr_30_avg = np.mean([np.mean(device.iplr_30min) for device in self._ashp_list if device.running_status])

        device_num_diff = 0
        if self._season == Season.cooling:
            if out_temp_15_avg < out_temp_set_15_avg - 1 and iplr_15_avg > 0.85:
                device_num_diff = 1
            if out_temp_30_avg > out_temp_set_30_avg + 1 and iplr_30_avg <= 0.85:
                device_num_diff = 1
                self._is_warning = True

            if out_temp_15_avg >= out_temp_set_15_avg and iplr_15_avg < 0.6:
                device_num_diff = -1
        elif self._season == Season.heating:
            if out_temp_15_avg > out_temp_set_15_avg + 1 and iplr_15_avg > 0.85:
                device_num_diff = 1
            if out_temp_30_avg > out_temp_set_30_avg + 1 and iplr_30_avg <= 0.85:
                device_num_diff = 1
                self._is_warning = True

            if out_temp_15_avg <= out_temp_set_15_avg + 0.3 and iplr_15_avg < 0.6:
                device_num_diff = -1

        return device_num_diff, float(out_temp_set_30_avg)

    def out_temp_set_adjustment(self) -> float:
        out_temp_set = np.NAN
        if self._outdoor_temp_list:
            current_outdoor_temp = self._outdoor_temp_list[-1]
            if self._season == Season.cooling:
                if current_outdoor_temp >= 30.0:
                    out_temp_set = 8.0
                elif current_outdoor_temp >= 28.0:
                    out_temp_set = 8.5
                elif current_outdoor_temp >= 26.0:
                    out_temp_set = 9.0
                elif current_outdoor_temp > 22.0:
                    out_temp_set = 10.0
                else:
                    out_temp_set = 11.0
            elif self._season == Season.heating:
                if current_outdoor_temp > 10.0:
                    out_temp_set = 40.0
                elif current_outdoor_temp >= 0.0:
                    out_temp_set = 45.0
                elif current_outdoor_temp > -10.0:
                    out_temp_set = 50.0
                else:
                    out_temp_set = 55.0

        return out_temp_set

    def build_up(self, diff: float, out_temp_set: float):
        if diff > 0:
            self._ashp_list = sorted(self._ashp_list, key=attrgetter("acc_run_time"))
            for device in self._ashp_list:
                if not device.running_status:
                    device.equip_switch_set = True
                    diff -= 1
                    if diff == 0:
                        break
        elif diff < 0:
            self._ashp_list = sorted(self._ashp_list, key=attrgetter("acc_run_time"), reverse=True)
            for device in self._ashp_list:
                if device.running_status:
                    device.equip_switch_set = False
                    diff += 1
                    if diff == 0:
                        break

        if not np.isnan(out_temp_set):
            for device in self._ashp_list:
                if device.equip_switch_set:
                    device.out_temp_set = out_temp_set

    def turn_off_all(self):
        for device in self._ashp_list:
            device.equip_switch_set = False

    def antifreeze(self):
        pass

    def run(self):
        t = arrow.utcnow()
        on_time = arrow.get(self._ashp_list[0].on_time, 'HHmmss')
        off_time = arrow.get(self._ashp_list[0].off_time, 'HHmmss')
        if off_time.hour > on_time.hour:
            if not on_time.hour < t.hour < off_time.hour:
                self.turn_off_all()
                return
        else:
            if off_time.hour < t.hour < on_time.hour:
                self.turn_off_all()
                return

        if t.minute < 30:
            if t.hour == 1:
                if not np.any([device.running_status for device in self._ashp_list]):  # no device is running
                    device_num_diff, out_temp_set = self.initialization()
                    self.build_up(device_num_diff, out_temp_set)
            else:
                device_num_diff, out_temp_set = self.device_num_adjustment()
                self.build_up(device_num_diff, out_temp_set)
        else:
            out_temp_set = self.out_temp_set_adjustment()
            self.build_up(0, out_temp_set)

    def get_results(self) -> tuple[list[ASHP], float, bool]:
        return self._ashp_list, self._interval, self._is_warning


async def build_ashp_instructions(params: ASHPRequest) -> dict[str, list | float | bool]:
    controller = ASHPGroupController(params.device_list, params.outdoor_temp, params.season)
    try:
        controller.run()
    except (TypeError, IndexError):
        raise MissingIOTDataError
    device_list, interval, is_warning = controller.get_results()
    instructions = {
        "device_list": device_list,
        "interval": interval,
        "is_warning": is_warning
    }

    return instructions