Format fixes

app/api/routers/devices.py

@@ -1,277 +1,222 @@
 import numpy as np
-from fastapi import APIRouter, Depends, Query
+from fastapi import APIRouter, Depends
 from loguru import logger
 from sqlalchemy.orm import Session
 
 import app.models.domain.devices as domain_devices
 from app.api.dependencies.db import get_db
 from app.controllers.equipment.ahu.basic import build_acatah_freq_set
-from app.controllers.equipment.ahu.supply_air_temperature_set import build_acatah_supply_air_temperature_set
-from app.controllers.equipment.ahu.supply_air_temperature_set import get_next_supply_air_temperature_set
+from app.controllers.equipment.ahu.supply_air_temperature_set import (
+    build_acatah_supply_air_temperature_set,
+)
 from app.controllers.equipment.ahu.switch import build_acatah_switch_set
-from app.controllers.equipment.ahu.thermal_mode import get_thermal_mode, build_acatah_thermal_mode_set
+from app.controllers.equipment.ahu.thermal_mode import build_acatah_thermal_mode_set
 from app.controllers.equipment.fcu.basic import build_acatfc_instructions
-from app.controllers.equipment.fcu.early_start import build_acatfc_early_start_prediction
-from app.controllers.equipment.fcu.early_start import get_recommended_early_start_time
-from app.controllers.equipment.pau.freq_set import get_next_acatfu_freq_set, build_acatfu_freq_set
-from app.controllers.equipment.pau.supply_air_temperature_set import build_acatfu_supply_air_temperature
-from app.controllers.equipment.pau.supply_air_temperature_set import get_next_acatfu_supply_air_temperature_set
-from app.controllers.equipment.pau.switch import get_switch_action, build_acatfu_switch_set
+from app.controllers.equipment.fcu.early_start import (
+    build_acatfc_early_start_prediction,
+)
+from app.controllers.equipment.pau.freq_set import build_acatfu_freq_set
+from app.controllers.equipment.pau.supply_air_temperature_set import (
+    build_acatfu_supply_air_temperature,
+)
+from app.controllers.equipment.pau.switch import build_acatfu_switch_set
 from app.controllers.equipment.vav import build_acatva_instructions
 from app.controllers.equipment.ventilation_fan.switch import build_acvtsf_switch_set
 from app.controllers.equipment.vrf.basic import build_acatvi_instructions
-from app.models.domain.devices import DevicesInstructionsBaseResponse, DevicesEarlyStartTime
 
 router = APIRouter()
 
 
