Sfoglia il codice sorgente

complete target readjust logic

chenhaiyang 4 anni fa
parent
commit
32bf81761a

+ 0 - 41
app/api/routers/items.py

@@ -1,41 +0,0 @@
-# -*- coding: utf-8 -*-
-
-from datetime import datetime, time, timedelta
-from typing import Optional
-from uuid import UUID
-
-from fastapi import APIRouter, HTTPException, Path, Query, Body
-
-router = APIRouter()
-
-
-@router.get('/{item_id}')
-async def read_items(
-    item_id: int = Path(..., title='The ID of the item to get'),
-    q: Optional[str] = Query(None, alias='item-query'),
-):
-    results = {'item_id': item_id}
-    if q:
-        results.update({'q': q})
-    return results
-
-
-@router.put('/{item_id}')
-async def update_item(
-        item_id: UUID,
-        start_datetime: Optional[datetime] = Body(None),
-        end_datetime: Optional[datetime] = Body(None),
-        repeat_at: Optional[time] = Body(None),
-        process_after: Optional[timedelta] = Body(None),
-):
-    start_process = start_datetime + process_after
-    duration = end_datetime - start_process
-    return {
-        "item_id": item_id,
-        "start_datetime": start_datetime,
-        "end_datetime": end_datetime,
-        "repeat_at": repeat_at,
-        "process_after": process_after,
-        "start_process": start_process,
-        "duration": duration,
-    }

+ 39 - 5
app/api/routers/targets.py

@@ -1,17 +1,51 @@
 # -*- coding: utf-8 -*-
 
-from fastapi import APIRouter, HTTPException, Path, Query
+from fastapi import APIRouter, HTTPException, Query, Depends
+from loguru import logger
 
+from app.controllers.targets import readjust_all_target
 from app.models.targets import TargetReadjustInResponse
+from app.utils.date import get_time_str
 
 router = APIRouter()
 
 
-@router.get('/readjust', response_model=TargetReadjustInResponse)
+@router.get('/readjust', response_model=TargetReadjustInResponse, tags=['targets'])
 async def readjust_target(
         project_id: str = Query(..., max_length=50, regex='^Pj'),
         space_id: str = Query(..., max_length=50, regex='^Sp'),
-        wechat_timestamp: str = Query(None, min_length=8, max_length=8),
+        wechat_timestamp: str = Query(None, min_length=14, max_length=14),
 ):
-    # TODO
-    return
+    try:
+        need_run_room_control = await readjust_all_target(project_id, space_id, wechat_timestamp)
+    except Exception as e:
+        logger.error(e)
+        raise HTTPException(
+            status_code=500,
+            detail='Oops, something wrong has happened'
+        )
+
+    response = {
+        'projectId': project_id,
+        'roomId': space_id,
+        'flag': 1 if need_run_room_control else 0,
+        'time': wechat_timestamp,
+    }
+    return response
+
+
+@router.get('/readjust/test/{project_id}/{space_id}', tags=['targets'])
+async def test_readjust(project_id: str, space_id: str, wechat_time: str = Depends(get_time_str)):
+    feedback = {
+        'a little cold': 1,
+        'so cold': 0,
+        'a little hot': 1,
+        'so hot': 0,
+        'noisy or blowy': 0,
+        'so stuffy': 0,
+        'switch off': 0,
+        'switch on': 0,
+    }
+    await readjust_all_target(project_id, space_id, wechat_time, feedback)
+
+    return {'Message': 'Running background'}

+ 0 - 27
app/api/routers/users.py

@@ -1,27 +0,0 @@
-# -*- coding: utf-8 -*-
-
-from typing import Optional
-
-from fastapi import APIRouter, Depends
-
-router = APIRouter()
-
-
-async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
-    return {"q": q, "skip": skip, "limit": limit}
-
-
-@router.get('/users', tags=['users'])
-async def read_users(commons: dict = Depends(common_parameters)):
-    # return [{'username': 'Foo'}, {'username': 'Bar'}]
-    return commons
-
-
-@router.get('/user/me', tags=['users'])
-async def read_user_me():
-    return {'username': 'fake current user'}
-
-
-@router.get('/users/{username}', tags=['users'])
-async def read_user(username: str):
-    return {'username': username}

+ 12 - 0
app/controllers/controller.py

