import hashlib import json import multiprocessing import os import sys import time from datetime import datetime from typing import Dict, List, Tuple from urllib.parse import urlparse, parse_qs import pandas as pd import requests class EnosAPIClient: TOKEN_CACHE_FILE = "enos_token_cache.json" def __init__(self, app_key: str, app_secret: str, apigw_address: str, org_id: str, is_edge: bool = True, cache_dir: str = None): self.app_key = app_key self.app_secret = app_secret self.apigw_address = apigw_address self.org_id = org_id self.protocol = "http" if is_edge else "https" self.cache_dir = cache_dir or os.path.join(os.getcwd(), ".enos_cache") os.makedirs(self.cache_dir, exist_ok=True) self.cache_file = os.path.join(self.cache_dir, self.TOKEN_CACHE_FILE) self.access_token = None self.token_expire_time = 0 self._load_token_from_cache() def _sha256_lower(self, text: str) -> str: return hashlib.sha256(text.encode('utf-8')).hexdigest().lower() def _load_token_from_cache(self) -> bool: try: if not os.path.exists(self.cache_file): return False with open(self.cache_file, 'r', encoding='utf-8') as f: cache_data = json.load(f) cache_key = f"{self.app_key}_{self.apigw_address}" token_info = cache_data.get(cache_key) if token_info and time.time() < token_info.get('expire_time', 0) - 300: self.access_token = token_info.get('access_token') self.token_expire_time = token_info.get('expire_time') return True return False except: return False def _save_token_to_cache(self, access_token: str, expire_seconds: int): try: cache_data = {} if os.path.exists(self.cache_file): with open(self.cache_file, 'r', encoding='utf-8') as f: cache_data = json.load(f) cache_key = f"{self.app_key}_{self.apigw_address}" cache_data[cache_key] = { 'access_token': access_token, 'expire_time': time.time() + expire_seconds } with open(self.cache_file, 'w', encoding='utf-8') as f: json.dump(cache_data, f, indent=2) except: pass def get_access_token(self, force_refresh: bool = False) -> Tuple[bool, str]: if not force_refresh and self.access_token and time.time() < self.token_expire_time - 300: return True, "使用缓存Token" timestamp = int(time.time() * 1000) encryption = self._sha256_lower(f"{self.app_key}{timestamp}{self.app_secret}") url = f"{self.protocol}://{self.apigw_address}/apim-token-service/v2.0/token/get" try: resp = requests.post(url, json={ "appKey": self.app_key, "encryption": encryption, "timestamp": timestamp }, timeout=10) result = resp.json() if result.get("status") == 0: token_data = result.get("data", {}) self.access_token = token_data.get("accessToken") expire = token_data.get("expire", 7200) self.token_expire_time = time.time() + expire self._save_token_to_cache(self.access_token, expire) return True, "获取Token成功" return False, result.get("msg", "未知错误") except Exception as e: return False, str(e) def _generate_signature(self, url: str, body: str = "") -> Dict: parsed = urlparse(url) params = parse_qs(parsed.query) sorted_keys = sorted(params.keys()) params_data = "".join([f"{k}{params[k][0]}" for k in sorted_keys if params[k]]) if body: params_data += body timestamp = str(int(time.time() * 1000)) sign_data = f"{self.access_token}{params_data}{timestamp}{self.app_secret}" apim_sign = self._sha256_lower(sign_data) return { "apim-accesstoken": self.access_token, "apim-signature": apim_sign, "apim-timestamp": timestamp, "Content-Type": "application/json; charset=utf-8" } def call_api(self, url: str, method: str = "GET", body: Dict = None) -> Tuple[bool, Dict]: if not self.access_token or time.time() >= self.token_expire_time: success, msg = self.get_access_token(force_refresh=True) if not success: return False, {"error": f"Token获取失败: {msg}"} body_str = json.dumps(body, ensure_ascii=False) if body else "" headers = self._generate_signature(url, body_str) try: if method.upper() == "GET": resp = requests.get(url, headers=headers, timeout=30) else: resp = requests.post(url, headers=headers, data=body_str.encode('utf-8'), timeout=30) result = resp.json() if result.get("code") == 0: return True, result return False, result except Exception as e: return False, {"error": str(e)} def split_list(lst: List, chunk_size: int) -> List[List]: """将列表按指定大小分组""" return [lst[i:i + chunk_size] for i in range(0, len(lst), chunk_size)] def query_historical_data(client: EnosAPIClient, mdm_id: str, point_ids: List[str], start_time: str, end_time: str, interval: str = "RAW", retry_times=0) -> Tuple[ bool, List[Dict]]: """ 查询历史测点数据（支持测点自动分批） """ all_items = [] point_groups = split_list(point_ids, 20) # 每20个测点一组 print(f"共 {len(point_ids)} 个测点，分 {len(point_groups)} 批查询") for idx, group in enumerate(point_groups, 1): points_str = ",".join(group) url = (f"{client.protocol}://{client.apigw_address}/cds-timeseries-service/v1.0/tsdb-detail" f"?action=query" f"&orgId={client.org_id}" f"&mdmIds={mdm_id}" f"&pointIdsWithLogic={points_str}" f"&startTime={start_time}" f"&endTime={end_time}" f"&interval={interval}") success, result = client.call_api(url, "GET") if success: items = result.get("data", {}).get("items", []) all_items.extend(items) print(f"批次 {idx}/{len(point_groups)}:获取 {len(items)} 条记录") else: if retry_times <= 5: retry_times = retry_times + 1 error_msg = result.get("error", result.get("msg", "未知错误")) print(f"批次 {idx}/{len(point_groups)}: 查询失败 - {error_msg},次数: {retry_times}") query_historical_data(client, mdm_id, point_ids, start_time, end_time, interval, retry_times) time.sleep(0.01) else: print(f"风机:{mdm_id}, {len(point_groups)}, 查询{retry_times}仍失败") return False, [] return True, all_items def get_prev_day(date: str | None | datetime = None, prev_days: int = 1) -> str: """ 获取上一天日期 :param date: 日期字符串，格式为YYYY-MM-DD :param prev_days: 上几天，默认1天 :return: 上几天日期字符串，格式为YYYY-MM-DD ，例如：2026-06-01 """ if not date: date = datetime.now().strftime("%Y-%m-%d") date = pd.to_datetime(date) prev_day = date - pd.Timedelta(days=prev_days) return prev_day.strftime("%Y-%m-%d") def get_prev_days(date: str | None | datetime = None, prev_days: int = 1) -> list: """ 获取上一天日期 :param date: 日期字符串，格式为YYYY-MM-DD :param prev_days: 上几天，默认1天 :return: 上几天日期字符串，格式为YYYY-MM-DD ，例如：2026-06-01 """ if not date: date = datetime.now().strftime("%Y-%m-%d") date = pd.to_datetime(date) prev_datas = [] if prev_days > 0: for i in range(prev_days, 0, -1): prev_datas.append((date - pd.Timedelta(days=i)).strftime("%Y-%m-%d")) else: for i in range(0, -prev_days + 1): prev_datas.append((date + pd.Timedelta(days=i)).strftime("%Y-%m-%d")) return prev_datas def query_by_day(client: EnosAPIClient, mdm_id: str, point_ids: List[str], query_date: str) -> None: """ 按天分批查询数据 """ json_datas = {} day_start = f"{query_date} 00:00:00" day_end = f"{query_date} 23:59:59" print(f"查询日期: {query_date}") success, items = query_historical_data( client, mdm_id, point_ids, day_start, day_end, "RAW" ) if success: for item in items: localtime = item.get("localtime") if not localtime in json_datas.keys(): json_datas[localtime] = {} json_datas[localtime].update(item) print(f"当天总计: {len(items)} 条记录") else: print(f"查询失败: {items}") # 保存所有数据 if json_datas: df = pd.DataFrame().from_dict(json_datas.values()) filename = f"{mdm_id}_{query_date}.parquet" df.to_parquet(filename, index=False) print(f"总共获取 {df.shape} 条记录，已保存到 {filename}") else: print(" 未获取到任何数据") def get_wind_farm(read_db): df_wind_farm = pd.read_csv('wind_farm.csv', encoding='utf-8') df_wind_farm_db = df_wind_farm[df_wind_farm['store_id'] == read_db] return df_wind_farm_db def get_wind_turbine(): pass # ===================== 使用示例 ===================== def main(read_db): # 配置参数（请替换为实际值） APP_KEY = "a6af2233-227e-4684-ab59-9e5f74e5718c" APP_SECRET = "56f24801-0429-486f-b4db-ba61a33a101c" API_GW_ADDRESS = "ag-cdt1.eniot.io" ORG_ID = "o16021383932361" # 设备ID列表 MDM_IDS = ["03puSZ72"] # 测点ID列表（100多个测点，会自动按20个一组分批） POINT_IDS = "last(WCNV.DECCNVAI005),last(WCNV.DECCNVAI008),last(WCNV.DECCNVAI006),last(WCNV.DECCNVAI009),last(WCNV.DECCNVAI007),last(WCNV.DECCNVAI010),last(WTRM.TrmTmpShfBrg),first(WYAW.DECYAWDI004),last(WYAW.DECYAWAI001),first(WTUR.TurbineSts),first(WTUR.TurbineSts_Map),last(WGEN.GenSpd),last(WGEN.TemGenNonDE),last(WGEN.TemGenDriEnd),last(WROT.TemAxis1Ctrl),last(WROT.TemAxis2Ctrl),last(WROT.TemAxis3Ctrl),last(WROT.TemB1Mot),last(WROT.TemB2Mot),last(WROT.TemB3Mot),last(WROT.CurBlade1Motor),last(WROT.CurBlade2Motor),last(WROT.CurBlade3Motor),last(WROT.DECROTAI004),last(WROT.DECROTAI005),last(WROT.DECROTAI006),last(WROT.Blade1Speed),last(WROT.Blade2Speed),last(WROT.Blade3Speed),last(WCNV.DECCNVAI012),last(WCNV.DECCNVAI011),last(WTOW.TemTower),last(WTUR.DECTURAI016),last(WTUR.DECTURAI015),first(WTUR.DECTURDI030),last(WGEN.TemGenStaU),last(WGEN.TemGenStaV),last(WGEN.TemGenStaW),last(WGEN.Torque),last(WYAW.TotalTwist),first(WTUR.AIStatusCode),last(WGEN.GenReactivePW),last(WGEN.GenActivePW),last(WTUR.DECTURAI003),last(WYAW.NacellePosition),last(WNAC.TemNacelle),last(WNAC.TemOut),last(WNAC.TemNacelleCab),last(WROT.Blade1Position),last(WROT.Blade2Position),last(WROT.Blade3Position),last(WROT.Blade1Setpoint),last(WROT.Blade2Setpoint),last(WROT.Blade3Setpoint),last(WTRM.PowerStoreTRBS),last(WNAC.DECNACAI004),last(WNAC.TheoryActivePW),last(WCNV.VolConL1),last(WCNV.CurConL1),last(WCNV.VolConL2),last(WCNV.CurConL2),last(WCNV.VolConL3),last(WCNV.CurConL3),last(WGEN.TorqueSetpoint),last(WNAC.DECNACAI003),last(WCNV.GridFreq),first(WYAW.YawCCWSts),first(WTUR.CCTurbineSts),first(WYAW.YawCWSts),last(WNAC.WindDirection),first(WTUR.GenState),last(WTRM.RotorSpd),last(WNAC.WindSpeed),last(WTRM.TemGeaMSND),last(WTRM.TemGeaMSDE),last(WTRM.GBoxOilPmpP),last(WTRM.TemGeaOil),last(WTRM.GBoxSpd)".split( ",") POINT_IDS = POINT_IDS[0:15] # 初始化客户端 client = EnosAPIClient(APP_KEY, APP_SECRET, API_GW_ADDRESS, ORG_ID) # 获取Token success, msg = client.get_access_token() if not success: print(f"error: {msg}") return print(f"success: {msg}") for query_date in get_prev_days(prev_days=1): with multiprocessing.Pool(6, maxtasksperchild=5) as pool: # 查询数据 pool.starmap(query_by_day, [(client, mdm_id, POINT_IDS, query_date) for mdm_id in MDM_IDS]) if __name__ == "__main__": read_db = 1 args = sys.argv if len(args) > 1: read_db = int(args[1]) print(read_db) main(read_db)