AirTerminalDevice/_opt/boundary/_utils/data_service.py

import json
import datetime
import time
import pandas as pd
import traceback
import sys
import os
from functools import partial
from dateutil import tz

try:
    from workflowlib import requests
except:
    import requests


urlcfg = {
    'getpointdata_url'      : "data/getpointdata",
    'getpointsdata_url'     : "data/getpointsdata",
    'getpointsdataforai_url': "data/getpointsdataforai",
    'getpointsruntime_url'  : "data/getpointsruntime",
    'getcurrdata_url'       : "ai/getcurrdata",
    'gethisdata_url'        : "ai/gethisdata",
    'putaidata_url'         : "ai/putaidata",
    'uploadaifile_url'      : "ai/uploadaifile",
    'addpointdatum_url'     : "ai/addpointdatum",
}


class PointReader:
    
    root_url               = os.environ.get('DATA_ROOT_URL')
    upload_url             = os.environ.get('DATA_UPLOAD_URL')
    getpointdata_url       = urlcfg["getpointdata_url"]
    getpointsdata_url      = urlcfg["getpointsdata_url"]
    getpointsdataforai_url = urlcfg["getpointsdataforai_url"]
    getpointsruntime_url   = urlcfg["getpointsruntime_url"]
    getcurrdata_url        = urlcfg["getcurrdata_url"]
    gethisdata_url         = urlcfg["gethisdata_url"]
    putaidata_url          = urlcfg["putaidata_url"]
    uploadaifile_url       = urlcfg["uploadaifile_url"]
    addpointdatum_url      = urlcfg["addpointdatum_url"]

    dtfromts = partial(datetime.datetime.fromtimestamp, tz=tz.gettz('Asia/Shanghai'))
    # 最大连接重试次数，连接失败等待时间
    max_try    = 10
    post_sleep = 1
    
    def __init__(self,url=None) -> None:
        if url is not None:
            self.url = url
            print(f'使用临时url:{self.url}')
        elif self.root_url is None:
            raise Exception('未在环境变量中获取到 DATA_ROOT_URL')
        else:
            self.url = self.root_url + self.getpointsdata_url
    
    # 传入包含多个point id的list，返回一个dataframe里面包含了ts，point_id，value
    def get_points_data(self,point_ids, from_time, to_time, interval=1, type_=3, ts2dt_col=None, return_type='dict'):
        """
        :param point_ids: list
        :param from_time: datetime 开始时间
        :param to_time: datetime 结束时间
        :param interval: int=1 时间间隔
        :param type_: =3 后端业务要求
        :param ts2dt_col: list timestamp需要转换为datetime的列名
        :param return_type: str in {'dict', 'df', 'dfcol'} default='dict' 指定返回的数据结构
        'dict' 返回 {point_id: DataFrame} （原始结构）
        'df' 返回各点位加入 point_id 列再按行拼合后的结果
        'dfcol' 返回各点位以时间戳为索引，按列拼合，并用 point_id 作为 value 的列名，有 value 时才生效
        :return: DataFrame
        """
        post_data = {
            "point_ids": point_ids,
            "begin"    : round(from_time.timestamp()),
            "end"      : round(to_time.timestamp()),
            "interval" : interval,
            "type"     : type_,
        }
        rem_try = self.max_try
        while rem_try > 0:
            try:
                resp = requests.post(url=self.url, data=json.dumps(post_data),timeout=60)
                data = resp.json()['data']
                if data:
                    res = dict()
                    for point in data:
                        res[point['point_id']] = pd.DataFrame(point['data'])
                        if ts2dt_col is not None:
                            res[point['point_id']] = self.ts2dt(res[point['point_id']], ts2dt_col)
                        # res[point['point_id']].set_index(['ts'], inplace=True)
                    if return_type == 'dict':
                        return res
                    elif return_type == 'df':
                        for point in res.keys():
                            res[point]['point_id'] = point
                        return pd.concat(res.values(), axis=0)
                    elif return_type == 'dfcol':
                        res_df = pd.DataFrame()
                        for point_id, df_ in res.items():
                            res_df = pd.concat(
                                [res_df, df_.set_index('ts').rename(columns={'value': point_id})],
                                axis=1)
                        return res_df.reset_index()
                else:
                    rem_try -= 1
                    time.sleep(self.post_sleep)
            except Exception as e:
                self.error_print(sys._getframe().f_code.co_name)
                rem_try -= 1
                time.sleep(self.post_sleep)
        if rem_try == 0:
            print("\nget_points_data failed")

    # 内部函数：打印报错信息
    def error_print(self,func_name):
        print()
        print(f"{self.dtfromts(time.time())}:")
        print(f"function {func_name} error!")
        print(f"Exception Info:")
        e_type, e_value, e_traceback = sys.exc_info()
        print(e_type)
        print(e_value)
        traceback.print_tb(e_traceback)
        print()

    # 内部函数：将timestamp转换成datetime
    def ts2dt(self,df: pd.DataFrame, cols):
        for col in cols:
            df[col] = pd.Series(map(self.dtfromts, df[col]))
        return df

