Browse Source

add data query services

chenhaiyang 4 years ago
parent
commit
074c9de87d
3 changed files with 273 additions and 0 deletions
  1. 79 0
      app/services/platform.py
  2. 42 0
      app/services/service.py
  3. 152 0
      app/services/transfer.py

+ 79 - 0
app/services/platform.py

@@ -0,0 +1,79 @@
+# -*- coding: utf-8 -*-
+
+from enum import Enum
+from typing import Dict
+
+import arrow
+import numpy as np
+from httpx import AsyncClient, URL
+from loguru import logger
+
+from app.services.service import Service
+from app.utils.date import get_time_str, TIME_FMT
+from app.utils.math import round_half_up
+
+SERVER = 'http://api.sagacloud.cn'
+SECRET = 'saga123456'
+
+
+class InfoCode(str, Enum):
+    temperature = 'Tdb'
+    co2 = 'CO2'
+    hcho = 'HCHO'
+    pm2d5 = 'PM2d5'
+    humidity = 'RH'
+
+
+class DataPlatformService(Service):
+
+    def __init__(self, client: AsyncClient, project_id: str, space_id: str):
+        super(DataPlatformService, self).__init__(client)
+        self._project_id = project_id
+        self._space_id = space_id
+        self._base_url = URL(SERVER)
+        self._now_time = get_time_str()
+        self._secret = SECRET
+
+    def _common_parameters(self) -> Dict:
+        return {'projectId': self._project_id, 'secret': self._secret}
+
+    async def get_realtime_data(self, code: InfoCode) -> float:
+        url = self._base_url.join('data-platform-3/hisdata/query_by_obj')
+        params = self._common_parameters()
+        start_time = get_time_str(60 * 60, flag='ago')
+        payload = {
+            'criteria': {
+                'id': self._space_id,
+                'code': code.value,
+                'receivetime': {
+                    '$gte': start_time,
+                    '$lte': self._now_time,
+                }
+            }
+        }
+        raw_info = await self._post(url, params, payload)
+
+        value = np.NAN
+        if raw_info.get('Content'):
+            latest_data = raw_info.get('Content')[-1].get('data')
+            latest_time = raw_info.get('Content')[-1].get('receivetime')
+            if arrow.get(latest_time, TIME_FMT).shift(minutes=15) < arrow.get(self._now_time, TIME_FMT):
+                logger.info(f'delayed data - {self._space_id}: ({latest_time}, {latest_data})')
+            value = round_half_up(latest_data, 2)
+
+        return value
+
+    async def get_realtime_temperature(self) -> float:
+        return await self.get_realtime_data(InfoCode.temperature)
+
+    async def get_realtime_co2(self) -> float:
+        return await self.get_realtime_data(InfoCode.co2)
+
+    async def get_realtime_hcho(self) -> float:
+        return await self.get_realtime_data(InfoCode.hcho)
+
+    async def get_realtime_pm2d5(self) -> float:
+        return await self.get_realtime_data(InfoCode.pm2d5)
+
+    async def get_realtime_humidity(self) -> float:
+        return await self.get_realtime_data(InfoCode.humidity)

+ 42 - 0
app/services/service.py

@@ -0,0 +1,42 @@
+# -*- coding: utf-8 -*-
+
+from functools import wraps
+from typing import Dict, Optional
+
+from fastapi import HTTPException
+from httpx import AsyncClient, URL
+from loguru import logger
+
+
+def api_exception(func):
+    @wraps(func)
+    async def wrapper(*args, **kwargs):
+        try:
+            r = await func(*args, **kwargs)
+        except Exception:
+            raise HTTPException(
+                status_code=500,
+                detail='The server did not receive a correct response from the upstream server.'
+            )
+        else:
+            return r
+
+    return wrapper
+
+
+class Service:
+
+    def __init__(self, client: AsyncClient) -> None:
+        self._client = client
+
+    @api_exception
+    async def _get(self, url: URL, params: Optional[Dict] = None) -> Dict:
+        raw_response = await self._client.get(url, params=params)
+
+        return raw_response.json()
+
+    @api_exception
+    async def _post(self, url: URL, params: Optional[Dict] = None, payload: Optional[Dict] = None) -> Dict:
+        raw_response = await self._client.post(url, params=params, json=payload)
+
+        return raw_response.json()

+ 152 - 0
app/services/transfer.py

