Hang_KG/api/test.py

# from pathlib import Path
# import json
# import pandas as pd
#
# # 获取当前脚本路径
# current_dir = Path(__file__).parent
# json_path = current_dir.parent / "config" / "config.json"
#
# # 读取 JSON 配置文件
# with open(json_path, "r", encoding="utf-8") as f:
#     config = json.load(f)
#
# conditions = config["startupConditions"]
#
#
# def analyze_startup_conditions(dataframe, conditions):
#     """
#     分析 DataFrame 是否满足启动条件。
#
#     参数:
#         dataframe (pd.DataFrame): 原始数据
#         conditions (list): 包含 column 和 state 的字典列表
#
#     返回:
#         dict: 包含是否成功、匹配的列、分析数据等信息
#     """
#     result = {
#         "success": False,
#         "matched_columns": [],
#         "start_index": None,
#         "analysis_data": None,
#         "msg": ""
#     }
#
#     # 提取列名和状态值
#     required_columns = [cond["column"] for cond in conditions]
#     thresholds = {cond["column"]: cond["state"] for cond in conditions}
#
#     # 检查是否有缺失列
#     missing_cols = [col for col in required_columns if col not in dataframe.columns]
#     if missing_cols:
#         result["msg"] = f"缺少以下列: {', '.join(missing_cols)}"
#         return result
#
#     # 构建掩码：判断每一行是否所有列都 >= 对应 state
#     condition_mask = pd.DataFrame({
#         col: (dataframe[col] >= thresholds[col]) for col in required_columns
#     }).all(axis=1)
#
#     if condition_mask.any():
#         # 找到第一个满足所有条件的行
#         start_index = condition_mask.idxmax()
#         result["success"] = True
#         result["start_index"] = start_index
#         result["matched_columns"] = required_columns
#         result["analysis_data"] = dataframe.loc[start_index:]
#         result["msg"] = "✅ 启动完成，已找到符合条件的数据段"
#     else:
#         result["msg"] = "⛔ 液压泵未全部达到25，无分析数据"
#
#     return result
#
#
# if __name__ == '__main__':
#     csv_path = r"D:\project\daima\huang\Hang_KG\data\排气原始数据.csv"
#     df = pd.read_csv(csv_path, encoding="UTF-8")
#
#     analysis_result = analyze_startup_conditions(df, conditions)
#
#     print("📌 状态:", analysis_result["msg"])
#     if analysis_result["success"]:
#         print("\n🚀 从以下行开始的数据：")
#         print(analysis_result["analysis_data"].head())  # 只显示前几行用于示例

from pathlib import Path
import json
import pandas as pd

# 获取当前脚本路径
current_dir = Path(__file__).parent
json_path = current_dir.parent / "config" / "config.json"

# 读取 JSON 配置文件
with open(json_path, "r", encoding="utf-8") as f:
    config = json.load(f)


def analyze_startup_conditions(dataframe):
    """
    分析 DataFrame 是否满足启动条件。

    参数:
        dataframe (pd.DataFrame): 原始数据
        conditions (list): 包含 column 和 state 的字典列表

    返回:
        dict: 包含是否成功、匹配的列、分析数据等信息
    """
    result = {
        "success": False,
        "matched_columns": [],
        "start_index": None,
        "analysis_data": None,
        "msg": ""
    }

    # 提取列名和状态值
    conditions = config["startupConditions"]
    required_columns = [cond["column"] for cond in conditions]
    thresholds = {cond["column"]: cond["state"] for cond in conditions}

    # 检查是否有缺失列
    missing_cols = [col for col in required_columns if col not in dataframe.columns]
    if missing_cols:
        result["msg"] = f"缺少以下列: {', '.join(missing_cols)}"
        return result

    # 构建掩码：判断每一行是否所有列都 >= 对应 state
    condition_mask = pd.DataFrame({
        col: (dataframe[col] >= thresholds[col]) for col in required_columns
    }).all(axis=1)

    if condition_mask.any():
        # 找到第一个满足所有条件的行
        start_index = condition_mask.idxmax()
        print(start_index)
        result["success"] = True
        result["start_index"] = start_index
        result["matched_columns"] = required_columns
        result["analysis_data"] = dataframe.loc[start_index:]
        result["msg"] = "✅ 启动完成，已找到符合条件的数据段"
    else:
        result["msg"] = "⛔ 液压泵未全部达到25，无分析数据"

    return result


if __name__ == '__main__':
    csv_path = r"D:\project\daima\huang\Hang_KG\data\排气原始数据.csv"
    df = pd.read_csv(csv_path, encoding="UTF-8")

    analysis_result = analyze_startup_conditions(df)

    print("📌 状态:", analysis_result["msg"])
    if analysis_result["success"]:
        print("\n🚀 从以下行开始的数据：")
        print(analysis_result["analysis_data"].head())  # 只显示前几行用于示例