guanlan-src/src/windscada/config.py

# -*- coding: utf-8 -*-
"""windscada 场配置 (契约化可迁场 + 场站目录扫描辨识).

## 一、离线数据怎么放 (2026-09-11 用户令)

所有离线数据统一放 **`data/raw/`**, 其**下一级目录即「场站名称」**; 系统**扫描辨识**, 不写死目录名:

    data/raw/
      <场站名称>/                     ← 扫描辨识的抓手: 下一级目录就是场站
        scada_10min/                  SCADA 10min 导出, 逐台平铺 WTG01.csv…WTG38.csv
        scada_1min/                   (可选) 1min 导出
        故障报警/                      报警事件导出 (SpreadsheetML *.xls)
        风机故障记录/                  检修工单台账 (*.xls/xlsx, 内按 {年}年故障记录/ 分年)
        油样报告/                      油液化验报告 (*.pdf)
      西门子4.0技术资料/              机理层 (厂商资料), 仍在 raw 根下, 按 A2 未入场站目录

辨识规则 (逐条, 命中即止, 依据可打印):
  ① 场配置的 `raw_station` 与目录名**完全相同**;
  ② 目录名与场配置 `src_farm_names` (原始件里的场站名写法) 互为子串 —— 如配置「如东」对上目录「如东海上风电场」;
  ③ `data/raw` 下**只有一个**场站目录 (单站部署) → 用它, 但报告里标出"凭单站唯一性", 不假装是精确匹配;
  ④ 其余情形 (多个目录且都不匹配) → **不猜**, 返回"未识别", 页面显示无数据并列出扫到的目录。
命中/未命中的依据由 station_report() 给出, 供 CLI 与维护页显示。

## 二、场配置怎么来

2026-08-28 多场化改造: 场定义 = 外部配置 (`configs/farms/<场名>.json`) + 内置默认 (rudong 是已跑通的正本,
保证零配置也能起)。当前场由 `WINDSCADA_FARM` 或 set_current() 指定。
**外场配置若显式给了 src_*, 以它为准**; 没给就按上面的扫描结果派生 (外场不必再抄一遍路径)。
"""
from pathlib import Path
import json
import os

ROOT = Path(os.environ.get('WINDSCADA_ROOT') or Path(__file__).resolve().parents[2])
FARM_DIR = ROOT / 'configs' / 'farms'

# 一个场必须给齐的键 — 缺任何一个都会在加载时响亮报错, 而不是等到分析中途才崩
REQUIRED = ('name', 'n_turbines', 'turbines', 'src_10min', 'src_alarm', 'store', 'rated_kw')

# 原始件根: env WINDSCADA_RUDONG_SRC (或 configs/serve.json 的 raw_dir, 启动器会导出为该环境变量);
# 默认 <安装目录>/data/raw。原始件不随包分发, 目录不存在时页面照常显示"约定位置"。
RUDONG_SRC = os.environ.get('WINDSCADA_RUDONG_SRC') or str(ROOT / 'data' / 'raw')
RAW_ROOT = Path(RUDONG_SRC)

# 场站目录下的**约定子目录名** — 这五个名字是摄入侧的接口, 改名等于换接口, 要同步 README 与维护页
STATION_SUBDIRS = ('scada_10min', 'scada_1min', '故障报警', '风机故障记录', '油样报告')
_SRC_KEYS = (('src_10min', 'scada_10min'), ('src_1min', 'scada_1min'), ('src_alarm', '故障报警'),
             ('src_workorder', '风机故障记录'), ('src_oil', '油样报告'))

_BUILTIN = {
    'rudong': {
        'name': '如东海上风电场', 'n_turbines': 38,
        'turbines': [f'WTG{i:02d}' for i in range(1, 39)],
        'raw_station': '如东',                               # data/raw/<场站名称> 的**首选**目录名
        # 原始件里的**场站名写法**: 集团口径导出件 (月度/年度台账) 用「如海风电场」,
        # 而本场配置名是「如东海上风电场」。台账/报警摄入按这些别名把本场行筛出来 ——
        # 实测 2021年3月那张汇总表里混着民勤/来福/宝力格/大岗子/宏基… 十来个场站的行。
        'src_farm_names': ['如海', '如东'],
        'store': ROOT / 'outputs/rudong/windscada',          # L0 标准仓
        'contract': ROOT / 'reference/rudong/windscada_contract.yaml',   # M1 产出
        'rated_kw': 4000,
    }
}
_CACHE = {}
_CURRENT = [os.environ.get('WINDSCADA_FARM') or 'rudong']


def scan_stations(root=None):
    """扫 data/raw 的**下一级目录** = 场站目录。→ {场站名: dict(dir, subdirs, n_files)}

    只列目录, 不读内容 —— 现场目录里可能是十几 GB 的 CSV。
    """
    root = Path(root or RAW_ROOT)
    out = {}
    if not root.is_dir():
        return out
    for d in sorted(p for p in root.iterdir() if p.is_dir()):
        subdirs = {n: d / n for n in STATION_SUBDIRS if (d / n).is_dir()}
        n_files = sum(1 for p in d.rglob('*') if p.is_file())
        out[d.name] = dict(dir=d, subdirs=subdirs, n_files=n_files)
    return out