@@ -0,0 +1,152 @@
+# -*- coding: utf-8 -*-
+
+from enum import Enum
+from typing import Tuple, Dict, List
+
+import arrow
+import numpy as np
+import pandas as pd
+from httpx import AsyncClient, URL
+
+from app.services.service import Service
+from app.utils.date import get_time_str, TIME_FMT
+
+SERVER = 'http://api.sagacloud.cn'
+
+
+class Season(str, Enum):
+    cooling = 'Cooling'
+    heating = 'Warm'
+    transition = 'Transition'
+
+
+class TransferService(Service):
+
+    def __init__(self, client: AsyncClient, project_id: str, space_id: str) -> None:
+        super(TransferService, self).__init__(client)
+        self._project_id = project_id
+        self._space_id = space_id
+        self._base_url = URL(SERVER)
+        self._now_time = get_time_str()
+
+    def _common_parameters(self) -> Dict:
+        return {'projectId': self._project_id, 'spaceId': self._space_id}
+
+    async def get_season(self) -> Season:
+        url = self._base_url.join('duoduo-service/transfer/environment/getSeasonType')
+        params = {
+            'projectId': self._project_id,
+            'date': self._now_time,
+        }
+        raw_info = await self._get(url, params)
+
+        return Season(raw_info.get('data'))
+
+    async def get_env_info(self) -> Tuple[int, pd.DataFrame, Dict]:
+        url = self._base_url.join('duoduo-service/transfer/environment/databyroom')
+        start_time = get_time_str(100 * 60, flag='ago')
+        params = self._common_parameters()
+        params.update(({'startTime': start_time, 'endTime': self._now_time}))
+        raw_info = await self._get(url, params)
+
+        if raw_info.get('meterdata') and len(raw_info.get('meterdata')) > 1:
+            temp_df = pd.DataFrame(raw_info.get('meterdata'))
+            temp_df = temp_df.rename(columns=temp_df.iloc[0, :]).iloc[1:, :]
+            temp_df['date'] = pd.to_datetime(temp_df['date'])
+            temp_df.set_index('date', inplace=True)
+        else:
+            temp_df = pd.DataFrame()
+
+        return raw_info.get('flag'), temp_df, raw_info.get('target')
+
+    async def is_temporary(self) -> bool:
+        url = self._base_url.join('duoduo-service/transfer/environment/temp/target')
+        params = self._common_parameters()
+        params.update({'time': self._now_time})
+        raw_info = await self._get(url, params)
+        flag = False
+        if raw_info.get('flag') == 1:
+            flag = True
+
+        return flag
+
+    async def get_feedback(self, wechat_time: str) -> Dict:
+        url = self._base_url.join('duoduo-service/transfer/environment/feedbackCount')
+        params = self._common_parameters()
+        params.update({'time': wechat_time})
+        raw_info = await self._get(url, params)
+
+        meaning_dict = {
+            'Id1': 'a little cold',
+            'Id2': 'so cold',
+            'Id3': 'a little hot',
+            'Id4': 'so hot',
+            'Id5': 'noisy or blowy',
+            'Id6': 'so stuffy',
+            'Id7': 'more sunshine',
+            'Id8': 'less sunshine',
+            'Id9': 'send a repairman',
+            'Id10': 'switch off',
+            'Id11': 'nice',
+            'Id12': 'switch on',
+        }
+
+        feedback_dic = {meaning_dict.get(k): v for k, v in raw_info.items() if k != 'result'}
+
+        return feedback_dic
+
+    async def get_custom_target(self) -> pd.DataFrame:
+        url = self._base_url.join('duoduo-service/transfer/environment/target')
+        params = self._common_parameters()
+        params.update({'date': arrow.get(self._now_time, TIME_FMT).date().strftime('%Y%m%d')})
+        raw_info = await self._get(url, params)
+
+        custom_target_df = pd.DataFrame(raw_info.get('data'))
+        # custom_target_df['timestamp'] = pd.to_datetime(custom_target_df['date'] + custom_target_df['time'])
+        custom_target_df.set_index('time', inplace=True)
+
+        return custom_target_df
+
+    async def env_database_set(self, form: str, value: float) -> None:
+        url = self._base_url.join('duoduo-service/transfer/environment/hispoint/set')
+        params = self._common_parameters()
+        time_str = arrow.get(self._now_time, TIME_FMT).date().strftime('%H%M%S')
+        params.update({'time': time_str, 'type': form, 'value': value})
+        await self._get(url, params)
+
+    async def env_database_get(self) -> Dict[str, pd.DataFrame]:
+        url = self._base_url.join('duoduo-service/transfer/environment/hispoint/get')
+        params = self._common_parameters()
+        params.update({'date': self._now_time})
+        raw_info = await self._get(url, params)
+
+        result = {}
+        if raw_info.get('result') == 'success':
+            for k, v in raw_info.items():
+                if k != 'result':
+                    if len(v) > 0:
+                        temp = {}
+                        data = np.array(v)
+                        temp.update({'timestamp': data[:, 0]})
+                        temp.update({'value': data[:, 1].astype(np.float)})
+                        result.update({k: pd.DataFrame(temp)})
+                    else:
+                        result.update({k: pd.DataFrame()})
+
+        return result
+
+    async def set_custom_target(self, form: str, target_value: Dict[str, List[float]], flag: str = '1') -> None:
+        url = self._base_url.join('duoduo-service/transfer/environment/target/setting')
+        params = {
+            'spaceId': self._space_id,
+            'timepoint': self._now_time,
+            'type': form,
+            'flag': flag
+        }
+        await self._post(url, params=params, payload=target_value)
+
+    async def set_temporary_custom(self) -> None:
+        url = self._base_url.join('duoduo-service/transfer/environment/setServiceFlag')
+        params = self._common_parameters()
+        params.update({'time': self._now_time})
+        await self._get(url, params)