DatangDataPy/scada/collection_data.py

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)