#!/usr/bin/env python3 """Cylinder-temperature scan (4#~9#) with rule-based 轻微/异常/严重异常 events. Examples -------- python _tmp_temperature/scan.py --device 6# --start "2026-01-01" --end "2026-03-01" python _tmp_temperature/scan.py --device 4# 5# 6# 7# 8# 9# python _tmp_temperature/scan.py --device 5# --include-packing All outputs stay inside ``_tmp_temperature/out/`` so the existing ``models/`` tree is untouched until the rules are validated. """ from __future__ import annotations import argparse import sys from datetime import datetime, timedelta from pathlib import Path ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from _tmp_temperature import unified as temperature_algorithm from _tmp_temperature import hf_validation from _tmp_temperature import data as temperature_data from _tmp_temperature.config import ( CYLINDERS, PACKING, UnitConfig, build_unit_config, load_device_batches, normalize_devices, ) OUT_DIR = Path(__file__).resolve().parent / "out" def parse_time(value: str) -> datetime: for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"): try: return datetime.strptime(value, fmt) except ValueError: continue raise ValueError(f"无法解析时间：{value!r}") def _auto_range(config: UnitConfig): from models.common.db import connect placeholders = ",".join(["%s"] * len(config.batches)) connection = connect() try: with connection.cursor() as cursor: cursor.execute( f"SELECT MIN(sample_time) AS mn, MAX(sample_time) AS mx " f"FROM pks_long_sample WHERE import_batch_id IN ({placeholders})", config.batches, ) row = cursor.fetchone() finally: connection.close() start = row["mn"].replace(minute=0, second=0, microsecond=0) end = row["mx"].replace(minute=0, second=0, microsecond=0) + timedelta(hours=1) return start, end def threshold_overrides(args) -> dict: mapping = { "mild_spread": args.mild_spread, "abnormal_spread": args.abnormal_spread, "severe_spread": args.severe_spread, "mild_peer": args.mild_peer, "abnormal_peer": args.abnormal_peer, "severe_peer": args.severe_peer, "mild_z": args.mild_z, "abnormal_z": args.abnormal_z, "severe_z": args.severe_z, "mild_delta": args.mild_delta, "abnormal_delta": args.abnormal_delta, "severe_delta": args.severe_delta, "confirm_hours": args.confirm_hours, } return {key: value for key, value in mapping.items() if value is not None} def point_meta(config: UnitConfig, include_packing: bool) -> dict[str, dict]: points = list(CYLINDERS[config.unit]) if include_packing and config.unit in PACKING: points.extend(PACKING[config.unit]) return { column: {"stage": stage, "cylinder": cylinder, "label": config.labels.get(column, column)} for stage, cylinder, column in points } def hourly_long_rows(rows: list[dict], config: UnitConfig, include_packing: bool) -> list[dict]: meta = point_meta(config, include_packing) output = [] for row in rows: if not row.get("valid"): continue hour = row["hour_start"].strftime("%Y-%m-%d %H:%M:%S") for column, info in meta.items(): if row.get(f"{column}__avg") is None: continue output.append({ "机组": config.name, "批次": row["import_batch_id"], "小时": hour, "点位": column, "点位描述": info["label"], "级": info["stage"], "缸": info["cylinder"], "运行覆盖率": round(row.get("coverage") or 0.0, 4), "均值": row.get(f"{column}__avg"), "最小值": row.get(f"{column}__min"), "最大值": row.get(f"{column}__max"), "标准差": row.get(f"{column}__std"), "原始最小值": row.get(f"{column}__minraw"), "零值样本数": row.get(f"{column}__zeros"), "高报警样本数": row.get(f"{column}__high"), "高高报警样本数": row.get(f"{column}__high_high"), }) return output def main() -> None: parser = argparse.ArgumentParser(description="PKS 缸体温度扫描（4#~9#）") parser.add_argument("--device", nargs="+", default=["4#", "5#", "6#", "7#", "8#", "9#"]) parser.add_argument("--start") parser.add_argument("--end") parser.add_argument("--chunk-days", type=int, default=30) parser.add_argument("--include-packing", action="store_true", help="5#/6# 同时扫描气缸填料温度") parser.add_argument("--output-dir", default=str(OUT_DIR)) parser.add_argument("--no-hourly-long", action="store_true") for name in ("mild_spread", "abnormal_spread", "severe_spread", "mild_peer", "abnormal_peer", "severe_peer", "mild_peer_low", "abnormal_peer_low", "severe_peer_low", "mild_z", "abnormal_z", "severe_z", "mild_delta", "abnormal_delta", "severe_delta"): parser.add_argument(f"--{name.replace('_', '-')}", type=float) parser.add_argument("--confirm-hours", type=int) parser.add_argument("--min-hours", type=int, default=1, help="最短输出段小时数（默认1，观察段也输出）") args = parser.parse_args() selected = normalize_devices(args.device) overrides = threshold_overrides(args) all_rows: list[dict] = [] all_qc: list[dict] = [] all_features: list[dict] = [] all_segments: list[dict] = [] all_cycles: list[dict] = [] all_channels: list[dict] = [] all_observations: list[dict] = [] all_hf: list[dict] = [] all_long: list[dict] = [] for device in selected: unit = int(device[0]) config = build_unit_config(unit, include_packing=args.include_packing) if args.start or args.end: start = parse_time(args.start) if args.start else None end = parse_time(args.end) if args.end else None if start is None or end is None: auto_start, auto_end = _auto_range(config) start = start or auto_start end = end or auto_end else: start, end = _auto_range(config) print(f"[{device}] 批次 {config.batches} {start} ~ {end} 扫描 {len(config.columns)} 个缸温点 ...", flush=True) rows = temperature_data.fetch_unit_hourly(config, start, end, args.chunk_days) qc, features, segments, cycles, channels, observations, hf_segments = temperature_algorithm.run(rows, config, overrides, args.min_hours) print( f"[{device}] 有效小时 {sum(1 for row in rows if row.get('valid')):,} " f"特征 {len(features):,} 段 {len(segments):,} 通道异常 {len(channels):,}", flush=True, ) all_rows.extend(rows) all_qc.extend(qc) all_features.extend(features) all_segments.extend(segments) all_cycles.extend(cycles) all_channels.extend(channels) all_observations.extend(observations) all_hf.extend(hf_segments) if not args.no_hourly_long: all_long.extend(hourly_long_rows(rows, config, args.include_packing)) output_dir = Path(args.output_dir) confirmed_hf, hf_candidates = hf_validation.filter_hf(all_hf) temperature_algorithm.write_results(output_dir, all_qc, all_features, all_segments, all_cycles, all_channels, all_observations, confirmed_hf, hf_candidates) if not args.no_hourly_long: fields = ["机组", "批次", "小时", "点位", "点位描述", "级", "缸", "运行覆盖率", "均值", "最小值", "最大值", "标准差", "原始最小值", "零值样本数", "高报警样本数", "高高报警样本数"] temperature_algorithm.write_csv(output_dir / "pks_temperature_hourly.csv", all_long, fields, {field: field for field in fields}) print(f"温度小时长表：{len(all_long):,} 行 -> {output_dir / 'pks_temperature_hourly.csv'}") print(f"结果目录：{output_dir}") for level in ("轻微", "异常", "严重异常"): print(f" {level}段：{sum(1 for segment in all_segments if segment['温度阶段'] == level):,}") print(f" 观察：{len(all_observations):,}") print(f" 高频候选：{len(hf_candidates):,} 高频确认：{len(confirmed_hf):,}") for path in ("温度小时特征.csv", "温度异常段.csv", "温度演化周期.csv", "温度通道质控.csv"): print(f" {path}") if __name__ == "__main__": main()