电机测试数据处理遇UnboundLocalError及多参数工况校验需求

一、UnboundLocalError 错误原因与修复 #

错误根源 #

1. 变量作用域冲突：装饰器check_rpm中，np = check_speed(np)语句将np声明为局部变量，但赋值前就引用了它，Python优先使用局部变量，导致找不到已绑定的值。
2. 未定义/错误变量：check_speed中的running_Speed_min等阈值未从已构建的test_conditions_df中提取；check_retur_flow里的qd是笔误（应为q），transition未定义（应为字符串"Transition"）。

修复后的代码片段 #

import pandas as pd

test_conditions = ["running","start up","accelerating"]
test_parameters = ["IP (psig)","Tin1(deg)","FlowIN(gpm)","Return Flow(gpm)",
                   "H(psig)","D(psig)","speed(rpm)","2.2T(deg. F)"]
test_codes = [[0,0,0,0,0,0,0,0,1,1,1,1,1,1,1,1,2,2,2,2,2,2,2,2],[0,1,2,3,4,5,6,7,0,1,2,3,4,5,6,7,0,1,2,3,4,5,6,7]]
test_param_columns = ["Min","Max"]
data = [[10,25],[0,150],[3,5],[2,4.2],[45,10000],[440,555.3],[4250,4350],[0,10000],[10,25],[0,150],[69.69,85.58],
        [6.05,7.85],[96,10000],[1879,2222.3],[6969.69,7722.22],[0,10000],[10,25],[0,150],[17.77,21.21],[2.99,3.88],[75,10000],
        [660,799],[6200.3,6567],[0,10000]]
stcs = pd.MultiIndex(levels = [test_conditions,test_parameters], codes = test_codes)
test_conditions_df = pd.DataFrame(index = stcs, columns = test_param_columns, data= data)

# 从阈值DataFrame中提取转速阈值
running_Speed_min = test_conditions_df.loc[("running", "speed(rpm)"), "Min"]
running_Speed_max = test_conditions_df.loc[("running", "speed(rpm)"), "Max"]
start_up_Speed_min = test_conditions_df.loc[("start up", "speed(rpm)"), "Min"]
start_up_Speed_max = test_conditions_df.loc[("start up", "speed(rpm)"), "Max"]
accelerating_Speed_min = test_conditions_df.loc[("accelerating", "speed(rpm)"), "Min"]
accelerating_Speed_max = test_conditions_df.loc[("accelerating", "speed(rpm)"), "Max"]

# 提取回流参数阈值
running_return_Flow_min = test_conditions_df.loc[("running", "Return Flow(gpm)"), "Min"]
running_return_Flow_max = test_conditions_df.loc[("running", "Return Flow(gpm)"), "Max"]
start_up_return_Flow_min = test_conditions_df.loc[("start up", "Return Flow(gpm)"), "Min"]
start_up_return_Flow_max = test_conditions_df.loc[("start up", "Return Flow(gpm)"), "Max"]
accelerating_return_Flow_min = test_conditions_df.loc[("accelerating", "Return Flow(gpm)"), "Min"]
accelerating_return_Flow_max = test_conditions_df.loc[("accelerating", "Return Flow(gpm)"), "Max"]

def check_speed(speed_val):
    if running_Speed_min < speed_val < running_Speed_max:
        return "running"
    elif start_up_Speed_min < speed_val < start_up_Speed_max:
        return "start up"
    elif accelerating_Speed_min < speed_val < accelerating_Speed_max:
        return "accelerating"
    else:
        return "Transition"

def check_rpm(fn):
    def wrapper(param_val, speed_val):
        param_status = fn(param_val)
        speed_status = check_speed(speed_val)
        return param_status if param_status == speed_status else "Transition"
    return wrapper 

@check_rpm
def check_return_flow(q):
    if running_return_Flow_min < q < running_return_Flow_max:
        return "running"
    elif start_up_return_Flow_min < q < start_up_return_Flow_max:
        return "start up"
    elif accelerating_return_Flow_min < q < accelerating_return_Flow_max:
        return "accelerating"
    else:
        return "Transition"

# 测试调用
q = 5.99
speed_val = 4509
result = check_return_flow(q, speed_val)
print(result)

二、多参数工况校验的合理方案 #

装饰器不适合批量处理DataFrame数据，推荐使用Pandas向量化运算，逻辑清晰且效率更高：

完整实现代码 #

import pandas as pd
from nptdms import TdmsFile

# 1. 读取TDMS数据（替换为你的文件路径）
tdms_file = TdmsFile.read("motor_test_data.tdms")
df = tdms_file.as_dataframe()

# 2. 复用你已构建的工况阈值DataFrame
test_conditions = ["running","start up","accelerating"]
test_parameters = ["IP (psig)","Tin1(deg)","FlowIN(gpm)","Return Flow(gpm)",
                   "H(psig)","D(psig)","speed(rpm)","2.2T(deg. F)"]
test_codes = [[0,0,0,0,0,0,0,0,1,1,1,1,1,1,1,1,2,2,2,2,2,2,2,2],[0,1,2,3,4,5,6,7,0,1,2,3,4,5,6,7,0,1,2,3,4,5,6,7]]
test_param_columns = ["Min","Max"]
data = [[10,25],[0,150],[3,5],[2,4.2],[45,10000],[440,555.3],[4250,4350],[0,10000],[10,25],[0,150],[69.69,85.58],
        [6.05,7.85],[96,10000],[1879,2222.3],[6969.69,7722.22],[0,10000],[10,25],[0,150],[17.77,21.21],[2.99,3.88],[75,10000],
        [660,799],[6200.3,6567],[0,10000]]
stcs = pd.MultiIndex(levels = [test_conditions,test_parameters], codes = test_codes)
test_conditions_df = pd.DataFrame(index = stcs, columns = test_param_columns, data= data)

# 3. 构建阈值字典：{工况: {参数: (min, max)}}
condition_thresholds = {}
for condition in test_conditions:
    condition_thresholds[condition] = {
        param: (test_conditions_df.loc[(condition, param), "Min"], 
                test_conditions_df.loc[(condition, param), "Max"])
        for param in test_parameters
    }

# 4. 向量化批量判断工况
df["Condition"] = "Transition"
for condition in test_conditions:
    mask = pd.Series([True]*len(df), index=df.index)
    for param, (min_val, max_val) in condition_thresholds[condition].items():
        mask &= (df[param] > min_val) & (df[param] < max_val)
    df.loc[mask, "Condition"] = condition

# 查看结果
print(df[test_parameters + ["Condition"]].head())

优势说明 #

• 向量化运算避免了逐行循环，数据量越大效率提升越明显；
• 阈值字典统一管理，后续修改工况或参数时只需调整字典结构，无需改动判断逻辑；
• 直接在原始DataFrame上新增工况列，便于后续数据分析与报告生成。

内容的提问来源于stack exchange，提问作者Dom