guanlan-src/app_backEnd/app_backEnd_guanlan/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 多场化改造: 场定义 = 外部配置 (`app_ETL/configs/farms/<场名>.yaml`, 历史 `.json` 兼容) + 内置默认
(rudong 是已跑通的正本, 保证零配置也能起)。当前场由 `WINDSCADA_FARM` 或 set_current() 指定。
**外场配置若显式给了 src_*, 以它为准**; 没给就按上面的扫描结果派生 (外场不必再抄一遍路径)。
★ 2026-09-17 (用户令 2 统一配置): 路径一律走 `src/paths.py` 的 `P.farm_config()` / `P.config_dir('farms')`,
不再在本文件里手拼 `app_ETL/configs/farms/...`; 格式约定 = **YAML**(新场写 yaml), 且 yaml 与 json 都能被
`available()` 列出 —— 此前只认 json, 目录里 8 个 yaml 场定义等于"配了看不见"。
"""

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                      # 极端兜底（包结构异常时按位置上跳）
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[3]
from pathlib import Path
import json
import os
import sys as _sys

_sys.path.insert(0, str(_install_root(__file__)))          # 便于 `python src/windscada/config.py`
from src import paths as P                                            # 路径唯一真源 (ROOT/相对解析都在它那)

ROOT = P.ROOT
FARM_DIR = P.FARMS

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

# 原始件根: data/raw (env WINDSCADA_RUDONG_SRC 或 configs/serve.json 的 raw_dir 可覆盖, 启动器会导出)。
# 注意这里给的是**绝对路径对象**, 但它是由 ROOT 派生的 —— 安装目录整个拷走也不会失配。
RUDONG_SRC = str(P.RAW_ROOT)
RAW_ROOT = P.RAW_ROOT

# 场站目录下的**约定子目录名** — 这些名字是摄入侧的接口, 改名等于换接口, 要同步 README 与维护页
# 振动侧两目录 (2026-09-12 用户令: 「CMS 振动评估报告」遵循 data/raw/如东/windcms、
# 「振动线 handoff」遵循 data/raw/如东/m5_cms_tcm) —— 它们此前只在 outputs/ 下以**产物**形式存在,
# 源件不在 data/raw, 于是"从零重算"时振动侧无源可算。加进来后扫描/落位/数据层页都能认它们。
STATION_SUBDIRS = ('scada_10min', 'scada_1min', '故障报警', '风机故障记录', '油样报告',
                   'windcms', 'm5_cms_tcm',
                   # 2026-09-17 用户令: 现场交来的**年度 MDB 归档**（25年.zip / 26年.zip, 内层是
                   # `<年>/<月>/<年-月-类>.zip → .mdb`）就地放在场站目录下, 取代此前的"已导出 CSV"
                   # 那两个 zip（10分钟数据.zip / 1分钟数据.zip）。原始收资是 Access 库, 故单独一层。
                   'scada_mdb')
_SRC_KEYS = (('src_10min', 'scada_10min'), ('src_1min', 'scada_1min'), ('src_alarm', '故障报警'),
             ('src_workorder', '风机故障记录'), ('src_oil', '油样报告'),
             ('src_windcms', 'windcms'), ('src_m5', 'm5_cms_tcm'), ('src_mdb', 'scada_mdb'))

_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': P.store('rudong'),                          # L0 标准仓 (outputs/rudong/windscada)
        'contract': P.contract('rudong'),                    # M1 产出 (reference/rudong/windscada_contract.yaml)
        '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', P.store(name))

    # ★场站目录: 扫描辨识 (见模块头)。扫不到就指向"约定位置", 让页面能告诉人该往哪放。
    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 farm_files():
    """`app_ETL/configs/farms/` 下的候选场定义文件 (yaml/json), 不含模板/CSV。"""
    if not FARM_DIR.is_dir():
        return []
    pats = ('*.yaml', '*.yml', '*.json')
    out = []
    for p in pats:
        out += [f for f in FARM_DIR.glob(p) if not f.name.startswith('_')]
    return sorted(out)


def is_farm_def(f) -> bool:
    """这个文件是不是一份**能加载的场定义**(必需键齐)。"""
    try:
        c = _load_farm_file(Path(f))
    except Exception:
        return False
    return isinstance(c, dict) and all(c.get(k) for k in REQUIRED)


def available():
    """列出可用场: 内置 + 能加载的 `app_ETL/configs/farms/<场名>.{yaml,json}`。

    ★ 2026-09-17 修两件事 (用户令 2 "统一配置"):
      ① 原来只认 `*.json` ⇒ 目录里的 `.yaml` **配了看不见**; 现在 yaml/json 都认 (新场写 yaml);
      ② 但这个目录里混着**另一种 schema** 的文件 (机型/场站物理约束 profile: `meta` + `physical_constraints`),
         它们**不是**场定义, 列进"可用场"会在加载时炸。所以只列真能加载的, 其余的由
         `foreign_farm_files()` 报出来, 由 `scripts/config_audit.py` 记账 (见 docs §11)。
    """
    out = dict.fromkeys(_BUILTIN, '内置')
    for f in farm_files():
        if is_farm_def(f):
            out[f.stem] = str(f.relative_to(ROOT))
    return out


def foreign_farm_files():
    """`app_ETL/configs/farms/` 下"不是场定义"的文件 → [(文件, 依据)] (登记在册, 不参与场加载)。"""
    bad = []
    for f in farm_files():
        if is_farm_def(f):
            continue
        try:
            c = _load_farm_file(f)
            keys = ','.join(sorted(c)[:4]) if isinstance(c, dict) else type(c).__name__
        except Exception as e:
            keys = f'解析失败: {type(e).__name__}'
        bad.append((f, keys))
    return bad


def _load_farm_file(f: Path):
    """读场定义 —— 按后缀选解析器 (yaml 优先/约定; json 兼容历史)。"""
    text = f.read_text(encoding='utf-8-sig')
    if f.suffix.lower() in ('.yaml', '.yml'):
        import yaml
        return yaml.safe_load(text) or {}
    return json.loads(text)


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 = P.farm_config(name)                      # yaml 优先, json 兼容 (路径真源在 src/paths.py)
    if f is None or not is_farm_def(f):
        extra = ''
        if f is not None:
            extra = (f' —— 该文件存在但**不是场定义**(缺必需键 {list(REQUIRED)}); '
                     f'如果是机型/物理约束 profile, 它属于另一种 schema, 见 docs §11')
        raise SystemExit(f'未知场 {name}; 可用: {list(available())}{extra} '
                         f'(新场请在 {FARM_DIR.relative_to(ROOT)}/ 放 <场名>.yaml, 必需键见 config.REQUIRED)')
    _CACHE[name] = _expand(_load_farm_file(f), 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']))