# -*- coding: utf-8 -*- r"""观澜 v2 路径中心 —— 跨平台部署的唯一路径真源 (2026-09-11 用户令)。 ## 约定 (硬约束) 1. **代码与配置里只写相对路径**, 相对 **安装根 (ROOT)** —— 不写机器相关绝对路径 (`/Users/…`、`/Volumes/…`、`C:\…`、`D:\…`、`F:\…`)。 2. 运行时**由本模块解析成绝对路径**, 解析基准是 ROOT —— **不是 cwd**。 cwd 相对是隐形坑: 从别处调用同一个脚本 (`python <绝对路径>/scripts/x.py`) 或换工作目录, 路径就会指到别处; 本仓 2026-09-11 实测有 20+ 处 `Path('outputs/rudong/…')` 属此类, 已全部归到本模块。 3. 需要写进产物/清单/页面的路径字符串, 用 **POSIX 相对形式** (`outputs/rudong/…`, 见 `rel()`), Windows 与 Linux 通用; 不要用 `os.sep` 拼库存字符串 (那只适合"给人看"的显示, 见 `disp()`)。 4. ROOT 的确定: 环境变量 `WINDSCADA_ROOT` (冻结打包时由启动器设) → 否则按本文件位置回溯 (`src/paths.py` 的上一级) —— 因此**整个安装目录可以整个拷到别的电脑/别的盘**, 不需要改任何路径。 ## 跨平台要点 - 分隔符: 一律 `pathlib`, 需要字符串时 `.as_posix()`; - 大小写: Linux 区分大小写 —— 目录名只用本模块常量, 不靠大小写变体; - 解释器: venv 按平台探测 (`.venv/Scripts/python.exe` / `.venv/bin/python`, 见 `venv_python()`); - 中文目录名 (如 `data/raw/如东`、`故障报警`): UTF-8 存写, 两侧平台均可; 页面显示走 `disp()`。 """ from __future__ import annotations try: from ._root import install_root as _install_root except ImportError: # 直接当脚本跑(python <本文件>)时没有包上下文 from _root import install_root as _install_root import os import pathlib import sys # ---- 安装根: env → 本文件位置回溯 (src/paths.py → <安装目录>) ---------------------------- ROOT = pathlib.Path(os.environ.get('WINDSCADA_ROOT') or _install_root(__file__)) # ---- 输入侧 (离线数据): data/raw, 下一级目录 = 场站名 (见 docs/数据目录结构与落位约定) ---- RAW_ROOT = pathlib.Path(os.environ.get('WINDSCADA_RUDONG_SRC') or (ROOT / 'data' / 'raw')) # ---- 非场站维度的固定位置 (全部为 ROOT 相对) ------------------------------------------- CONFIGS = ROOT / 'configs' # P9(2026-09-28 用户令):模块专属配置下放到各模块 configs/;安装根 configs/ 只留**跨模块共用单件**。 # 次序固定(先安装根,再按模块),同一逻辑配置只许出现在一处 —— 出现两处即报错(配置必须单源)。 MODULE_CONFIG_ROOTS = tuple(ROOT / m / 'configs' for m in ('app_ETL', 'app_frontEnd', 'app_backEnd', 'app_qualityGate')) REFERENCE = ROOT / 'reference' RELEASE = ROOT / 'release' PORTAL = RELEASE / 'portal.html' VIEWER = RELEASE / 'viewer' RESOURCES = ROOT / 'resources' SIM_DIR = RESOURCES / 'oem_envision_sc1_rudong2014' DOCS = ROOT / 'docs' LOGS = ROOT / 'logs' RUN = ROOT / 'run' SCRIPTS = ROOT / 'scripts' SRC = ROOT / 'src' WHEELS = ROOT / 'wheels' DATA = ROOT / 'data' SERVE_JSON = CONFIGS / 'serve.json' MODELS_JSON = CONFIGS / 'models.json' # ---- 配置目录约定与唯一取用口 (2026-09-17 用户令 2: 统一配置目录及配置文件) ---------------- # 为什么要有这一节: 原来各模块自己拼 `ROOT / 'configs' / 'xxx.yaml'` (实测 9 个模块各拼各的), # 于是"配置放哪"随时间漂移: 顶层散着 serve.json/models.json/portal_pages.yaml, 还混进过一个 # `serve.json.bak-bomfix`; `scripts/audit_chinese_terms.py` 甚至要**试三个位置**才找得到 terms 库。 # 现在: 目录分工写死在下面, 代码**只从 `config()` / `config_dir()` 取路径**, 不许再手拼字符串; # `scripts/config_audit.py` 会按这套约定查实物与代码(见 docs §11)。 # # configs/<域>/<名字>. 域 = canonical | contracts | farms | terms | <新增> # configs/serve.json 运行期单件配置 (端口/路径真源) # configs/models.json 运行期单件配置 (本机模型档) # configs/portal_pages.yaml 运行期单件配置 (门户页面归口登记表, 见 §10) TOP_LEVEL_CONFIGS = ('serve.json', 'models.json', 'portal_pages.yaml') CONFIG_DOMAINS = ('canonical', 'contracts', 'farms', 'terms') CONFIG_EXTS = ('.yaml', '.yml', '.json', '.csv') def _config_roots() -> tuple[pathlib.Path, ...]: """配置的候选根(安装根在前,模块在后)。""" return (CONFIGS,) + MODULE_CONFIG_ROOTS def _config_hits(*parts: str, kind: str) -> list[pathlib.Path]: check = (lambda q: q.is_file()) if kind == 'file' else (lambda q: q.is_dir()) return [r.joinpath(*parts) for r in _config_roots() if check(r.joinpath(*parts))] def config(*parts: str, must_exist: bool = False) -> pathlib.Path: """配置文件的唯一取用口: `P.config('terms', 'display_map.yaml')` / `P.config('serve.json')`。 只做路径解析, 不读文件 (读法由调用方决定: yaml/json/csv 各不相同); P9 起**多根查找**:安装根 `configs/` → 各模块 `configs/`(次序固定)。 同一逻辑配置出现在两处 ⇒ 直接抛错(配置必须单源, 否则"改了没生效"是必然的)。 `must_exist=True` 时不存在就抛 FileNotFoundError —— 配置缺失应该在启动时报出来, 别静默用默认值。 """ hits = _config_hits(*parts, kind='file') if len(hits) > 1: raise RuntimeError('同一配置出现在多处(配置必须单源): ' + ' · '.join(rel(q) for q in hits)) p = hits[0] if hits else CONFIGS.joinpath(*parts) if must_exist and not p.is_file(): raise FileNotFoundError(f'缺配置文件 {p} (约定见 src/paths.py 配置一节 / docs §11)') return p def config_dir(*parts: str) -> pathlib.Path: """配置目录 (域) 的取用口: `P.config_dir('farms')`(多根查找 + 单源校验)。""" hits = _config_hits(*parts, kind='dir') if len(hits) > 1: raise RuntimeError('同一配置域出现在多处(配置必须单源): ' + ' · '.join(rel(q) for q in hits)) return hits[0] if hits else CONFIGS.joinpath(*parts) FARMS = config_dir('farms') # 场定义目录(P9 起在 app_ETL/configs/farms) def farm_config(name: str | None = None) -> pathlib.Path | None: """场定义配置文件 —— **格式统一为 YAML**, 兼容历史 `.json` (有就优先用)。 2026-09-17 实测的坑: `app_ETL/configs/farms/` 下有 8 个场定义是 `.yaml`, 而 `available()` 只认 `*.json` ⇒ 这些场**根本列不出来**(等于配置写了没人看见); 目录里那个模板还叫 `_模板.json.example`, 与实物格式相反。现在两边都认, 且约定"新的场定义写 yaml"。 """ f = farm(name) for ext in ('.yaml', '.yml', '.json'): p = FARMS / f'{f}{ext}' if p.is_file(): return p return None def farm(name: str | None = None) -> str: """当前场名: 显式 → env WINDSCADA_FARM → 'rudong'。 这里不 import src.windscada.config 以免循环 (config 反过来要用本模块)。 需要 set_current() 那种运行期切换时, 调用方把场名显式传进来即可。""" return name or os.environ.get('WINDSCADA_FARM') or 'rudong' # ---- 产物侧: outputs/<场名>/… (页面取数的仓) ------------------------------------------- def out_root(name: str | None = None) -> pathlib.Path: return ROOT / 'outputs' / farm(name) def store(name: str | None = None) -> pathlib.Path: """L0 标准仓 (parquet) —— 页面主取数处。""" return out_root(name) / 'windscada' def ont(name: str | None = None) -> pathlib.Path: """本体对象库目录 (objects.json / 检索索引 / turbine_params)。""" return out_root(name) / 'ontology' def objects_json(name: str | None = None) -> pathlib.Path: return ont(name) / 'objects.json' def cms(name: str | None = None) -> pathlib.Path: """CMS 振动诊断产物目录。""" return out_root(name) / 'windcms' def m5(name: str | None = None) -> pathlib.Path: """振动线 handoff / TCM 兼容件目录。""" return out_root(name) / 'm5_cms_tcm' def tcm_replay(name: str | None = None) -> pathlib.Path: return out_root(name) / 'tcm_compatible_replay' def sop(name: str | None = None) -> pathlib.Path: """SOP 中间件与评审落盘目录。""" return out_root(name) / 'sop' def guanlan(name: str | None = None) -> pathlib.Path: """事实契约与对外派生 (可上云面孔)。""" return out_root(name) / 'guanlan' def pitch(name: str | None = None) -> pathlib.Path: return out_root(name) / 'pitch' def paradigm(name: str | None = None) -> pathlib.Path: """范式实验件 (E3/E5/E8 底稿) —— 事实契约的输入之一 (2026-09-16 补: 原先直接用 `ROOT/'outputs'/'rudong'/'paradigm_r1'` 拼, 既写死场名又绕过了本模块)。""" return out_root(name) / 'paradigm_r1' def report_dir(name: str | None = None) -> pathlib.Path: """报告交付件目录 (`交接_振动→状态评估报告_*.md` / `现场单_*.md`) —— 由振动线出件, 并被 `src/windcms/config.py` 的 knowledge_docs 引用 (2026-09-16 补: 该目录在 v0.2.0 里 没有明确归属, 一直以 `out_root()/'report'` 的裸拼形式出现)。""" return out_root(name) / 'report' def cloud(name: str | None = None) -> pathlib.Path: """可上云面孔 (脱敏后的契约/派生件/页面) —— `scripts/guanlan_cloud_*.py` 的落点。""" return guanlan(name) / 'cloud' def contract(name: str | None = None) -> pathlib.Path: """场契约 (机型判据参数), 属 reference 侧, 不在 outputs。""" return REFERENCE / farm(name) / 'windscada_contract.yaml' def station_dir(name: str | None = None) -> pathlib.Path: """本场原始件目录 data/raw/<场站名称> —— 具体由场配置扫描结果决定, 这里只给"约定位置"兜底 (场配置里 windscada.config.farm()['raw_station_dir'] 才是权威)。""" return RAW_ROOT / farm(name) # ---- 解释器与字符串形式 ---------------------------------------------------------------- def venv_python() -> pathlib.Path | None: """本安装目录下的 venv 解释器 (跨平台); 不存在返回 None。""" for rel in (('Scripts', 'python.exe'), ('bin', 'python'), ('bin', 'python3')): p = ROOT / '.venv' / pathlib.Path(*rel) if p.exists(): return p return None def resolve(p) -> pathlib.Path: """把"可能是相对路径"的值解析成绝对路径: 相对基准是 ROOT (**不是 cwd**)。""" q = pathlib.Path(p) return q if q.is_absolute() else (ROOT / q) def rel(p) -> str: """给人/给清单的**相对**路径字符串 (POSIX 形式); 不在 ROOT 内则给绝对 POSIX。""" q = pathlib.Path(p) try: return q.resolve().relative_to(ROOT.resolve()).as_posix() except ValueError: return q.as_posix() def disp(p) -> str: """给人看的显示路径: 安装目录内的写成 `<安装目录>/…`, 分隔符随本机; 其余原样绝对路径。 (只用于显示 —— 不要把它写进产物或清单。)""" q = pathlib.Path(p) try: return str(pathlib.Path('<安装目录>') / q.resolve().relative_to(ROOT.resolve())) except ValueError: return str(q) # 显示用分隔符: 只服务于给人看的文本 (页面「位置」列)。写进产物/清单的路径请用 rel() 的 POSIX 形式。 SEP = os.sep def disp_dir(p) -> str: """目录的**显示形**: 末尾带本机分隔符 (页面「位置」列用). 路径本身一律走 `rel()` 的 POSIX 形式; 这里的 os.sep 只服务于"给人看"。""" return disp(p) + os.sep def python_exe() -> str: """跑子进程/脚本用的解释器: venv → 否则当前解释器。""" v = venv_python() return str(v) if v else sys.executable def platform_tag() -> str: """平台标识 (探测脚本/日志用): windows / linux / darwin。""" return {'nt': 'windows', 'posix': 'linux'}.get(os.name, os.name) if sys.platform != 'darwin' else 'darwin' if __name__ == '__main__': # 自检: python -m src.paths 或 python src/paths.py print(f'ROOT : {ROOT} (存在: {ROOT.is_dir()})') print(f'平台 : {platform_tag()} cwd: {pathlib.Path.cwd()}') print(f'原始件根 : {rel(RAW_ROOT)} (存在: {RAW_ROOT.is_dir()})') print(f'venv 解释器: {venv_python() or "(无, 用 " + sys.executable + ")"}') for label, p in (('store', store()), ('ontology', ont()), ('windcms', cms()), ('m5', m5()), ('sop', sop()), ('guanlan', guanlan()), ('release', RELEASE), ('sim_dir', SIM_DIR)): print(f' {label:9s} {rel(p):42s} 存在={p.exists()}')