|
|
@@ -0,0 +1,393 @@
|
|
|
+#!/usr/bin/env python3
|
|
|
+# -*- coding: utf-8 -*-
|
|
|
+r"""现场收资接入：从场站交来的年度 zip（内层是每月每类 `.mdb`）提取并转成观澜 `data/raw/<场站>/`。
|
|
|
+
|
|
|
+## 场站交来的东西长什么样（2026-09-17 实测 `F:\temp\如东风场数据`）
|
|
|
+
|
|
|
+```
|
|
|
+25年.zip (1,022 MB) / 26年.zip (961 MB)
|
|
|
+ └ 26年/1月/2026-01-cnt.zip → 2026-01-cnt.mdb (52.7 MB)
|
|
|
+ 2026-01-std.zip → 2026-01-std.mdb ( 2.3 MB) ← tblGrid: 10 分钟电网/机组量测
|
|
|
+ 2026-01-tur.zip → 2026-01-tur.mdb (47.6 MB)
|
|
|
+ 2026-01-scd.zip → 2026-01-scd.mdb ( 5.6 MB)
|
|
|
+ … 每月 12 类：cnt din dot flg grd int prs scd std sum tmp tur
|
|
|
+```
|
|
|
+
|
|
|
+即 **Access 数据库按月分类**（不是 CSV）。实测 `std.mdb` 里是单表 `tblGrid`：
|
|
|
+
|
|
|
+```
|
|
|
+TimeStamp | Station | WPSStatus | TimestampStation | CurrentL1..L3 | VoltageL1..L3 |
|
|
|
+ActivePower | ReActivePower | ActivePowerExport | ReActivePowerExport | …
|
|
|
+```
|
|
|
+`TimeStamp` 是 10 分钟节拍，`Station` 是机组号（如 91/92 —— 与 WTG 号的对应关系见 §映射）。
|
|
|
+
|
|
|
+## 读 MDB 的路线（离线、零 pip 依赖）
|
|
|
+
|
|
|
+Windows 上用 **.NET `System.Data.Odbc` + 系统自带的 `Microsoft Access Driver (*.mdb, *.accdb)`**
|
|
|
+（本机实测可用；驱动来自 Office/Access Runtime，纯服务器上可能没有 ⇒ 本器会**先探针、缺了就响亮报并给补救**）。
|
|
|
+Linux 上用 `mdbtools` 的 `mdb-export`（本机没有）。两条路都不需要往包里塞新 pip 包。
|
|
|
+
|
|
|
+## 用法
|
|
|
+
|
|
|
+ python scripts/extract_site_data.py --src <年度zip所在目录> --list
|
|
|
+ python scripts/extract_site_data.py --src <…> --extract [--year 26] [--month 1] [--out data/raw/如东/scada_mdb]
|
|
|
+ python scripts/extract_site_data.py --src <…> --probe [--class std]
|
|
|
+ python scripts/extract_site_data.py --src <…> --to-csv --class std [--year 26 --month 1] [--turbines WTG01,WTG02]
|
|
|
+
|
|
|
+退出码: 0 成功 · 2 来源/参数不齐 · 4 平台无 MDB 读取能力（已给补救办法） · 5 转换/校验失败
|
|
|
+"""
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import argparse
|
|
|
+import hashlib
|
|
|
+import json
|
|
|
+import pathlib
|
|
|
+import shutil
|
|
|
+import subprocess
|
|
|
+import sys
|
|
|
+import tempfile
|
|
|
+import zipfile
|
|
|
+
|
|
|
+ROOT = pathlib.Path(__file__).resolve().parents[1]
|
|
|
+sys.path.insert(0, str(ROOT))
|
|
|
+from src import paths as P # noqa: E402
|
|
|
+
|
|
|
+WIN32 = sys.name == 'nt' if hasattr(sys, 'name') else False
|
|
|
+WIN32 = sys.platform.startswith('win')
|
|
|
+DRIVER = 'Microsoft Access Driver (*.mdb, *.accdb)'
|
|
|
+# 类目 → 观澜侧的落点（先按已知的上线：只有 std 与 sum 是 10min 量测；其余待现场口径确认）
|
|
|
+CLASS_TO_RAW = {
|
|
|
+ 'std': 'scada_10min', # 实测: tblGrid = 10 分钟电网/机组量测（TimeStamp/Station/ActivePower/…）
|
|
|
+ 'sum': 'scada_10min', # 待确认（体量最大, 疑为统计汇总）
|
|
|
+ 'cnt': 'scada_10min', # 待确认
|
|
|
+ 'tur': 'scada_10min', # 待确认（47 MB, 疑为机组状态）
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+# 现场交来的**已导出 CSV** zip → 观澜 raw 目录（2026-09-17 实测：与 data/raw 现有 38 件逐名对应）
|
|
|
+ZIP_TO_RAW = {
|
|
|
+ '10分钟数据': 'scada_10min', # WTG01.csv … WTG38.csv（38 件 / 14.4 GB）
|
|
|
+ '1分钟数据': 'scada_1min', # 01E.csv … 38B.csv（38 件 / 12.8 GB，机组代号命名）
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+def verify_raw(src: pathlib.Path, farm: str | None = None) -> int:
|
|
|
+ """对账：现场 zip 里的 CSV ↔ `data/raw/<场>/<类>/` 现有文件（逐件比大小）。
|
|
|
+
|
|
|
+ 这一步回答的是"**现有的输入数据到底是不是从这两个 zip 来的**" —— 不去解压十几 GB 就能验。
|
|
|
+ """
|
|
|
+ farm = farm or P.farm()
|
|
|
+ zips = [z for z in find_year_zips(src) if z.stem in ZIP_TO_RAW]
|
|
|
+ if not zips:
|
|
|
+ print(f'[X] {src} 下没有 {"/".join(ZIP_TO_RAW)}.zip')
|
|
|
+ return 2
|
|
|
+ print(f'== 现场 zip ↔ data/raw/{farm}/ 对账 ==')
|
|
|
+ bad = 0
|
|
|
+ for z in zips:
|
|
|
+ d = station_dir(farm) / ZIP_TO_RAW[z.stem]
|
|
|
+ with zipfile.ZipFile(z) as zf:
|
|
|
+ ents = [e for e in zf.infolist() if not e.is_dir()]
|
|
|
+ same = miss = diff = 0
|
|
|
+ miss_names, diff_names = [], []
|
|
|
+ for e in ents:
|
|
|
+ tgt = d / pathlib.Path(e.filename).name
|
|
|
+ if not tgt.is_file():
|
|
|
+ miss += 1
|
|
|
+ miss_names.append(tgt.name)
|
|
|
+ elif tgt.stat().st_size == e.file_size:
|
|
|
+ same += 1
|
|
|
+ else:
|
|
|
+ diff += 1
|
|
|
+ diff_names.append(f'{tgt.name}(zip {e.file_size:,} vs 盘 {tgt.stat().st_size:,})')
|
|
|
+ bad += miss + diff
|
|
|
+ print(f' {z.name} → {P.rel(d)}: zip {len(ents)} 件 · 逐字节同大小 {same} · 缺 {miss} · 大小不同 {diff}')
|
|
|
+ for n in miss_names[:4]:
|
|
|
+ print(f' 缺: {n}')
|
|
|
+ for n in diff_names[:4]:
|
|
|
+ print(f' 异: {n}')
|
|
|
+ print(f' 结论: {"现有 raw 输入与现场 zip 逐件一致（可追溯）" if bad == 0 else str(bad) + " 件需要重新提取"} '
|
|
|
+ f'（缺件提取: --extract-csv --only-missing）')
|
|
|
+ return 0 if bad == 0 else 5
|
|
|
+
|
|
|
+
|
|
|
+def extract_csv(src: pathlib.Path, farm: str | None = None, only_missing: bool = False) -> int:
|
|
|
+ """把"已导出 CSV"的现场 zip 解到 `data/raw/<场>/<类>/`（幂等：同名同大小即跳过）。"""
|
|
|
+ farm = farm or P.farm()
|
|
|
+ zips = [z for z in find_year_zips(src) if z.stem in ZIP_TO_RAW]
|
|
|
+ if not zips:
|
|
|
+ print(f'[X] {src} 下没有 {"/".join(ZIP_TO_RAW)}.zip')
|
|
|
+ return 2
|
|
|
+ done = skipped = 0
|
|
|
+ for z in zips:
|
|
|
+ d = station_dir(farm) / ZIP_TO_RAW[z.stem]
|
|
|
+ d.mkdir(parents=True, exist_ok=True)
|
|
|
+ print(f' {z.name} → {P.rel(d)}')
|
|
|
+ with zipfile.ZipFile(z) as zf:
|
|
|
+ for e in zf.infolist():
|
|
|
+ if e.is_dir():
|
|
|
+ continue
|
|
|
+ tgt = d / pathlib.Path(e.filename).name
|
|
|
+ if tgt.is_file() and tgt.stat().st_size == e.file_size:
|
|
|
+ skipped += 1
|
|
|
+ continue
|
|
|
+ if only_missing and tgt.is_file():
|
|
|
+ skipped += 1
|
|
|
+ continue
|
|
|
+ print(f' + {tgt.name} {e.file_size / 1e6:.1f} MB')
|
|
|
+ with zf.open(e) as fh, open(tgt, 'wb') as out:
|
|
|
+ shutil.copyfileobj(fh, out, 1024 * 1024 * 8)
|
|
|
+ done += 1
|
|
|
+ print(f' 完成: 新解 {done} 件 · 跳过（已同大小）{skipped} 件')
|
|
|
+ print(' 接下去: python scripts/raw_data_check.py（体检）→ python scripts/rebuild_all.py（重算）')
|
|
|
+ return 0
|
|
|
+
|
|
|
+
|
|
|
+def station_dir(farm: str | None = None) -> pathlib.Path:
|
|
|
+ """`data/raw/<场站>/` —— ★场站**目录名**可与 farm 键不同（键 = rudong，目录 = 如东），
|
|
|
+ 所以必须走唯一取用口 `raw_station_dir()`；第一版按 `P.RAW_ROOT / farm` 拼，于是"38 件全缺"
|
|
|
+ （对着 data/raw/rudong 找, 而东西在 data/raw/如东）。"""
|
|
|
+ from src.windscada.config import raw_station_dir
|
|
|
+ return pathlib.Path(raw_station_dir(farm or P.farm()))
|
|
|
+
|
|
|
+
|
|
|
+def find_year_zips(src: pathlib.Path) -> list[pathlib.Path]:
|
|
|
+
|
|
|
+ return sorted([p for p in src.glob('*.zip') if p.is_file()])
|
|
|
+
|
|
|
+
|
|
|
+def list_structure(src: pathlib.Path) -> int:
|
|
|
+ zips = find_year_zips(src)
|
|
|
+ if not zips:
|
|
|
+ print(f'[X] {src} 下没有年度 zip')
|
|
|
+ return 2
|
|
|
+ print(f'== 现场收资结构 · {src} ==')
|
|
|
+ for z in zips:
|
|
|
+ with zipfile.ZipFile(z) as zf:
|
|
|
+ names = zf.namelist()
|
|
|
+ months = sorted({n.split('/')[1] for n in names if n.count('/') >= 1 and n.split('/')[1]})
|
|
|
+ classes = sorted({n.split('/')[-1].split('-')[-1].replace('.zip', '')
|
|
|
+ for n in names if n.lower().endswith('.zip')})
|
|
|
+ print(f' {z.name} {z.stat().st_size / 1e6:.0f} MB · {len(names)} 条目')
|
|
|
+ print(f' 月份 {len(months)} 个: {", ".join(months)}')
|
|
|
+ print(f' 类目 {len(classes)} 个: {", ".join(classes)}')
|
|
|
+ return 0
|
|
|
+
|
|
|
+
|
|
|
+def extract(src: pathlib.Path, out: pathlib.Path, year: str | None, month: str | None) -> int:
|
|
|
+ zips = [z for z in find_year_zips(src) if (not year or z.stem.startswith(year))]
|
|
|
+ if not zips:
|
|
|
+ print('[X] 没有匹配的年度 zip')
|
|
|
+ return 2
|
|
|
+ out.mkdir(parents=True, exist_ok=True)
|
|
|
+ man_p = out / '_manifest.json'
|
|
|
+ man = json.loads(man_p.read_text(encoding='utf-8')) if man_p.is_file() else dict(zips=[], files=[])
|
|
|
+ done = 0
|
|
|
+ for z in zips:
|
|
|
+ with zipfile.ZipFile(z) as zf:
|
|
|
+ inner = [n for n in zf.namelist() if n.lower().endswith('.zip')
|
|
|
+ and (not month or f'/{month}月/' in n)]
|
|
|
+ print(f' {z.name}: 内层 {len(inner)} 个分类 zip' + (f'（限 {month} 月）' if month else ''))
|
|
|
+ for n in inner:
|
|
|
+ parts = n.split('/') # 26年/1月/2026-01-std.zip
|
|
|
+ rel = pathlib.Path(*parts[1:]) if len(parts) > 1 else pathlib.Path(parts[0])
|
|
|
+ dst_dir = out / rel.parent
|
|
|
+ dst_dir.mkdir(parents=True, exist_ok=True)
|
|
|
+ with zf.open(n) as fh, zipfile.ZipFile(fh) as inner_zf:
|
|
|
+ for m in inner_zf.namelist():
|
|
|
+ tgt = dst_dir / pathlib.Path(m).name
|
|
|
+ if tgt.is_file() and tgt.stat().st_size == inner_zf.getinfo(m).file_size:
|
|
|
+ continue
|
|
|
+ tgt.write_bytes(inner_zf.read(m))
|
|
|
+ h = hashlib.sha256(tgt.read_bytes()).hexdigest()[:16]
|
|
|
+ man['files'] = [x for x in man['files'] if x['path'] != str(tgt.relative_to(out))]
|
|
|
+ man['files'].append(dict(path=str(tgt.relative_to(out)), bytes=tgt.stat().st_size,
|
|
|
+ sha256_16=h, from_zip=f'{z.name}:{n}'))
|
|
|
+ done += 1
|
|
|
+ print(f' + {tgt.relative_to(out)} {tgt.stat().st_size / 1e6:.1f} MB sha256:{h}')
|
|
|
+ man['zips'] = sorted({*man.get('zips', []), *[z.name for z in zips]})
|
|
|
+ man_p.write_text(json.dumps(man, ensure_ascii=False, indent=1) + '\n', encoding='utf-8')
|
|
|
+ print(f' 完成: 新解 {done} 个 MDB → {P.rel(out)}（清单 {P.rel(man_p)}）')
|
|
|
+ return 0
|
|
|
+
|
|
|
+
|
|
|
+def have_mdb_route() -> tuple[bool, str]:
|
|
|
+ if WIN32:
|
|
|
+ ps = ('$d=(Get-OdbcDriver | Where-Object { $_.Name -like "*Access Driver (*.mdb*" }).Count;'
|
|
|
+ 'if($d -gt 0){"ok"}else{"no"}')
|
|
|
+ try:
|
|
|
+ r = subprocess.run(['powershell', '-NoProfile', '-Command', ps], capture_output=True, text=True,
|
|
|
+ errors='replace', timeout=120)
|
|
|
+ ok = 'ok' in (r.stdout or '')
|
|
|
+ return ok, 'Windows + .NET System.Data.Odbc + 系统 Access ODBC 驱动' if ok else '缺 Access ODBC 驱动'
|
|
|
+ except Exception as e:
|
|
|
+ return False, f'探测失败: {type(e).__name__}'
|
|
|
+ ok = bool(shutil.which('mdb-export') or shutil.which('mdb-tables'))
|
|
|
+ return ok, 'mdbtools (mdb-export)' if ok else '缺 mdbtools'
|
|
|
+
|
|
|
+
|
|
|
+def mdb_tables(mdb: pathlib.Path) -> list[str]:
|
|
|
+ if WIN32:
|
|
|
+ ps = (f'$cs="Driver={{{DRIVER}}};Dbq={mdb};";'
|
|
|
+ '$cn=New-Object System.Data.Odbc.OdbcConnection($cs);$cn.Open();'
|
|
|
+ '$cn.GetSchema("Tables") | Where-Object { $_.TABLE_TYPE -eq "TABLE" } | '
|
|
|
+ 'ForEach-Object { $_.TABLE_NAME };$cn.Close()')
|
|
|
+ r = subprocess.run(['powershell', '-NoProfile', '-Command', ps], capture_output=True, text=True,
|
|
|
+ errors='replace', timeout=300)
|
|
|
+ if r.returncode != 0:
|
|
|
+ raise RuntimeError((r.stderr or r.stdout or '').strip()[:300])
|
|
|
+ return [x.strip() for x in (r.stdout or '').splitlines() if x.strip()]
|
|
|
+ r = subprocess.run(['mdb-tables', '-1', str(mdb)], capture_output=True, text=True, errors='replace', timeout=300)
|
|
|
+ return [x.strip() for x in (r.stdout or '').splitlines() if x.strip()]
|
|
|
+
|
|
|
+
|
|
|
+def mdb_to_csv(mdb: pathlib.Path, table: str, out_csv: pathlib.Path, where: str = '') -> int:
|
|
|
+ """MDB 单表 → CSV（UTF-8）。Windows 走 PowerShell/.NET Odbc；Linux 走 mdb-export。"""
|
|
|
+ out_csv.parent.mkdir(parents=True, exist_ok=True)
|
|
|
+ if WIN32:
|
|
|
+ q = f'SELECT * FROM [{table}]' + (f' WHERE {where}' if where else '')
|
|
|
+ ps = f'''$cs="Driver={{{DRIVER}}};Dbq={mdb};"
|
|
|
+$cn=New-Object System.Data.Odbc.OdbcConnection($cs);$cn.Open()
|
|
|
+$cmd=$cn.CreateCommand();$cmd.CommandText="{q}"
|
|
|
+$da=New-Object System.Data.Odbc.OdbcDataAdapter($cmd)
|
|
|
+$dt=New-Object System.Data.DataTable;[void]$da.Fill($dt)
|
|
|
+$dt | Export-Csv -Path "{out_csv}" -NoTypeInformation -Encoding UTF8
|
|
|
+"rows=$($dt.Rows.Count)"
|
|
|
+$cn.Close()'''
|
|
|
+ tf = pathlib.Path(tempfile.gettempdir()) / '_mdb_dump.ps1'
|
|
|
+ tf.write_text(ps, encoding='utf-8-sig')
|
|
|
+ r = subprocess.run(['powershell', '-NoProfile', '-ExecutionPolicy', 'Bypass', '-File', str(tf)],
|
|
|
+ capture_output=True, text=True, errors='replace', timeout=3600)
|
|
|
+ print(' ' + (r.stdout or '').strip().replace('\n', ' ')[:80])
|
|
|
+ if r.returncode != 0:
|
|
|
+ print(' [X] ' + (r.stderr or r.stdout or '').strip()[:200])
|
|
|
+ return 5
|
|
|
+ return 0
|
|
|
+ cmd = ['mdb-export', str(mdb), table] + (['-q', where] if where else [])
|
|
|
+ with open(out_csv, 'w', encoding='utf-8', newline='') as fh:
|
|
|
+ r = subprocess.run(cmd, stdout=fh, stderr=subprocess.PIPE, text=True, errors='replace', timeout=3600)
|
|
|
+ if r.returncode != 0:
|
|
|
+ print(' [X] ' + (r.stderr or '')[:200])
|
|
|
+ return 5
|
|
|
+ return 0
|
|
|
+
|
|
|
+
|
|
|
+def probe(out: pathlib.Path, cls: str | None) -> int:
|
|
|
+ ok, route = have_mdb_route()
|
|
|
+ print(f'== MDB 读取能力 ==\n {route}: {"可用" if ok else "不可用"}')
|
|
|
+ if not ok:
|
|
|
+ print(' 补救: Windows 装 Access Runtime（或 Office）后自带 Access ODBC 驱动; '
|
|
|
+ 'Linux 装 mdbtools（apt/yum install mdbtools）; 或让现场把 MDB 导出成 CSV。')
|
|
|
+ return 4
|
|
|
+ mdbs = sorted(out.rglob('*.mdb'))
|
|
|
+ mdbs = [m for m in mdbs if (not cls or f'-{cls}.' in m.name)]
|
|
|
+ if not mdbs:
|
|
|
+ print(f' [i] {P.rel(out)} 下还没有 MDB —— 先跑 --extract')
|
|
|
+ return 0
|
|
|
+ print(f' 已解出 MDB {len(mdbs)} 个（限类目 {cls or "全部"}）')
|
|
|
+ for m in mdbs[:8]:
|
|
|
+ try:
|
|
|
+ t = mdb_tables(m)
|
|
|
+ print(f' {m.relative_to(out)}: 表 {t}')
|
|
|
+ except Exception as e:
|
|
|
+ print(f' {m.relative_to(out)}: [X] {type(e).__name__}: {str(e)[:120]}')
|
|
|
+ return 0
|
|
|
+
|
|
|
+
|
|
|
+def main() -> int:
|
|
|
+ ap = argparse.ArgumentParser(description='现场收资（年度 zip → 每月每类 MDB）提取与接入')
|
|
|
+ ap.add_argument('--src', default=str(pathlib.Path('F:/temp/如东风场数据')),
|
|
|
+ help='年度 zip 所在目录（默认按现场交来的目录名）')
|
|
|
+ ap.add_argument('--out', default=str(P.RAW_ROOT / P.farm() / 'scada_mdb'),
|
|
|
+ help='MDB 落点（默认 data/raw/<场站>/scada_mdb）')
|
|
|
+ ap.add_argument('--list', action='store_true', help='只列结构（年度 zip / 月份 / 类目）')
|
|
|
+ ap.add_argument('--extract', action='store_true', help='解压出 MDB（幂等, 带清单）')
|
|
|
+ ap.add_argument('--probe', action='store_true', help='探针: 读取能力 + 已解出 MDB 的表')
|
|
|
+ ap.add_argument('--verify-raw', action='store_true', help='对账: 现场 CSV zip ↔ data/raw 现有文件(逐件比大小)')
|
|
|
+ ap.add_argument('--extract-csv', action='store_true', help='把现场 CSV zip 解到 data/raw（幂等）')
|
|
|
+ ap.add_argument('--only-missing', action='store_true', help='与 --extract-csv 合用: 只补缺件, 不覆盖已存在的')
|
|
|
+ ap.add_argument('--to-csv', action='store_true', help='MDB → 观澜 data/raw 的 CSV（按类目映射）')
|
|
|
+ ap.add_argument('--class', dest='cls', default=None, help='类目: cnt/sum/std/tur/…')
|
|
|
+ ap.add_argument('--year', default=None, help='年度前缀（如 26）')
|
|
|
+ ap.add_argument('--month', default=None, help='月份数字（如 1）')
|
|
|
+ ap.add_argument('--table', default=None, help='指定表名（默认取该 MDB 的第一张表）')
|
|
|
+ ap.add_argument('--turbines', default=None, help='只导这些机组（逗号分隔, 如 WTG01,WTG02）')
|
|
|
+ a = ap.parse_args()
|
|
|
+ src, out = pathlib.Path(a.src), pathlib.Path(a.out)
|
|
|
+ if a.list:
|
|
|
+ return list_structure(src)
|
|
|
+ if a.extract:
|
|
|
+ return extract(src, out, a.year, a.month)
|
|
|
+ if a.verify_raw:
|
|
|
+ return verify_raw(src)
|
|
|
+ if a.extract_csv:
|
|
|
+ return extract_csv(src, only_missing=a.only_missing)
|
|
|
+ if a.probe:
|
|
|
+ return probe(out, a.cls)
|
|
|
+ if a.to_csv:
|
|
|
+ return to_csv(out, a.cls, a.year, a.month, a.table, a.turbines)
|
|
|
+ ap.print_help()
|
|
|
+ return 2
|
|
|
+
|
|
|
+
|
|
|
+def to_csv(out: pathlib.Path, cls: str | None, year: str | None, month: str | None,
|
|
|
+ table: str | None, turbines: str | None) -> int:
|
|
|
+ ok, route = have_mdb_route()
|
|
|
+ if not ok:
|
|
|
+ print(f'[X] 本机没有 MDB 读取能力（{route}）—— 见 --probe 的补救办法')
|
|
|
+ return 4
|
|
|
+ if not cls:
|
|
|
+ print('[X] --to-csv 需要 --class（如 std）')
|
|
|
+ return 2
|
|
|
+ mdbs = [m for m in sorted(out.rglob(f'*-{cls}.mdb')) if (not year or f'{year}年' in str(m))
|
|
|
+ and (not month or f'{month}月' in str(m))]
|
|
|
+ if not mdbs:
|
|
|
+ print(f'[X] 没找到 *-{cls}.mdb（先 --extract）')
|
|
|
+ return 2
|
|
|
+ raw_dir = station_dir() / CLASS_TO_RAW.get(cls, f'scada_{cls}')
|
|
|
+ print(f'== MDB → CSV · 类目 {cls} · {len(mdbs)} 个月 · 落点 {P.rel(raw_dir)} ==')
|
|
|
+ want = {t.strip().upper() for t in turbines.split(',')} if turbines else None
|
|
|
+ for m in mdbs:
|
|
|
+ try:
|
|
|
+ tbls = mdb_tables(m)
|
|
|
+ except Exception as e:
|
|
|
+ print(f' {m.name}: [X] {type(e).__name__}: {str(e)[:120]}')
|
|
|
+ continue
|
|
|
+ tbl = table or (tbls[0] if tbls else None)
|
|
|
+ if not tbl:
|
|
|
+ print(f' {m.name}: 没有表')
|
|
|
+ continue
|
|
|
+ tmp_csv = pathlib.Path(tempfile.gettempdir()) / f'_{m.stem}_{tbl}.csv'
|
|
|
+ rc = mdb_to_csv(m, tbl, tmp_csv)
|
|
|
+ if rc:
|
|
|
+ continue
|
|
|
+ import pandas as pd
|
|
|
+ df = pd.read_csv(tmp_csv, dtype=str)
|
|
|
+ print(f' {m.name} / {tbl}: {len(df)} 行 × {len(df.columns)} 列 → {list(df.columns)[:8]}')
|
|
|
+ # 现场口径: Station = 机组号(91→WTG01), 每个机组一个文件、时间列名统一成 Time
|
|
|
+ st = next((c for c in df.columns if c.lower() in ('station', 'stationid', 'wtg')), None)
|
|
|
+ ts = next((c for c in df.columns if 'time' in c.lower() and 'station' not in c.lower()), None)
|
|
|
+ if not st or not ts:
|
|
|
+ print(f' [i] 没识别出 Station/TimeStamp 列 —— 该月份先原样留存 {P.rel(tmp_csv)}')
|
|
|
+ continue
|
|
|
+ base = df[st].astype(str).str.extract(r'(\d+)')[0].astype('Int64')
|
|
|
+ df['_wtg'] = 'WTG' + (base - (90 if base.min() and base.min() >= 90 else 0)).astype('Int64').astype(str).str.zfill(2)
|
|
|
+ for w, g in df.groupby('_wtg'):
|
|
|
+ if want and w.upper() not in want:
|
|
|
+ continue
|
|
|
+ g2 = g.drop(columns=['_wtg']).rename(columns={ts: 'Time'})
|
|
|
+ dst = raw_dir / f'{w}.csv'
|
|
|
+ head = not dst.is_file()
|
|
|
+ g2.to_csv(dst, mode='a' if not head else 'w', header=head, index=False, encoding='utf-8')
|
|
|
+ if head:
|
|
|
+ print(f' → {P.rel(dst)} {len(g2)} 行（Time 列 = {ts}）')
|
|
|
+ print(' 完成。接下去: python scripts/raw_data_check.py（体检）→ python scripts/rebuild_all.py')
|
|
|
+ return 0
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == '__main__':
|
|
|
+ for _s in (sys.stdout, sys.stderr):
|
|
|
+ try:
|
|
|
+ _s.reconfigure(errors='replace')
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+ sys.exit(main())
|
zhouyang.xie