-@router.get('/instructions/acatah/thermal-mode-set', response_model=DevicesInstructionsBaseResponse)
-async def get_acatah_thermal_mode_set(
-        project_id: str = Query(..., max_length=50, regex='^Pj', alias='projectId'),
-        device_id: str = Query(..., max_length=50, regex='^Eq', alias='equipId')
+@router.post(
+    "/instructions/acatah/thermal-mode-set",
+    response_model=domain_devices.ACATAHThermalModeSetResponse,
+)
+async def get_acatah_thermal_mode_set_v2(
+    params: domain_devices.ACATAHThermalModeSetRequest,
 ):
-    thermal_mode = await get_thermal_mode(project_id, device_id)
-
-    return {
-        'projectId': project_id,
-        'equipId': device_id,
-        'output': {
-            'thermalModeSet': thermal_mode
-        }
-    }
-
-
-@router.post('/instructions/acatah/thermal-mode-set', response_model=domain_devices.ACATAHThermalModeSetResponse)
-async def get_acatah_thermal_mode_set_v2(params: domain_devices.ACATAHThermalModeSetRequest):
     thermal_mode = build_acatah_thermal_mode_set(params)
 
-    logger.info(f'{params.device_id}: thermal mode set - {thermal_mode}')
-    resp = {'thermal_mode_set': thermal_mode}
+    logger.info(f"{params.device_id}: thermal mode set - {thermal_mode}")
+    resp = {"thermal_mode_set": thermal_mode}
 
     return resp
 
 
-@router.get('/instructions/acatah/supply-air-temperature-set', response_model=DevicesInstructionsBaseResponse)
-async def get_acatah_supply_air_temperature_set(
-        project_id: str = Query(..., max_length=50, regex='^Pj', alias='projectId'),
-        device_id: str = Query(..., max_length=50, regex='^Eq', alias='equipId')
-):
-    next_supply_air_temperature_set = await get_next_supply_air_temperature_set(project_id, device_id)
-
-    return {
-        'projectId': project_id,
-        'equipId': device_id,
-        'output': {
-            'supplyAirTemperatureSet': next_supply_air_temperature_set
-        }
-    }
-
-
 @router.post(
-    '/instructions/acatah/supply-air-temperature-set',
-    response_model=domain_devices.ACATAHSupplyAirTempSetResponse
+    "/instructions/acatah/supply-air-temperature-set",
+    response_model=domain_devices.ACATAHSupplyAirTempSetResponse,
 )
-async def get_acatah_supply_air_temperature_set_v2(params: domain_devices.ACATAHSupplyAirTempSetRequest):
+async def get_acatah_supply_air_temperature_set_v2(
+    params: domain_devices.ACATAHSupplyAirTempSetRequest,
+):
     supply_air_temperature_set = build_acatah_supply_air_temperature_set(params)
 
     logger.info(supply_air_temperature_set)
 
-    resp = {'supply_air_temperature_set': supply_air_temperature_set}
+    resp = {"supply_air_temperature_set": supply_air_temperature_set}
 
     return resp
 
 
-@router.post('/instructions/acatah/freq-set', response_model=domain_devices.ACATAHFreqSetResponse)
+@router.post(
+    "/instructions/acatah/freq-set", response_model=domain_devices.ACATAHFreqSetResponse
+)
 async def get_acatah_freq_set(params: domain_devices.ACATAHFreqSetRequest):
     freq_set = await build_acatah_freq_set(params)
 
     logger.info(params)
-    logger.info(f'{params.device_id}: freq set - {freq_set}')
+    logger.info(f"{params.device_id}: freq set - {freq_set}")
 
-    resp = {'freq_set': freq_set}
+    resp = {"freq_set": freq_set}
 
     return resp
 
 
-@router.post('/instructions/acatah/switch-set', response_model=domain_devices.ACATAHSwitchSetResponse)
+@router.post(
+    "/instructions/acatah/switch-set",
+    response_model=domain_devices.ACATAHSwitchSetResponse,
+)
 async def get_acatah_switch_set(params: domain_devices.ACATAHSwitchSetRequest):
     switch_set = await build_acatah_switch_set(params)
 
     logger.info(params)
-    logger.info(f'{params.device_id}: switch set - {switch_set}')
+    logger.info(f"{params.device_id}: switch set - {switch_set}")
 
-    resp = {'switch_set': switch_set}
+    resp = {"switch_set": switch_set}
 
     return resp
 
 
-@router.post('/instructions/acvtsf/switch-set', response_model=domain_devices.ACVTSFSwitchSetResponse)
+@router.post(
+    "/instructions/acvtsf/switch-set",
+    response_model=domain_devices.ACVTSFSwitchSetResponse,
+)
 async def get_acvtsf_switch_set(params: domain_devices.ACVTSFSwitchSetRequest):
     switch_set = await build_acvtsf_switch_set(params)
 
     logger.info(params)
-    logger.info(f'{params.device_id}: switch set - {switch_set}')
+    logger.info(f"{params.device_id}: switch set - {switch_set}")
 
-    resp = {'switch_set': switch_set}
+    resp = {"switch_set": switch_set}
 
     return resp
 
 
-@router.get('/instructions/acatfu/equip-switch-set', response_model=DevicesInstructionsBaseResponse)
-async def get_acatfu_equip_switch_set(
-        project_id: str = Query(..., max_length=50, regex='^Pj', alias='projectId'),
-        device_id: str = Query(..., max_length=50, regex='^Eq', alias='equipId')
-):
-    action = await get_switch_action(project_id, device_id)
-
-    return {
-        'projectId': project_id,
-        'equipId': device_id,
-        'output': {
-            'equipSwitchSet': action
-        }
-    }
-
-
-@router.post('/instructions/acatfu/switch-set', response_model=domain_devices.ACATFUSwitchSetResponse)
+@router.post(
+    "/instructions/acatfu/switch-set",
+    response_model=domain_devices.ACATFUSwitchSetResponse,
+)
 async def get_acatfu_switch_set(params: domain_devices.ACATFUSwitchSetRequest):
     switch_set = await build_acatfu_switch_set(params)
 
     logger.info(params)
-    logger.info(f'{params.device_id}: switch set - {switch_set}')
+    logger.info(f"{params.device_id}: switch set - {switch_set}")
 
-    resp = {'switch_set': switch_set}
+    resp = {"switch_set": switch_set}
 
     return resp
 
 
-@router.get('/instructions/acatfu/supply-air-temperature-set', response_model=DevicesInstructionsBaseResponse)
-async def get_acatfu_supply_air_temperature_set(
-        project_id: str = Query(..., max_length=50, regex='^Pj', alias='projectId'),
-        device_id: str = Query(..., max_length=50, regex='^Eq', alias='equipId')
-):
-    temperature_set = await get_next_acatfu_supply_air_temperature_set(project_id, device_id)
-
-    return {
-        'projectId': project_id,
-        'equipId': device_id,
-        'output': {
-            'supplyAirTemperatureSet': temperature_set
-        }
-    }
-
-
 @router.post(
-    '/instructions/acatfu/supply-air-temperature-set',
-    response_model=domain_devices.ACATFUSupplyAirTempSetResponse
+    "/instructions/acatfu/supply-air-temperature-set",
+    response_model=domain_devices.ACATFUSupplyAirTempSetResponse,
 )
-async def get_acatfu_supply_air_temperature_set_v2(params: domain_devices.ACATFUSupplyAirTempSetRequest):
+async def get_acatfu_supply_air_temperature_set_v2(
+    params: domain_devices.ACATFUSupplyAirTempSetRequest,
+):
     supply_air_temperature_set = build_acatfu_supply_air_temperature(params)
 
     logger.info(supply_air_temperature_set)
 
-    resp = {'supply_air_temperature_set': supply_air_temperature_set}
+    resp = {"supply_air_temperature_set": supply_air_temperature_set}
 
     return resp
 
 
-@router.get('/instructions/acatfu/freq-set', response_model=DevicesInstructionsBaseResponse)
-async def get_acatfu_freq_set(
-        project_id: str = Query(..., max_length=50, regex='^Pj', alias='projectId'),
-        device_id: str = Query(..., max_length=50, regex='^Eq', alias='equipId')
-):
-    freq_set = await get_next_acatfu_freq_set(project_id, device_id)
-
-    return {
-        'projectId': project_id,
-        'equipId': device_id,
-        'output': {
-            'fanFreqSet': freq_set
-        }
-    }
-
-
-@router.post('/instructions/acatfu/freq-set', response_model=domain_devices.ACATFUFreqSetResponse)
+@router.post(
+    "/instructions/acatfu/freq-set", response_model=domain_devices.ACATFUFreqSetResponse
+)
 async def get_acatfu_freq_set(params: domain_devices.ACATFUFreqSetRequest):
     freq_set = build_acatfu_freq_set(params)
 
-    logger.info(f'{params.device_id} - {freq_set}')
+    logger.info(f"{params.device_id} - {freq_set}")
 
-    resp = {'freq_set': freq_set}
+    resp = {"freq_set": freq_set}
 
     return resp
 
 
-@router.get('/early-start/prediction/acatfc', response_model=DevicesEarlyStartTime)
-async def get_acatfc_early_start_time(
-        project_id: str = Query(..., max_length=50, regex='^Pj', alias='projectId'),
-        space_id: str = Query(..., max_length=50, regex='^Sp', alias='spaceId'),
-        db: Session = Depends(get_db)
-):
-    minutes = await get_recommended_early_start_time(db, project_id, space_id)
-    return {
-        'projectId': project_id,
-        'spaceId': space_id,
-        'minutes': minutes
-    }
-
-
-@router.post('/prediction/acatfc/early-start', response_model=domain_devices.ACATFCEarlyStartPredictionResponse)
+@router.post(
+    "/prediction/acatfc/early-start",
+    response_model=domain_devices.ACATFCEarlyStartPredictionResponse,
+)
 async def get_acatfc_early_start_prediction(
-        params: domain_devices.ACATFCEarlyStartPredictionRequest,
-        db: Session = Depends(get_db)
+    params: domain_devices.ACATFCEarlyStartPredictionRequest,
+    db: Session = Depends(get_db),
 ):
     minutes = await build_acatfc_early_start_prediction(params, db)
 
     logger.info(params)
-    logger.info(f'{params.space_id} - {minutes}')
+    logger.info(f"{params.space_id} - {minutes}")
 
-    resp = {'minutes': minutes}
+    resp = {"minutes": minutes}
 
     return resp
 
 
-@router.post('/instructions/acatvi', response_model=domain_devices.ACATVIInstructionsResponse)
-async def get_acatvi_instructions(params: domain_devices.ACATVIInstructionsRequest):
-    instructions = await build_acatvi_instructions(params)
+@router.post(
+    "/instructions/acatvi", response_model=domain_devices.ACATVIInstructionsResponse
+)
+async def get_acatvi_instructions(
+    params: domain_devices.ACATVIInstructionsRequest, db: Session = Depends(get_db)
+):
+    instructions = await build_acatvi_instructions(params, db)
 
     logger.info(params)
-    logger.info(f'{params.device_id} - {instructions}')
+    logger.info(f"{params.device_id} - {instructions}")
 
     return instructions
 
 
-@router.post('/instructions/acatfc', response_model=domain_devices.ACATFCInstructionsResponse)
+@router.post(
+    "/instructions/acatfc", response_model=domain_devices.ACATFCInstructionsResponse
+)
 async def get_acatfc_instructions(params: domain_devices.ACATFCInstructionsRequest):
     instructions = await build_acatfc_instructions(params)
 
     logger.info(params)
-    logger.info(f'{params.device_id} - {instructions}')
+    logger.info(f"{params.device_id} - {instructions}")
 
     response = domain_devices.ACATFCInstructionsResponse(
         onOff=instructions.switch_set,
         speed=instructions.speed_set,
         temperature=instructions.temperature_set,
         mode=instructions.switch_set,
-        water=instructions.water_valve_switch_set
+        water=instructions.water_valve_switch_set,
     )
 
     return response
 
 
-@router.post('/instructions/acatva', response_model=domain_devices.ACATVAInstructionsResponse)
+@router.post(
+    "/instructions/acatva", response_model=domain_devices.ACATVAInstructionsResponse
+)
 async def get_acatva_instructions(params: domain_devices.ACATVAInstructionsRequest):
     instructions = await build_acatva_instructions(params)
 
     logger.info(params)
-    logger.info(f'{params.device_id} - {instructions}')
+    logger.info(f"{params.device_id} - {instructions}")
 
     resp = dict()
     if not np.isnan(instructions.supply_air_flow_set):
-        resp.update({'SupplyAirFlowSet': instructions.supply_air_flow_set})
+        resp.update({"SupplyAirFlowSet": instructions.supply_air_flow_set})
     if not np.isnan(instructions.virtual_realtime_temperature):
-        resp.update({'VirtualRealtimeTemperature': instructions.virtual_realtime_temperature})
+        resp.update(
+            {"VirtualRealtimeTemperature": instructions.virtual_realtime_temperature}
+        )
     if not np.isnan(instructions.virtual_temperature_target_set):
-        resp.update({'TargetTemperatureSet': instructions.virtual_temperature_target_set})
+        resp.update(
+            {"TargetTemperatureSet": instructions.virtual_temperature_target_set}
+        )
 
     return resp

app/crud/model_path/early_start.py

@@ -1,16 +1,29 @@
+from typing import Optional
 from sqlalchemy.orm import Session
 
 from app.crud.base import CRUDBase
 from app.models.ml_models_path.early_start import EarlyStartDTRModelPath
-from app.schemas.model_path.early_start import EarlyStartDTRModelPathCreate, EarlyStartDTRModelPathUpdate
+from app.schemas.model_path.early_start import (
+    EarlyStartDTRModelPathCreate,
+    EarlyStartDTRModelPathUpdate,
+)
 
 
 class CRUDModelPathEarlyStartDTR(
-    CRUDBase[EarlyStartDTRModelPath, EarlyStartDTRModelPathCreate, EarlyStartDTRModelPathUpdate]
+    CRUDBase[
+        EarlyStartDTRModelPath,
+        EarlyStartDTRModelPathCreate,
+        EarlyStartDTRModelPathUpdate,
+    ]
 ):
-
-    def get_path_by_device(self, db: Session, device_id: str) -> EarlyStartDTRModelPath:
-        return db.query(self.model).filter(EarlyStartDTRModelPath.device_id == device_id).first()
+    def get_path_by_device(
+        self, db: Session, device_id: str
+    ) -> Optional[EarlyStartDTRModelPath]:
+        return (
+            db.query(self.model)
+            .filter(EarlyStartDTRModelPath.device_id == device_id)
+            .first()
+        )
 
 
 model_path_early_start_dtr = CRUDModelPathEarlyStartDTR(EarlyStartDTRModelPath)

app/db/base_class.py

@@ -10,5 +10,5 @@ class Base:
     # Generate __tablename__ automatically
 
     @declared_attr
-    def __tablename__(self) -> str:
-        return self.__name__.lower()
+    def __tablename__(cls) -> str:
+        return cls.__name__.lower()