class PointWriter:
    upload_url        = os.environ.get('DATA_UPLOAD_URL')
    addpointdatum_url = urlcfg.get('addpointdatum_url')
    # 最大连接重试次数，连接失败等待时间
    max_try    = 10
    post_sleep = 1
    
    def __init__(self,url=None) -> None:
        if url is not None:
            self.url = url
            print(f'使用临时url:{self.url}')
        elif self.upload_url is None:
            raise Exception('未在环境变量中获取到 DATA_UPLOAD_URL')
        else:
            self.url = self.upload_url + self.addpointdatum_url
    
    # 上传数据至点位数据库
    def ai_add_point_data(self, point_id, ts, value):
        """
        :param point_id: str 数据点位，需要预先在点位库中创建好
        :param timestamp: str 当前时刻的时间 datetime
        :param value: str 点位数值
        :return: dict: 数据服务返回值，可用 dict['state'] == 0 判断存数据是否成功
        """
        url = self.upload_url + self.addpointdatum_url
        post_data = [
            {
                "point_id": str(point_id),
                "data":[{"ts": int(ts.timestamp()), "value": str(value)}]
                }
            ]

        rem_try = self.max_try
        while rem_try > 0:
            try:
                resp = requests.post(url=url,headers = {'Content-Type': 'application/json; charset=UTF-8'}, data = json.dumps(post_data),timeout=60)
                state = resp.json()['state']
                if state == 0:
                    print(f"\nput {point_id} data success!")
                    return resp.json()
                else:
                    print(f"strange resp: {resp.json()}")
                    rem_try -= 1
                    time.sleep(self.post_sleep)
                resp.close()
            except Exception as e:
                self.error_print(sys._getframe().f_code.co_name)
                rem_try -= 1
                time.sleep(self.post_sleep)
        if rem_try == 0:
            print("\nai_add_point_data failed")
            return None

    # 内部函数：打印报错信息
    def error_print(self,func_name):
        print()
        print(f"{self.dtfromts(time.time())}:")
        print(f"function {func_name} error!")
        print(f"Exception Info:")
        e_type, e_value, e_traceback = sys.exc_info()
        print(e_type)
        print(e_value)
        traceback.print_tb(e_traceback)
        print()
        

       
# # 内部函数：将datetime转换成timestamp
# def dt2ts(df: pd.DataFrame, cols):
#     for col in cols:
#         df[col] = pd.Series([dt.timestamp() for dt in df[col]])
#     return df





