from numpy import mean from unify_api.modules.alarm_manager.dao.alarm_setting_dao import \ company_extend_dao from unify_api.modules.carbon_neutral.components.carbon_reduce_cps import \ CiResp, TaResp, CeiResp from unify_api.modules.carbon_neutral.procedures.carbon_reduce_pds import \ env_benefit_ca from unify_api.modules.common.procedures.common_utils import carbon_index from unify_api.modules.elec_charge.dao.elec_charge_dao import \ point_day_power_dao, get_total_kwh_dao from unify_api.modules.home_page.procedures.count_info_pds import \ carbon_status_res from unify_api.utils.common_utils import round_2, division_two, \ multiplication_two, ChineseCalendar from unify_api.utils.time_format import day_of_month, proxy_power_slots async def carbon_summary_service(cid, start, end): """碳中和-统计概况信息""" # 1. 总面积, 能耗标准 extend_dic = await company_extend_dao(cid) total_area = "" energy_standard = "" for info in extend_dic: key = info["key"] if key == "total_area": total_area = int(info["value"]) if key == "energy_standard": energy_standard = int(info["value"]) # 2.总能耗, 即电量 res_kwh = await get_total_kwh_dao(cid, start, end) # res_kwh = await query_charge_aggs(start, end, [cid]) total_kwh = round_2(res_kwh["total_kwh"]) if res_kwh else "" # 3. 能耗密度, 为总能耗除以总面积结果 energy_density = round_2(division_two(total_kwh, total_area)) # 4. 折算年, 为能耗密度除以当月天数再乘以365 m_number = day_of_month(start) conversion_year = round_2( multiplication_two(division_two(energy_density, m_number), 365)) # 5. 环境效益 env_benefit = env_benefit_ca(conversion_year, energy_standard, total_area) return CiResp( total_area=total_area, total_kwh=total_kwh, energy_density=energy_density, conversion_year=conversion_year, energy_standard=energy_standard, env_benefit=env_benefit ) async def trend_analysis_service(cid, start, end): """碳中和-趋势分析""" slots = proxy_power_slots(start, end, "YYYY-MM-DD") kwh_info = await point_day_power_dao(cid, start, end) kwh_create_time = [kwh["create_time"] for kwh in kwh_info] kwh_info = [kwh["kwh"] for kwh in kwh_info] working_day = {"slots": [], "values": [], "avg_power": ""} non_working_day = {"slots": [], "values": [], "avg_power": ""} for slot in slots: if ChineseCalendar(slot).is_workday(): working_day["slots"].append(slot[5:]) if slot in kwh_create_time: index = kwh_create_time.index(slot) working_day["values"].append(round_2(kwh_info[index])) else: working_day["values"].append("") else: non_working_day["slots"].append(slot[5:]) if slot in kwh_create_time: index = kwh_create_time.index(slot) non_working_day["values"].append(round_2(kwh_info[index])) else: non_working_day["values"].append("") # 2. 补充平均值 working_value = working_day["values"] working_value_after = [i for i in working_value if i] if working_value_after: working_avg_power = round_2(mean(working_value_after)) working_day["avg_power"] = working_avg_power non_working_value = non_working_day["values"] non_working_value_after = [i for i in non_working_value if i] if non_working_value_after: non_working_avg_power = round_2( mean(non_working_value_after)) non_working_day["avg_power"] = non_working_avg_power return TaResp( working_day=working_day, non_working_day=non_working_day ) async def carbon_emission_index_service(cid, start, end): """首页-碳排指数等信息""" carbon_info = await carbon_summary_service(cid, start, end) # 1. 能耗标准 energy_standard = carbon_info.energy_standard # 2. 折算年 conversion_year = carbon_info.conversion_year # 3. 碳排指数 # 3.1 能耗偏差 if energy_standard: deviation = (energy_standard - conversion_year) / energy_standard if conversion_year else 1 # 3.2 碳排指数 c_index = carbon_index(deviation) # 4. 达标情况 carbon_status = carbon_status_res(c_index) else: deviation, c_index, carbon_status = "", "", "" return CeiResp( carbon_index=c_index, carbon_status=carbon_status, conversion_year=conversion_year, energy_standard=energy_standard )