@@ -0,0 +1,12 @@
+# -*- coding: utf-8 -*-
+
+from abc import ABC
+
+
+class Controller(ABC):
+
+    def __init__(self):
+        self._results = {}
+
+    def get_results(self):
+        return self._results

+ 369 - 0
app/controllers/targets.py

@@ -1,3 +1,372 @@
 # -*- coding: utf-8 -*-
 
+from abc import abstractmethod
+from typing import Dict, Tuple, Optional
 
+import arrow
+import numpy as np
+import pandas as pd
+from httpx import AsyncClient
+from loguru import logger
+
+from app.controllers.controller import Controller
+from app.resources.params import (
+    TEMPERATURE_RELATED_FEEDBACK_WEIGHT,
+    CO2_RELATED_FEEDBACK_WEIGHT,
+    SWITCH_RELATED_FEEDBACK
+)
+from app.services.platform import DataPlatformService
+from app.services.transfer import TransferService, Season
+from app.utils.date import get_time_str, get_quarter_minutes, TIME_FMT
+
+
+class TargetController(Controller):
+
+    def __init__(
+            self,
+            realtime_data: float,
+            feedback: Dict,
+            is_customized: bool,
+            is_temporary: bool,
+            current_targets: pd.DataFrame,
+    ) -> None:
+        super(TargetController, self).__init__()
+        self._realtime_data = realtime_data
+        self._feedback = feedback
+        self._is_customized = is_customized
+        self._is_temporary = is_temporary
+        self._current_targets = current_targets
+        self._now_time = arrow.get(get_time_str(), TIME_FMT).time().strftime('%H%M%S')
+        self._quarter_time = get_quarter_minutes(get_time_str())
+
+    async def calculate_diff(self, weight: Dict) -> float:
+        related_feedback = [v for k, v in self._feedback.items() if k in weight]
+        related_feedback = np.array(related_feedback)
+        weight = np.array(list(weight.values()))
+        feedback_count = related_feedback.sum()
+        diff = 0
+        if feedback_count > 0:
+            diff = np.dot(related_feedback, weight) / feedback_count
+
+        return diff
+
+    @abstractmethod
+    async def init_temporary(self):
+        pass
+
+    @abstractmethod
+    async def get_targets(self):
+        pass
+
+    async def generate_temporary(self, lower, upper):
+        now_str = get_time_str()
+        start = arrow.get(arrow.get(now_str, TIME_FMT).timestamp // (15 * 60) * (15 * 60))
+        end = start.shift(hours=2)
+        current = start
+        time_index = []
+        while current <= end:
+            temp = current.time().strftime('%H%M%S')
+            time_index.append(temp)
+            current = current.shift(minutes=15)
+        result = {item: [lower, upper] for item in time_index}
+        self._results.update({'temporary_targets': result})
+
+    async def readjust_global(self, latest_change: float, previous_changes: pd.DataFrame):
+        previous_changes = pd.concat([
+            pd.DataFrame({'timestamp': [self._now_time], 'value': [latest_change]}),
+            previous_changes,
+        ])
+        previous_changes.reset_index(inplace=True)
+        previous_changes['weight1'] = previous_changes['index'].apply(lambda x: (1 / (x + 1)) ** 3)
+        new_targets = []
+        time_index = self._current_targets.reset_index()['time']
+        for item in time_index:
+            previous_changes['delta'] = previous_changes['timestamp'].apply(
+                lambda x: abs(arrow.get(str(x), 'HHmmss') - arrow.get(item, 'HHmmss')).seconds // (15 * 60)
+            )
+            previous_changes['weight2'] = previous_changes['delta'].apply(lambda x: 0.5 ** x)
+            previous_changes['weight'] = previous_changes['weight1'] * previous_changes['weight2']
+            new_targets.append(
+                (previous_changes['value'] * previous_changes['weight']).sum() / previous_changes['weight'].sum()
+            )
+        self._current_targets['new_targets'] = new_targets
+
+    @abstractmethod
+    async def run(self):
+        pass
+
+
+class TemperatureTargetController(TargetController):
+
+    def __init__(
+            self,
+            realtime_data: float,
+            feedback: Dict,
+            is_customized: bool,
+            is_temporary: bool,
+            current_targets: pd.DataFrame,
+            season: Season,
+            previous_changes: Optional[pd.DataFrame] = None
+    ) -> None:
+        super(TemperatureTargetController, self).__init__(
+            realtime_data,
+            feedback,
+            is_customized,
+            is_temporary,
+            current_targets
+        )
+        self._season = season
+        self._previous_changes = previous_changes
+        self._LOWER_LIMIT = 22.0
+        self._UPPER_LIMIT = 28.0
+
+    async def init_temporary(self) -> Tuple[float, float]:
+        _VAR = 2
+        _RANGE = 1
+        new_lower_bound, new_upper_bound = 23.0, 25.0
+        if self._realtime_data is not np.NAN:
+            if self._season == Season.cooling:
+                if ('a little hot' in self._feedback
+                        or 'so hot' in self._feedback
+                        or 'switch on' in self._feedback):
+                    mid = self._realtime_data - _VAR
+                    new_lower_bound = mid - _RANGE
+                    new_upper_bound = mid + _RANGE
+            elif self._season == Season.heating:
+                if ('a little cold' in self._feedback
+                        or 'so cold' in self._feedback
+                        or 'switch on' in self._feedback):
+                    mid = self._realtime_data + _VAR
+                    new_lower_bound = mid - _RANGE
+                    new_upper_bound = mid + _RANGE
+
+        return new_lower_bound, new_upper_bound
+
+    async def get_targets(self):
+        current_lower_target = self._current_targets['temperatureMin'].loc[self._quarter_time]
+        current_upper_target = self._current_targets['temperatureMax'].loc[self._quarter_time]
+
+        return current_lower_target, current_upper_target
+
+    async def readjust_current(self, lower: float, upper: float, diff: float) -> Tuple[float, float]:
+        _RANGE = 2
+        new_lower_bound, new_upper_bound = lower, upper
+        logger.info(diff)
+        if self._realtime_data is np.NAN:
+            new_lower_bound += lower + diff
+            new_upper_bound += upper + diff
+        else:
+            if self._season == Season.cooling:
+                if (diff > 0 and self._realtime_data + _RANGE > upper
+                        or diff < 0 and self._realtime_data - _RANGE < upper):
+                    new_upper_bound = self._realtime_data + diff
+                    new_lower_bound = new_upper_bound - (upper - lower)
+            else:
+                if (diff > 0 and self._realtime_data + _RANGE > lower
+                        or diff < 0 and self._realtime_data - _RANGE < lower):
+                    new_lower_bound = self._realtime_data + diff
+                    new_upper_bound = new_lower_bound + (upper - lower)
+
+        return new_lower_bound, new_upper_bound
+
+    async def generate_global(self, lower: float, upper: float):
+        extent = upper - lower
+        if self._season == Season.cooling:
+            new_targets = self._current_targets['new_targets'].apply(lambda x: [x - extent, x])
+        else:
+            new_targets = self._current_targets['new_targets'].apply(lambda x: [x, x + extent])
+
+        time_index = self._current_targets.reset_index()['time']
+        result = {}
+        for i in range(len(time_index)):
+            result.update({time_index[i]: new_targets[i]})
+        self._results.update({'global_targets': result})
+
+    async def run(self):
+        diff = await self.calculate_diff(TEMPERATURE_RELATED_FEEDBACK_WEIGHT)
+        if diff != 0:
+            if not self._is_customized:
+                lower_bound, upper_bound = await self.init_temporary()
+                await self.generate_temporary(lower_bound, upper_bound)
+            else:
+                current_lower, current_upper = await self.get_targets()
+                lower_bound, upper_bound = await self.readjust_current(current_lower, current_upper, diff)
+                if self._season == Season.cooling:
+                    temp_target = upper_bound
+                else:
+                    temp_target = lower_bound
+                if not self._is_temporary:
+                    self._results.update({'latest_change': temp_target})
+                    await self.readjust_global(temp_target, self._previous_changes)
+                    await self.generate_global(current_lower, current_upper)
+                else:
+                    await self.generate_temporary(lower_bound, upper_bound)
+        else:
+            return
+
+
+class Co2TargetController(TargetController):
+
+    def __init__(
+            self,
+            realtime_data: float,
+            feedback: Dict,
+            is_customized: bool,
+            is_temporary: bool,
+            current_targets: pd.DataFrame,
+            previous_changes: Optional[pd.DataFrame] = None
+    ) -> None:
+        super(Co2TargetController, self).__init__(
+            realtime_data,
+            feedback,
+            is_customized,
+            is_temporary,
+            current_targets
+        )
+        self._previous_changes = previous_changes
+
+    async def init_temporary(self) -> float:
+        new_target = 1000
+        diff = await self.calculate_diff(CO2_RELATED_FEEDBACK_WEIGHT)
+        if self._realtime_data is not np.NAN:
+            new_target += diff
+
+        return new_target
+
+    async def get_targets(self):
+        current_upper_target = self._current_targets['co2Max'].loc[self._quarter_time]
+
+        return current_upper_target
+
+    async def readjust_current(self, lower: float, upper: float, diff: float) -> float:
+        new_target = upper - lower
+        if self._realtime_data is np.NAN:
+            new_target += diff
+        else:
+            if (diff > 50 and self._realtime_data + 100 > upper
+                    or diff < -50 and self._realtime_data - 100 < upper):
+                new_target = self._realtime_data + diff
+
+        return new_target
+
+    async def generate_global(self):
+        new_targets = self._current_targets['new_targets'].apply(lambda x: [0, x])
+        time_index = self._current_targets.reset_index()['time']
+        result = {}
+        for i in range(len(time_index)):
+            result.update({time_index[i]: new_targets[i]})
+        self._results.update({'global_targets': result})
+
+    async def run(self):
+        diff = await self.calculate_diff(CO2_RELATED_FEEDBACK_WEIGHT)
+        if diff != 0:
+            if not self._is_customized:
+                upper_bound = await self.init_temporary()
+                await self.generate_temporary(0, upper_bound)
+            else:
+                current_upper = await self.get_targets()
+                upper_bound = await self.readjust_current(0, current_upper, diff)
+                if not self._is_temporary:
+                    self._results.update({'latest_change': upper_bound})
+                    await self.readjust_global(upper_bound, self._previous_changes)
+                    await self.generate_global()
+                else:
+                    await self.generate_temporary(0, upper_bound)
+        else:
+            return
+
+
+async def readjust_all_target(project_id: str, space_id: str, wechat_time: str, feedback: Optional[Dict] = None):
+    async with AsyncClient() as client:
+        transfer = TransferService(client, project_id, space_id)
+        platform = DataPlatformService(client, project_id, space_id)
+
+        realtime_temperature = await platform.get_realtime_temperature()
+        realtime_co2 = await platform.get_realtime_co2()
+        current_targets = await transfer.get_custom_target()
+        feedback = await transfer.get_feedback(wechat_time)
+        is_temporary = await transfer.is_temporary()
+        season = await transfer.get_season()
+        previous_changes = await transfer.env_database_get()
+
+    quarter_minutes = get_quarter_minutes(get_time_str())
+    is_customized = True if quarter_minutes in current_targets.index else False
+
+    switch_on_related_feedback = SWITCH_RELATED_FEEDBACK
+    if feedback.get('switch off') and feedback.get('switch off') > 0:
+        need_switch_off = True
+        for item in switch_on_related_feedback:
+            if feedback.get(item) and feedback.get(item) > 0:
+                need_switch_off = False
+                break
+    else:
+        need_switch_off = False
+
+    need_run_room_control = False
+    if need_switch_off:
+        async with AsyncClient() as client:
+            transfer = TransferService(client, project_id, space_id)
+            await transfer.set_temporary_custom()
+
+        return need_run_room_control
+
+    temperature_results = {}
+    for item in list(TEMPERATURE_RELATED_FEEDBACK_WEIGHT.keys()):
+        if feedback.get(item) and feedback.get(item) > 0:
+            temperature_controller = TemperatureTargetController(
+                realtime_temperature,
+                feedback,
+                is_customized,
+                is_temporary,
+                current_targets[['temperatureMin', 'temperatureMax']].copy(),
+                season,
+                previous_changes['temperature']
+            )
+            await temperature_controller.run()
+            temperature_results = temperature_controller.get_results()
+            break
+
+    co2_results = {}
+    for item in list(CO2_RELATED_FEEDBACK_WEIGHT.keys()):
+        if feedback.get(item) and feedback.get(item) > 0:
+            co2_controller = Co2TargetController(
+                realtime_co2,
+                feedback,
+                is_customized,
+                is_temporary,
+                current_targets[['co2Min', 'co2Max']].copy(),
+                previous_changes['co2'],
+            )
+            await co2_controller.run()
+            co2_results = co2_controller.get_results()
+            break
+
+    if temperature_results or co2_results:
+        need_run_room_control = True
+        async with AsyncClient() as client:
+            transfer = TransferService(client, project_id, space_id)
+
+            if temperature_results.get('temporary_targets'):
+                logger.info('temperature temporary targets: ')
+                logger.info(temperature_results.get('temporary_targets'))
+                await transfer.set_custom_target('temperature', temperature_results.get('temporary_targets'), '0')
+            if temperature_results.get('global_targets'):
+                logger.info('temperature global targets:')
+                logger.info(temperature_results.get('global_targets'))
+                await transfer.set_custom_target('temperature', temperature_results.get('global_targets'), '1')
+            if temperature_results.get('latest_change'):
+                logger.info('temperature latest change:')
+                logger.info(temperature_results.get('latest_change'))
+                await transfer.env_database_set('temperature', temperature_results.get('latest_change'))
+
+            if co2_results.get('temporary_targets'):
+                logger.info(co2_results.get('temporary_targets'))
+                await transfer.set_custom_target('co2', co2_results.get('temporary_targets'), '0')
+            if co2_results.get('global_targets'):
+                logger.info(co2_results.get('global_targets'))
+                await transfer.set_custom_target('co2', co2_results.get('global_targets'), '1')
+            if co2_results.get('latest_change'):
+                logger.info(co2_results.get('latest_change'))
+                await transfer.env_database_set('co2', co2_results.get('latest_change'))
+
+    return need_run_room_control

+ 0 - 0
app/resources/params.py


+ 0 - 0
app/services/platform.py


+ 0 - 0
app/services/service.py


+ 0 - 47
app/services/temperature.py

@@ -1,47 +0,0 @@
-# -*- coding: utf-8 -*-
-
-from typing import Tuple
-import asyncio
-
-import httpx
-import pandas as pd
-
-from fastapi import Depends
-
-from app.utils.date import get_time_str
-
-TRANSFER_SERVER = 'http://api.sagacloud.cn/duoduo-service/transfer'
-DATA_PLATFORM = 'http://api.sagacloud.cn/'
-
-
-async def get_season_type(project_id: str, date: str = Depends(get_time_str)) -> str:
-    url = f'{TRANSFER_SERVER}/environment/getSeasonType'
-    params = {
-        'projectId': project_id,
-        'date': date,
-    }
-    with httpx.Client() as client:
-        r = client.get(url, params=params)
-        print(client.close())
-    print(client.close())
-    raw_info = r.json()
-
-    return raw_info.get('data')
-
-
-async def get_env_info(project_id: str, space_id: str) -> Tuple[str, pd.DataFrame, dict]:
-    url = f'{TRANSFER_SERVER}/environment/databyroom'
-    start_time = get_time_str(100 * 60, flag='ago')
-    end_time = get_time_str()
-    params = {
-        'projectId': project_id,
-        'spaceId': space_id,
-        'statTime': start_time,
-        'endTime': end_time,
-    }
-    with httpx.Client() as client:
-        pass
-
-
-if __name__ == '__main__':
-    asyncio.run(get_season_type('Pj1101020002'))

+ 0 - 0
app/services/transfer.py


+ 13 - 2
app/utils/date.py

@@ -2,6 +2,8 @@
 
 import arrow
 
+TIME_FMT = 'YYYYMMDDHHmmss'
+
 
 def get_time_str(delta: int = 0, flag: str = 'now') -> str:
     """
@@ -12,7 +14,6 @@ def get_time_str(delta: int = 0, flag: str = 'now') -> str:
     """
     utc = arrow.utcnow()
     local = utc.to('Asia/Shanghai')
-    _FMT = '%Y%m%d%H%M%S'
     if flag == 'ago':
         delta = -delta
         t = local.shift(seconds=delta)
@@ -21,6 +22,16 @@ def get_time_str(delta: int = 0, flag: str = 'now') -> str:
     else:
         t = local
 
-    time_str = t.format('%Y%m%d%H%M%S')
+    time_str = t.format(TIME_FMT)
 
     return time_str
+
+
+def get_quarter_minutes(time_str: str) -> str:
+    temp = arrow.get(time_str, TIME_FMT).timestamp // (15 * 60) * (15 * 60)
+    return arrow.get(temp).time().strftime('%H%M%S')
+
+
+if __name__ == '__main__':
+    print(get_time_str())
+    print(get_quarter_minutes(get_time_str()))

+ 0 - 0
app/utils/math.py