# # 传入一个point id，返回一个dataframe里面包含了ts，point_id，value
# def get_point_data(point_id, from_time, to_time, interval=1, type_=3, ts2dt_col=None):
#     """
#     :param point_id: string
#     :param from_time: datetime 开始时间
#     :param to_time: datetime 结束时间
#     :param interval: int=1 时间间隔
#     :param type_: =3 后端业务要求
#     :param ts2dt_col: list timestamp需要转换为datetime的列名
#     :return: DataFrame 包含时间
#     """
#     url = root_url + getpointdata_url
#     post_data = {
#         "point_id": point_id,
#         "begin": round(from_time.timestamp()),
#         "end": round(to_time.timestamp()),
#         "interval": interval,
#         "type": type_,
#     }
#     rem_try = max_try
#     while rem_try > 0:
#         try:
#             resp = requests.post(url=url, data=json.dumps(post_data))
#             ts_data = resp.json()['data']
#             if ts_data:
#                 res = pd.DataFrame(ts_data)
#                 if ts2dt_col is not None:
#                     res = ts2dt(res, ts2dt_col)
#                 # res.set_index(['ts'], inplace=True)
#                 return res
#             else:
#                 rem_try -= 1
#         except Exception as e:
#             error_print(sys._getframe().f_code.co_name)
#             rem_try -= 1
#             time.sleep(post_sleep)
#     if rem_try == 0:
#         print("\nget_point_data failed")



# # 传入包含多个point id的list，返回一个dataframe里面包含了ts，多个point_id求总后的value
# def sum_points_by_ts(point_ids, from_time, to_time, interval=1, type_=3, ts2dt_col=None):
#     """
#     :param point_ids: list
#     :param from_time: datetime 开始时间
#     :param to_time: datetime 结束时间
#     :param interval: int=1 时间间隔
#     :param type_: =3 后端业务要求
#     :param ts2dt_col: list timestamp需要转换为datetime的列名
#     :return: DataFrame
#     """
#     try:
#         data_dict = get_points_data(point_ids, from_time, to_time, interval, type_, ts2dt_col)
#         for point_id in data_dict.keys():
#             data_dict[point_id].set_index(['ts'], inplace=True)
#         return sum(_ for _ in data_dict.values()).reset_index()
#     except Exception as e:
#         error_print(sys._getframe().f_code.co_name)
#         time.sleep(post_sleep)
#         print("\nsum_points_by_ts failed")


# # 传入包含多个point id的list，返回所有point_id最新的一条数据
# def get_points_run_time(point_ids, ts2dt_col=['ts']):
#     """
#     :param point_ids: list
#     :param ts2dt_col: list=['ts'] timestamp需要转换为datetime的列名
#     :return: DataFrame
#     """
#     url = root_url + getpointsruntime_url
#     post_data = {
#         "point_ids": point_ids,
#     }
#     rem_try = max_try
#     while rem_try > 0:
#         try:
#             resp = requests.post(url=url, data=json.dumps(post_data))
#             data = resp.json()['data']
#             if data:
#                 res = pd.DataFrame(data)
#                 if ts2dt_col is not None:
#                     res = ts2dt(res, ts2dt_col)
#                 return res
#             else:
#                 rem_try -= 1
#                 time.sleep(post_sleep)
#         except Exception as e:
#             error_print(sys._getframe().f_code.co_name)
#             rem_try -= 1
#             time.sleep(post_sleep)
#     if rem_try == 0:
#         print("\nget_points_run_time failed")


# # 获取最新的AI数据
# def ai_get_curr_data(
#         model_id, model_version, algo, algo_version, module_id,
#         ts2dt_col=None,
# ):
#     """
#     :param model_id: str 模型编号
#     :param model_version: str 模型版本
#     :param algo: str 算法名称
#     :param algo_version: str 算法版本
#     :param module_id: str 模块编号
#     :param ts2dt_col: list timestamp需要转换为datetime的列名
#     :return: datetime, DataFrame 时间戳和数据
#     """
#     url = upload_url + getcurrdata_url
#     post_data = {
#         "model_id": model_id,
#         "model_version": model_version,
#         "algo": algo,
#         "algo_version": algo_version,
#         "module_id": module_id,
#     }
#     rem_try = max_try
#     while rem_try > 0:
#         try:
#             resp = requests.post(url=url, data=json.dumps(post_data))
#             data_resp = resp.json()['data']
#             if data_resp:
#                 # res = dict()
#                 # for row in data_resp:
#                 #     res[row['ts']] = pd.DataFrame(json.loads(row['data']))
#                 ts_dt = dtfromts(data_resp[0]['ts'])
#                 res = pd.DataFrame(json.loads(data_resp[0]['data']))
#                 if ts2dt_col is not None:
#                     res = ts2dt(res, ts2dt_col)
#                 return ts_dt, res
#             else:
#                 rem_try -= 1
#                 time.sleep(post_sleep)
#         except Exception as e:
#             error_print(sys._getframe().f_code.co_name)
#             rem_try -= 1
#             time.sleep(post_sleep)
#     if rem_try == 0:
#         print("\nai_get_curr_data failed")


