guanlan-src/app_algorithmModel/app_algorithmModel_guanlan/windcms/tcm.py

# -*- coding: utf-8 -*-
"""TCM 设计层: 功率档 (WPS-ActivePower 7 分箱) × mask 阈值 (黄/红) × 趋势. 只复现外壳, 不重造受保护公式."""
import numpy as np
import pandas as pd
from src.windcms.config import HIGH_BIN, BIN_ORDER
from src.windcms.data import mask_key                        # windcms 侧数据件（未随迁）

def bin_label(key):
    return key.replace('WPS-ActivePower ', '').rstrip(',') + ' kW'

def trend(scalars, masks, farm_prefix, turbine, sensor, meas, bin_key=HIGH_BIN):
    q = scalars[(scalars.turbine == turbine) & (scalars.sensor_name == sensor) & (scalars.meas_name == meas) & (scalars.condition_key == bin_key)]
    q = q.sort_values('trigger_time')[['trigger_time', 'scalar_value', 'window', 'alarm_type']].reset_index(drop=True)
    th = masks.get((mask_key(farm_prefix, turbine, sensor, meas), bin_key), {})
    return q, th

def fleet_ref(scalars, sensor, meas, bin_key=HIGH_BIN):
    q = scalars[(scalars.sensor_name == sensor) & (scalars.meas_name == meas) & (scalars.condition_key == bin_key)]
    per = q.groupby('turbine').scalar_value.median()
    return dict(fleet_median=float(per.median()) if len(per) else np.nan, fleet_p90=float(per.quantile(.9)) if len(per) else np.nan, n_turbines=int(len(per)))

def status_table(scalars, masks, farm_prefix, bin_key=HIGH_BIN, window=None):
    """每 (台,测点,标量) 末窗中位 vs mask 阈值 → Green/Yellow/Red/NoMask; 外加 ×fleet."""
    q = scalars[scalars.condition_key == bin_key]
    if window:
        q = q[q.window == window]
    else:
        last = q.groupby('turbine').window.max()
        q = q[q.window == q.turbine.map(last)]
    g = q.groupby(['turbine', 'sensor_name', 'meas_name']).scalar_value.agg(['median', 'max', 'size']).reset_index()
    fm = q.groupby(['sensor_name', 'meas_name']).scalar_value.median().rename('fleet_median')
    g = g.merge(fm, on=['sensor_name', 'meas_name'], how='left')
    g['x_fleet'] = g['median'] / g['fleet_median']
    def st(r):
        th = masks.get((mask_key(farm_prefix, r.turbine, r.sensor_name, r.meas_name), bin_key))
        if not th or not np.isfinite(th.get('red', np.nan)):
            return 'NoMask'
        if r['median'] >= th['red']: return 'Red'
        if r['median'] >= th['yellow']: return 'Yellow'
        return 'Green'
    g['mask_status'] = g.apply(st, axis=1)
    return g