def station_scan(cfg=None, root=None):
    """扫 + 辨 → dict(stations={...}, matched=Path|None, name=str|None, how=str, note=str)。

    how: 'raw_station' | 'alias' | 'single' | 'none' —— 页面/CLI 显示"凭什么认出这个场站目录"。
    """
    stations = scan_stations(root)
    cfg = cfg or {}
    want = str(cfg.get('raw_station') or '')
    aliases = [str(a) for a in (cfg.get('src_farm_names') or []) if a]
    if not stations:
        return dict(stations=stations, matched=None, name=None, how='none',
                    note=f'{Path(root or RAW_ROOT)} 下没有任何场站目录 → 系统无数据可呈现')
    if want and want in stations:
        return dict(stations=stations, matched=stations[want]['dir'], name=want, how='raw_station',
                    note=f'目录名与场配置 raw_station 完全相同: {want}')
    for nm, info in stations.items():
        if any(a in nm or nm in a for a in aliases):
            return dict(stations=stations, matched=info['dir'], name=nm, how='alias',
                        note=f'目录名 {nm} 与场名写法 {aliases} 匹配')
    if len(stations) == 1:
        nm = next(iter(stations))
        return dict(stations=stations, matched=stations[nm]['dir'], name=nm, how='single',
                    note=f'data/raw 下只有这一个场站目录 {nm} (单站部署) → 采用它; '
                         f'若这不是本场, 请把目录名改成 {want or "场配置里的场站名"}')
    return dict(stations=stations, matched=None, name=None, how='none',
                note=f'data/raw 下有 {len(stations)} 个场站目录 {sorted(stations)}, 但没有一个能对上'
                     f'本场 (raw_station={want!r} / 场名写法={aliases}) → 不猜, 页面显示无数据')


def station_report(cfg=None, root=None):
    """给 CLI 与维护页用的报告: 辨识结论 + 各约定子目录的件数。"""
    r = station_scan(cfg or farm(), root)
    counts = {}
    if r['matched']:
        for n in STATION_SUBDIRS:
            d = Path(r['matched']) / n
            counts[n] = sum(1 for p in d.rglob('*') if p.is_file()) if d.is_dir() else 0
    r['counts'] = counts
    return r


def _expand(cfg, name):
    """展开: turbines 简写 / 相对路径 / **按扫描结果派生 src_***。

    显式给了 src_* 的 (外场配置常见) 一律不动 —— 扫描只填空白, 不覆盖声明。
    """
    c = dict(cfg)
    t = c.get('turbines')
    if isinstance(t, str) and ':' in t:
        pat, rng = t.rsplit(':', 1)
        a, b = (int(x) for x in rng.split('-'))
        c['turbines'] = [pat.format(i) for i in range(a, b + 1)]
    c.setdefault('n_turbines', len(c.get('turbines') or []))
    for k in ('store', 'contract'):
        if c.get(k) and not str(c[k]).startswith('/'):
            c[k] = ROOT / str(c[k])
    c.setdefault('store', ROOT / f'outputs/{name}/windscada')

    # ★场站目录: 扫描辨识 (见模块头)。扫不到就指向"约定位置", 让页面能告诉人该往哪放。
    scan = station_scan(c)
    expected = RAW_ROOT / str(c.get('raw_station') or name)
    station = Path(scan['matched']) if scan['matched'] else expected
    c['raw_station_dir'] = str(station)
    c['station_how'] = scan['how']
    c['station_note'] = scan['note']
    c['n_stations'] = len(scan['stations'])
    for key, sub in _SRC_KEYS:
        c.setdefault(key, str(station / sub))

    miss = [k for k in REQUIRED if not c.get(k)]
    if miss:
        raise SystemExit(f'场配置 {name} 缺必需键: {miss} (见 config.REQUIRED)')
    return c


def available():
    """列出可用场: 内置 + configs/farms/*.json。"""
    out = dict.fromkeys(_BUILTIN, '内置')
    if FARM_DIR.is_dir():
        for f in sorted(FARM_DIR.glob('*.json')):
            out[f.stem] = str(f.relative_to(ROOT))
    return out


def set_current(name):
    farm(name)          # 先验证能加载
    _CURRENT[0] = name
    return name


def current():
    return _CURRENT[0]


def farm(name=None, refresh=False):
    """取场配置; refresh=True 重扫场站目录 (放了新数据、又不想重启服务时用)。"""
    name = name or _CURRENT[0]
    if name in _CACHE and not refresh:
        return _CACHE[name]
    if name in _BUILTIN:
        _CACHE[name] = _expand(_BUILTIN[name], name)
        return _CACHE[name]
    f = FARM_DIR / f'{name}.json'
    if not f.exists():
        raise SystemExit(f'未知场 {name}; 可用: {list(available())} '
                         f'(新场请在 {FARM_DIR.relative_to(ROOT)}/ 放 <场名>.json)')
    _CACHE[name] = _expand(json.loads(f.read_text(encoding='utf-8')), name)
    return _CACHE[name]


def raw_station_dir(name=None):
    """本场原始件根目录 = data/raw/<场站名称>。

    扫描辨识到就用它; 没辨识到则返回**约定位置** <raw_root>/<raw_station>, 这样维护页仍能
    指着正确的地方让人补数据 (而不是给一个空路径)。辨识依据见 farm()['station_note']。
    """
    return Path(str(farm(name)['raw_station_dir']))