from datetime import datetime from typing import Dict, Optional, Tuple import arrow import numpy as np from loguru import logger from sqlalchemy.orm import Session from sqlalchemy.exc import SQLAlchemyError from app.controllers.equipment.controller import EquipmentController from app.crud.device.device import device from app.crud.device.status_timestamp import blowy_feedback_time, high_speed_time from app.models.domain.devices import ACATVIInstructionsRequest from app.models.domain.feedback import FeedbackValue from app.schemas.device.device import DeviceCreate from app.schemas.equipment import VRF from app.schemas.device.status_timestamp import ( BlowyFeedbackTimeCreate, BlowyFeedbackTimeUpdate, HighSpeedTimeCreate, HighSpeedTimeUpdate, ) from app.utils.math import round_half_up class VRFController(EquipmentController): def __init__(self, device: VRF, target: float, realtime: float): super(VRFController, self).__init__() self.device = device self.target = target self.realtime = realtime def get_switch_set(self) -> str: if np.isnan(self.target): switch_set = "off" else: switch_set = "on" self.device.equip_switch_set = switch_set return switch_set def get_mode_set(self) -> str: if self.target > self.realtime: mode_set = "cooling" else: mode_set = "cooling" self.device.mode_set = mode_set return mode_set def get_temperature_set(self) -> float: if self.realtime >= 24.0: diff = self.device.return_air_temp - self.realtime if diff > 0: new_temperature_set = self.target + diff elif diff <= 0: new_temperature_set = self.target else: if np.isnan(self.device.current_temperature_set): new_temperature_set = self.realtime else: new_temperature_set = self.device.current_temperature_set elif self.realtime <= 22.0: new_temperature_set = 28.0 else: if np.isnan(self.device.current_temperature_set): new_temperature_set = 25.0 else: new_temperature_set = self.device.current_temperature_set new_temperature_set = round_half_up(new_temperature_set) self.device.temperature_set = new_temperature_set return self.device.temperature_set def get_speed_set(self) -> str: if self.realtime >= 24.0: speed = "M" elif self.realtime <= 22.0: speed = "LL" elif self.device.return_air_temp - self.target >= 3.0: speed = "HH" elif self.device.return_air_temp - self.target >= 1.0: speed = "M" else: if self.device.speed == "missing": speed = "M" else: speed = self.device.speed self.device.speed_set = speed return speed async def run(self): if self.get_switch_set() == "on": self.get_speed_set() self.get_mode_set() self.get_temperature_set() if self.device.temperature_set == self.device.current_temperature_set: self.device.temperature_set = 0.0 if self.device.speed_set == self.device.speed: self.device.speed_set = "hold" self.device.equip_switch_set = "hold" if not self.device.running_status: self.device.equip_switch_set = "on" self.get_speed_set() else: if not self.device.running_status: self.device.equip_switch_set = "hold" def get_results(self): return self.device class VRFController1(VRFController): def __init__( self, device: VRF, target: float, realtime: float, feedback: FeedbackValue, blow_feedback_time: datetime, high_speed_time: datetime, ): super(VRFController1, self).__init__(device, target, realtime) self.feedback = feedback self.blow_feedback_time = blow_feedback_time self.high_speed_time = high_speed_time def get_temperature_set(self) -> float: new_temperature_set = self.device.current_temperature_set # Default temperature set. if not self.device.running_status: new_temperature_set = self.target # lower limit. if self.realtime <= 22.0: new_temperature_set = 28.0 # recovery # if self.device.temperature_set == 28.0 and self.realtime > 25.0: # new_temperature_set = self.target # feedback if ( self.feedback == FeedbackValue.a_little_cold or self.feedback == FeedbackValue.so_cold ): if self.feedback == FeedbackValue.so_cold and self.device.speed == "LL": new_temperature_set = self.device.current_temperature_set + 2.0 else: new_temperature_set = self.device.current_temperature_set + 1.0 if self.device.return_air_temp <= self.device.current_temperature_set: if self.feedback == FeedbackValue.so_cold: new_temperature_set = self.device.current_temperature_set + 2.0 elif ( self.feedback == FeedbackValue.a_little_hot or self.feedback == FeedbackValue.so_hot ): if self.feedback == FeedbackValue.so_hot and self.device.speed == "HH": new_temperature_set = self.device.current_temperature_set - 2.0 else: new_temperature_set = self.device.current_temperature_set - 1.0 new_temperature_set = round_half_up(new_temperature_set) self.device.temperature_set = new_temperature_set return new_temperature_set def get_speed_set(self) -> str: new_speed = self.device.speed # Default speed set: if not self.device.running_status: new_speed = "M" # Lower limit. if self.realtime <= 22.0: new_speed = "LL" # Feedback. if self.feedback == FeedbackValue.so_cold: if self.device.return_air_temp > self.device.current_temperature_set: if self.device.speed == "HH": new_speed = "M" elif self.device.speed == "M": new_speed = "LL" elif self.feedback == FeedbackValue.so_hot: if self.device.speed == "LL": new_speed = "M" elif self.device.speed == "M": new_speed = "HH" elif self.feedback == FeedbackValue.noisy_or_blowy: if self.device.speed == "HH": new_speed = "M" elif self.device.speed == "M": new_speed = "LL" else: new_speed = "LL" a_now = arrow.utcnow() # Restriction. if a_now.date() == arrow.get(self.blow_feedback_time).date(): if new_speed == "HH": new_speed = "M" # Recovery. if (a_now - arrow.get(self.high_speed_time)).seconds > 60 * 60: if new_speed == "HH": new_speed = "M" self.device.speed_set = new_speed return new_speed async def query_status_time(db: Session, device_id: str) -> Tuple[datetime, datetime]: feedback_time_in_db = blowy_feedback_time.get_time_by_device(db, device_id) if feedback_time_in_db: feedback_time = feedback_time_in_db.timestamp else: past = arrow.utcnow().shift(hours=-24) feedback_time = past.naive if not device.get(db, device_id): device.create(db=db, obj_in=DeviceCreate(id=device_id)) blowy_feedback_time.create( db=db, obj_in=BlowyFeedbackTimeCreate( timestamp=feedback_time, device_id=device_id ), ) high_speed_time_in_db = high_speed_time.get_time_by_device(db, device_id) if high_speed_time_in_db: high_time = high_speed_time_in_db.timestamp else: past = arrow.utcnow().shift(hours=-24) high_time = past.naive if not device.get(db, device_id): device.create(db=db, obj_in=DeviceCreate(id=device_id)) high_speed_time.create( db=db, obj_in=HighSpeedTimeCreate(timestamp=high_time, device_id=device_id) ) return feedback_time, high_time @logger.catch() async def build_acatvi_instructions( params: ACATVIInstructionsRequest, db: Session ) -> Dict: target = ( np.NAN if params.space_temperature_target == -1.0 else params.space_temperature_target ) realtime = ( np.NAN if params.space_realtime_temperature == -1.0 else params.space_realtime_temperature ) return_air_temp = ( np.NAN if params.return_air_temperature == -1.0 else params.return_air_temperature ) current_temperature_set = ( np.NAN if params.current_temperature_set == -1.0 else round_half_up(params.current_temperature_set) ) vrf = VRF( return_air_temp=return_air_temp, current_temperature_set=current_temperature_set, speed=params.current_speed, running_status=params.running_status, ) act_blowy_feedback_time, act_high_speed_time = await query_status_time( db, params.device_id ) controller = VRFController1( vrf, target, realtime, params.feedback, act_blowy_feedback_time, act_high_speed_time, ) await controller.run() regulated_vrf = controller.get_results() if params.feedback == FeedbackValue.noisy_or_blowy: blowy_feedback_time_in_db = blowy_feedback_time.get_time_by_device( db=db, device_id=params.device_id ) if blowy_feedback_time_in_db: blowy_feedback_time.update( db=db, db_obj=blowy_feedback_time_in_db, obj_in=BlowyFeedbackTimeUpdate( timestamp=datetime.utcnow(), device_id=params.device_id ), ) else: blowy_feedback_time.create( db=db, obj_in=BlowyFeedbackTimeCreate( timestamp=datetime.utcnow(), device_id=params.device_id ), ) if regulated_vrf.speed_set == "HH": high_speed_time_in_db = high_speed_time.get_time_by_device( db=db, device_id=params.device_id ) if high_speed_time_in_db: high_speed_time.update( db=db, db_obj=high_speed_time_in_db, obj_in=HighSpeedTimeUpdate( timestamp=datetime.utcnow(), device_id=params.device_id ), ) else: high_speed_time.create( db=db, obj_in=HighSpeedTimeCreate( timestamp=datetime.utcnow(), device_id=params.device_id ), ) instructions = { "switch_set": regulated_vrf.equip_switch_set, "speed_set": regulated_vrf.speed_set, "temperature_set": regulated_vrf.temperature_set, "mode_set": regulated_vrf.mode_set, } return instructions