# # 获取历史AI数据
# def ai_get_his_data(
#         model_id, model_version, algo, algo_version, module_id,
#         from_time: datetime.datetime,
#         to_time: datetime.datetime,
#         ts2dt_col=None,
#         return_type='df'
# ):
#     """
#     :param model_id: str 模型编号
#     :param model_version: str 模型版本
#     :param algo: str 算法名称
#     :param algo_version: str 算法版本
#     :param module_id: str 模块编号
#     :param from_time: datetime 开始时间
#     :param to_time: datetime 结束时间
#     :param ts2dt_col: list timestamp需要转换为datetime的列名
#     :param return_type: str='df' in {'dict', 'df'} 指定返回的数据结构
#     'dict' 返回 {ts: DataFrame} （原始结构）
#     'df' 返回各 DataFrame 加入 ts 时间列再按行拼合后的结果
#     :return: DataFrame里面包含了ts, value
#     """
#     url = upload_url + gethisdata_url
#     post_data = {
#         "model_id": model_id,
#         "model_version": model_version,
#         "algo": algo,
#         "algo_version": algo_version,
#         "module_id": module_id,
#         "begin": round(from_time.timestamp()),
#         "end": round(to_time.timestamp()),
#     }
#     rem_try = max_try
#     while rem_try > 0:
#         try:
#             resp = requests.post(url=url, data=json.dumps(post_data))
#             data_resp = resp.json()['data']
#             if data_resp:
#                 res = dict()
#                 for row in data_resp:
#                     ts = dtfromts(row['ts'])
#                     res[ts] = pd.DataFrame(json.loads(row['data']))
#                     if ts2dt_col is not None:
#                         res[ts] = ts2dt(res[ts], ts2dt_col)
#                 if return_type == 'dict':
#                     return res
#                 elif return_type == 'df':
#                     for ts in res.keys():
#                         res[ts]['ts'] = ts
#                     return pd.concat(res.values(), axis=0)
#             else:
#                 rem_try -= 1
#                 time.sleep(post_sleep)
#         except Exception as e:
#             error_print(sys._getframe().f_code.co_name)
#             rem_try -= 1
#             time.sleep(post_sleep)
#     if rem_try == 0:
#         print("\nai_get_his_data failed")

# def ai_get_his_data2(
#         model_id, model_version, algo, algo_version, module_id,
#         from_time: datetime.datetime,
#         to_time: datetime.datetime,
#         ts2dt_col=None,
#         return_type='df'
# ):
#     """
#     :param model_id: str 模型编号
#     :param model_version: str 模型版本
#     :param algo: str 算法名称
#     :param algo_version: str 算法版本
#     :param module_id: str 模块编号
#     :param from_time: datetime 开始时间
#     :param to_time: datetime 结束时间
#     :param ts2dt_col: list timestamp需要转换为datetime的列名
#     :param return_type: str='df' in {'dict', 'df'} 指定返回的数据结构
#     'dict' 返回 {ts: DataFrame} （原始结构）
#     'df' 返回各 DataFrame 加入 ts 时间列再按行拼合后的结果
#     :return: DataFrame里面包含了ts, value
#     """
#     url = root_url + gethisdata_url
#     post_data = {
#         "model_id": model_id,
#         "model_version": model_version,
#         "algo": algo,
#         "algo_version": algo_version,
#         "module_id": module_id,
#         "begin": round(from_time.timestamp()),
#         "end": round(to_time.timestamp()),
#     }
#     rem_try = max_try
#     while rem_try > 0:
#         try:
#             resp = requests.post(url=url, data=json.dumps(post_data))
#             data_resp = resp.json()['data']
#             if data_resp:
#                 res = dict()
#                 for row in data_resp:
#                     ts = dtfromts(row['ts'])
#                     res[ts] = pd.DataFrame(json.loads(row['data']))
#                     if ts2dt_col is not None:
#                         res[ts] = ts2dt(res[ts], ts2dt_col)
#                 if return_type == 'dict':
#                     return res
#                 elif return_type == 'df':
#                     for ts in res.keys():
#                         res[ts]['ts'] = ts
#                     return pd.concat(res.values(), axis=0)
#             else:
#                 rem_try -= 1
#                 time.sleep(post_sleep)
#         except Exception as e:
#             error_print(sys._getframe().f_code.co_name)
#             rem_try -= 1
#             time.sleep(post_sleep)
#     if rem_try == 0:
#         print("\nai_get_his_data failed")


