from unify_api.modules.common.dao.common_dao import get_fields_by_mtid from pot_libs.logger import log from unify_api.modules.anshiu.dao.fine_monitor_dao import get_aiao_1min_dao, \ get_aiao_1day_dao, get_aiao_15min_dao, get_point_15min_chart_dao, \ get_point_1day_chart_dao, get_point_1min_chart_dao, get_point_monitor_dao, \ electric_index_list_dao, electric_index_location_dao from unify_api.modules.electric.procedures.electric_util import ( get_wiring_type ) from unify_api.modules.anshiu.components.fine_monitor_cps import ( ElectricIndexParam ) from unify_api.modules.anshiu.procedures.fine_monitor_pds import ( get_threshold_by_location, get_aiao_1min_pds, get_aiao_data_pds, get_point_1min_chart_pds, get_point_data_chart_pds, GENERAL_PARAM_FIELD_2, ELECTRIC_QUALITY_FIELD_2, GENERAL_PARAM_FIELD_3, ELECTRIC_QUALITY_FIELD_3, cal_electic_value, cal_aiao_value, cal_pt_value ) from unify_api.utils.time_format import get_time_diff async def get_adio_chart_data(location_group, location_info, date_start, date_end, intervel, slots): ''' 获取环境(温度与漏电流)的曲线数据 ''' temp, res = [], [] # 工况标准,取其中一个漏电流阈值 res_curr_th = await get_threshold_by_location( location_group) mtid = location_info[0]['mtid'] if 15 * 60 >= intervel > 60: # 取时间间隔为15min的数据 temp_res_data = await get_aiao_15min_dao(mtid, date_start, date_end) temp, res = await get_aiao_data_pds(slots, temp_res_data, res_curr_th) elif intervel == 86400: # 取时间间隔为1day的数据 temp_res_data = await get_aiao_1day_dao(mtid, date_start, date_end) temp, res = await get_aiao_data_pds(slots, temp_res_data, res_curr_th) else: # 取时间间隔为1min的数据 temp_res_data = await get_aiao_1min_dao(mtid, date_start, date_end) temp, res = await get_aiao_1min_pds(slots, temp_res_data, res_curr_th) return temp, res async def get_point_chart_data(point_id, date_start, date_end, intervel, slots): ''' 获取电气量 ''' # 获取当前监测点的接表法 ctnum, mtid = await get_wiring_type(point_id) if ctnum not in [2, 3]: log.error( f"elec_index point_id={point_id} ctnum={ctnum} 找不到ctnum , 监测点已经拆除") # 返回的数值不在2,3中的,一般是装置点已经拆除。默认先给一个默认值3 ctnum = 3 if ctnum == 2: stats_items = [ "pttl_mean", "qttl_mean", "uab_mean", "ucb_mean", "ia_mean", "ic_mean", ] else: stats_items = [ "pttl_mean", "qttl_mean", "ua_mean", "ub_mean", "uc_mean", "ia_mean", "ib_mean", "ic_mean", ] if 15 * 60 >= intervel > 60: elec_data = await get_point_15min_chart_dao(mtid, stats_items, date_start, date_end) i, v, power = await get_point_data_chart_pds(ctnum, slots, elec_data) elif intervel == 86400: elec_data = await get_point_1day_chart_dao(mtid, stats_items, date_start, date_end) i, v, power = await get_point_data_chart_pds(ctnum, slots, elec_data) else: elec_data = await get_point_1min_chart_dao(mtid, ctnum, date_start, date_end) i, v, power = await get_point_1min_chart_pds(ctnum, slots, elec_data) return power, i, v, ctnum async def electric_index_list_service(mtid, start_time, end_time, param_types=None): """ 用电指标数据列表 :param mtid: :param start_time: :param end_time: :param param_types: :return: """ general_param, electric_quality, safe_param = [], [], [] # 1,获取信息 monitor = await get_fields_by_mtid(mtid) if not monitor: return -1, general_param, electric_quality, safe_param time_diff = get_time_diff(start_time, end_time) if time_diff <= 24 * 3600: table_name = "point_15min_electric" aiao_table_name = "location_15min_aiao" else: table_name = "point_1day_electric" aiao_table_name = "location_1day_aiao" # 字段是否需要组装mtid(前端下载要求的) is_merge_mtid = mtid if param_types else None if monitor.get("m_type") == 126: pt_info = await get_fields_by_mtid(mtid, "pt_info", "pt_id") if not pt_info: return -1, general_param, electric_quality, safe_param pt_id = pt_info.get("pt_id") # 电容 ctnum, pt_datas = 0, [] # 安全参数 if not param_types or 'safe_param' in param_types: safe_param = cal_pt_value(pt_datas, is_merge_mtid) if not param_types: safe_param = electric_index_list_return_data(safe_param) else: monitor_info = await get_point_monitor_dao(mtid) if not monitor_info: return -1, general_param, electric_quality, safe_param ctnum = monitor_info.get("ctnum") or 3 datas, locations_datas, aiao_datas = [], [], [] if (not param_types or 'general_param' in param_types or 'electric_quality' in param_types): datas = await electric_index_list_dao(table_name, monitor_info.get("pid"), start_time, end_time) if not param_types or 'safe_param' in param_types: locations_datas, aiao_datas = await electric_index_location_dao( aiao_table_name, mtid, start_time, end_time) # 2,封装信息 if ctnum == 2: general_param_field = GENERAL_PARAM_FIELD_2 electric_quality_field = ELECTRIC_QUALITY_FIELD_2 else: general_param_field = GENERAL_PARAM_FIELD_3 electric_quality_field = ELECTRIC_QUALITY_FIELD_3 # 常规参数统计 if not param_types or 'general_param' in param_types: general_param = cal_electic_value(datas, general_param_field, is_merge_mtid) if not param_types: general_param = electric_index_list_return_data(general_param) # 电能质量统计 if not param_types or 'electric_quality' in param_types: electric_quality = cal_electic_value(datas, electric_quality_field, is_merge_mtid) if not param_types: electric_quality = electric_index_list_return_data( electric_quality) # 安全参数 if not param_types or 'safe_param' in param_types: safe_param = cal_aiao_value(locations_datas, aiao_datas, is_merge_mtid) if not param_types: safe_param = electric_index_list_return_data(safe_param) # 3,返回信息 return ctnum, general_param, electric_quality, safe_param async def electric_index_export_service(cid, mtids, start_time, end_time, param_types=None): """ 用电指标数据导出 :param cid: :param mtids: :param start_time: :param end_time: :param param_types: :return: """ general_param_list = [] electric_quality_list = [] safe_param_list = [] ctnums = [] for mtid in mtids: ctnum, general_one, electric_one, safe_one = await \ electric_index_list_service(mtid, start_time, end_time, param_types) if ctnum == 0: return 0, [], [], [] general_param_list.extend(general_one) electric_quality_list.extend(electric_one) safe_param_list.extend(safe_one) ctnums.append(ctnum) if len(set(ctnums)) > 1: return -1, [], [], [] general_param, electric_quality, safe_param = [], [], [] # 常规参数 if 'general_param' in param_types: general_param = electric_index_export_return_data(general_param_list) # 用电质量 if 'electric_quality' in param_types: electric_quality = electric_index_export_return_data( electric_quality_list) # 安全参数 if 'safe_param' in param_types: safe_param = electric_index_export_return_data(safe_param_list) # 2,返回信息 return 1, general_param, electric_quality, safe_param def electric_index_list_return_data(datas): """ 用电指标列表数据打包 :param datas: :return: """ index_list = [] for data in datas: electric_index = ElectricIndexParam(**data) index_list.append(electric_index) return index_list def electric_index_export_return_data(datas): """ 用电指标导出数据打包 :param datas: :return: """ export_dict = dict() for data in datas: if data["index"] not in export_dict: export_dict[data["index"]] = data else: export_dict[data["index"]].update(data) return list(export_dict.values())