# # 将 DataFrame 格式数据转换为 jsonlike 的 list 格式数据
# def df2jsonlike(df: pd.DataFrame):
#     res = []
#     for _, row in df.iterrows():
#         res.append(row.to_dict())
#     return res


# # 将模型所预测的数据存入数据库
# # 点位数据
# def ai_put_ai_data(
#         model_id, model_version, algo, algo_version, module_id,
#         ts_dt: datetime.datetime,
#         data_df: pd.DataFrame,
#         dt2ts_col=None
# ):
#     """
#     :param model_id: str 模型编号
#     :param model_version: str 模型版本
#     :param algo: str 算法名称
#     :param algo_version: str 算法版本
#     :param module_id: str 模块编号
#     :param ts_dt: datetime 业务发生时间，取用时的唯一可用时间戳
#     :param data_df: DataFrame 业务数据
#     :param dt2ts_col: list 业务数据中 datetime 列转换为 timestamp
#     :return: dict: 数据服务返回值，可用 dict['state'] == 0 判断存数据是否成功
#     """
#     url = upload_url + putaidata_url
#     if dt2ts_col is not None:
#         data_df = dt2ts(data_df, dt2ts_col)
#     post_data = {
#         "model_id": model_id,
#         "model_version": model_version,
#         "algo": algo,
#         "algo_version": algo_version,
#         "module_id": module_id,
#         "ts": round(ts_dt.timestamp()),
#         "data": df2jsonlike(data_df),
#     }
#     rem_try = max_try
#     while rem_try > 0:
#         try:
#             resp = requests.post(url=url, data=json.dumps(post_data))
#             state = resp.json()['state']
#             if state == 0:
#                 print(f"\nput {model_id}_{model_version}_{algo}_{algo_version}_{module_id} data success!")
#                 return resp.json()
#             else:
#                 print(f"strange resp: {resp.json()}")
#                 rem_try -= 1
#                 time.sleep(post_sleep)
#         except Exception as e:
#             error_print(sys._getframe().f_code.co_name)
#             rem_try -= 1
#             time.sleep(post_sleep)
#     if rem_try == 0:
#         print("\nai_put_ai_data failed")
#         return None


# # 将模型所预测的数据存入数据库
# # 能耗基线数据
# def ai_put_ai_data2(
#         model_id, model_version, algo, algo_version, module_id,
#         ts_dt: datetime.datetime,
#         data_df: pd.DataFrame,
#         dt2ts_col=None
# ):
#     """
#     :param model_id: str 模型编号
#     :param model_version: str 模型版本
#     :param algo: str 算法名称
#     :param algo_version: str 算法版本
#     :param module_id: str 模块编号
#     :param ts_dt: datetime 业务发生时间，取用时的唯一可用时间戳
#     :param data_df: DataFrame 业务数据
#     :param dt2ts_col: list 业务数据中 datetime 列转换为 timestamp
#     :return: dict: 数据服务返回值，可用 dict['state'] == 0 判断存数据是否成功
#     """
#     url = root_url + putaidata_url
#     if dt2ts_col is not None:
#         data_df = dt2ts(data_df, dt2ts_col)
#     post_data = {
#         "model_id"     : model_id,
#         "model_version": model_version,
#         "algo"         : algo,
#         "algo_version" : algo_version,
#         "module_id"    : module_id,
#         "ts"           : round(ts_dt.timestamp()),
#         "data"         : df2jsonlike(data_df) if isinstance(data_df, pd.DataFrame) else data_df,
#     }
#     rem_try = max_try
#     while rem_try > 0:
#         try:
#             resp = requests.post(url=url, data=json.dumps(post_data))
#             state = resp.json()['state']
#             if state == 0:
#                 print(f"\nput {model_id}_{model_version}_{algo}_{algo_version}_{module_id} data success!")
#                 return resp.json()
#             else:
#                 print(f"strange resp: {resp.json()}")
#                 rem_try -= 1
#                 time.sleep(post_sleep)
#         except Exception as e:
#             error_print(sys._getframe().f_code.co_name)
#             rem_try -= 1
#             time.sleep(post_sleep)
#     if rem_try == 0:
#         print("\nai_put_ai_data failed")
#         return None


# # 通过接口上传文件
# def ai_upload_ai_file(
#         model_id, algo, algo_version, module_id,
#         file
# ):
#     """
#     :param model_id: str 模型编号
#     :param model_version: str 模型版本
#     :param algo: str 算法名称
#     :param algo_version: str 算法版本
#     :param module_id: str 模块编号
#     :param ts_dt: datetime 业务发生时间，取用时的唯一可用时间戳
#     :param data_df: DataFrame 业务数据
#     :param dt2ts_col: list 业务数据中 datetime 列转换为 timestamp
#     :return: dict: 数据服务返回值，可用 dict['state'] == 0 判断存数据是否成功
#     """
#     url = upload_url + uploadaifile_url
#     post_data = {
#         "model_id"    : str(model_id),
#         "algo"        : str(algo),
#         "algo_version": str(algo_version),
#         "module_id"   : str(module_id),
#         "file"        : file
#     }

#     multipart_encoder = MultipartEncoder(
#         fields=post_data
#     )

#     rem_try = max_try
#     while rem_try > 0:
#         try:
#             resp = requests.post(url=url,headers={'Content-Type': multipart_encoder.content_type}, data=multipart_encoder)
#             state = resp.json()['state']
#             if state == 0:
#                 print(f"\nput {model_id}_{algo}_{algo_version}_{module_id} data success!")
#                 return resp.json()
#             else:
#                 print(f"strange resp: {resp.json()}")
#                 rem_try -= 1
#                 time.sleep(post_sleep)
#             resp.close()
#         except Exception as e:
#             error_print(sys._getframe().f_code.co_name)
#             rem_try -= 1
#             time.sleep(post_sleep)
#     if rem_try == 0:
#         print("\nai_upload_ai_file failed")
#         return None


# # 上传数据至点位数据库
# def ai_add_point_data(
#         point_id, ts, value
# ):
#     """
#     :param point_id: str 数据点位，需要预先在点位库中创建好
#     :param timestamp: str 当前时刻的时间 datetime
#     :param value: str 点位数值
#     :return: dict: 数据服务返回值，可用 dict['state'] == 0 判断存数据是否成功
#     """
#     url = root_url + addpointdatum_url
#     post_data = [{"point_id": str(point_id),
#                   "data":[{"ts": int(ts.timestamp()), 
#                            "value": str(value)}]}]

#     rem_try = max_try
#     while rem_try > 0:
#         try:
#             resp = requests.post(url=url,headers = {'Content-Type': 'application/json; charset=UTF-8'}, data = json.dumps(post_data))
#             state = resp.json()['state']
#             if state == 0:
#                 print(f"\nput {point_id} data success!")
#                 return resp.json()
#             else:
#                 print(f"strange resp: {resp.json()}")
#                 rem_try -= 1
#                 time.sleep(post_sleep)
#             resp.close()
#         except Exception as e:
#             error_print(sys._getframe().f_code.co_name)
#             rem_try -= 1
#             time.sleep(post_sleep)
#     if rem_try == 0:
#         print("\nai_add_point_data failed")
#         return None