"""压缩机示功图（P-V）。 读取波库中该测点**最近一次运行**（``rpm>0``）的 ``wave_file`` 首个周期 （``wave_sample_one``），按缸径/行程/连杆/余隙把曲柄转角换算为容积（L）， 返回盖侧 / 轴侧两副压力功图（P-V）。 首周期本身即一个完整冲程。7/8/9 号机按 ``月度报告/沁水压缩机参数20250102.docx`` 的相位角把共享键相角旋转到各缸自身 0°（显示角 = 共享角 − 相位角）；4/5/6 号机暂无资料，保持原逻辑。 轴侧容积扣除活塞杆直径 78mm，盖侧用全缸径。 """ from __future__ import annotations import re from typing import Any import numpy as np from db import connect CYLINDERS = ("一", "二", "三", "四", "五", "六") SIDES = ("盖侧", "轴侧") CYLINDER_BORE_MM = {"一": 360.0, "二": 490.0, "三": 390.0, "四": 490.0, "五": 390.0, "六": 490.0} PISTON_STROKE_MM = 148.0 CONNECTING_ROD_LENGTH_MM = 460.0 PISTON_ROD_DIAMETER_MM = 78.0 CLEARANCE_VOLUME_L_BY_BORE = {490.0: 1.51, 390.0: 0.74, 360.0: 0.62} # 7/8/9 号机各缸相位角（月度报告/沁水压缩机参数20250102.docx）。 CYLINDER_PHASE_DEGREE = {"一": 0.0, "二": 0.0, "三": 120.0, "四": 240.0, "五": 240.0, "六": 120.0} UNIT_PHASE_OVERRIDE = { "7": { "一": 141.0, "二": 160.0, "三": 280.0, "四": 280.0, "五": 40.0, "六": 40.0, }, "8": { "一": 8.0, "二": 23.0, "三": 138.0, "四": 145.0, "五": 262.0, "六": 262.0, }, "9": { "一": 16.0, "二": 27.0, "三": 153.0, "四": 153.0, "五": 269.0, "六": 269.0, }, } PHASE_ROTATION_UNITS = frozenset({"7", "8", "9"}) PULSES_PER_REVOLUTION = 8 MIN_SAMPLES = 60 def unit_number(device_code: str) -> str | None: match = re.search(r"([789])", device_code or "") return match.group(1) if match else None def piecewise_angle(trigger: np.ndarray, count: int) -> np.ndarray: """Crank angle from the 8 keyphase pulses; falls back to uniform sampling.""" high = np.nan_to_num(np.asarray(trigger, dtype=float), nan=-np.inf) >= 30.0 changes = np.diff(np.r_[False, high, False].astype(np.int8)) offsets = [int(offset) for offset in np.flatnonzero(changes == 1)] if len(offsets) < PULSES_PER_REVOLUTION or offsets[0] != 0: return np.linspace(0.0, 360.0, count, endpoint=False) offsets = offsets[:PULSES_PER_REVOLUTION] bounds = offsets + [count] angle = np.full(count, np.nan, dtype=float) for pulse in range(PULSES_PER_REVOLUTION): start, end = bounds[pulse], min(bounds[pulse + 1], count) if end <= start: continue angle[start:end] = np.linspace( pulse * 45.0, (pulse + 1) * 45.0, end - start, endpoint=False, ) if np.isnan(angle).any(): good = np.flatnonzero(~np.isnan(angle)) if not len(good): return np.linspace(0.0, 360.0, count, endpoint=False) angle = np.interp(np.arange(count), good, angle[good]) return angle def keyphase_angle(keyphase_info) -> float | None: if not keyphase_info: return None parts = str(keyphase_info).strip().split("-") try: return float(parts[-1]) except (TypeError, ValueError): return None def phase_degrees(unit: str, cylinder: str, keyphase_info=None) -> float: if unit not in PHASE_ROTATION_UNITS: return 0.0 unit_override = UNIT_PHASE_OVERRIDE.get(unit) if unit_override and cylinder in unit_override: return unit_override[cylinder] angle = keyphase_angle(keyphase_info) if angle is not None: return angle return CYLINDER_PHASE_DEGREE[cylinder] def volume_from_angle( cylinder: str, side: str, unit: str, angle_degree: np.ndarray, ) -> np.ndarray: bore = CYLINDER_BORE_MM[cylinder] clearance = CLEARANCE_VOLUME_L_BY_BORE[bore] crank_radius = PISTON_STROKE_MM / 2.0 rod_length = CONNECTING_ROD_LENGTH_MM rod_diameter = ( PISTON_ROD_DIAMETER_MM if side == "轴侧" and unit in PHASE_ROTATION_UNITS else 0.0 ) theta = np.deg2rad(np.asarray(angle_degree, dtype=float)) travel = ( crank_radius * (1.0 - np.cos(theta)) + rod_length - np.sqrt(rod_length**2 - (crank_radius * np.sin(theta)) ** 2) ) piston_area = np.pi * ((bore / 2.0) ** 2 - (rod_diameter / 2.0) ** 2) return clearance + piston_area * travel / 1_000_000.0 def _side(unit: str, cylinder: str, side: str) -> dict[str, Any] | None: point = f"{unit}号机组{cylinder}缸压力{side}" connection = connect() try: with connection.cursor() as cursor: cursor.execute( """ SELECT id, sample_time, rpm, keyphase_info FROM wave_file WHERE point_name = %s AND measurement_type = '压力' AND rpm > 0 ORDER BY sample_time DESC LIMIT 1 """, (point,), ) row = cursor.fetchone() if not row: return None file_id = int(row["id"]) cursor.execute( "SELECT signal_value, second_value FROM wave_sample_one " "WHERE wave_file_id = %s ORDER BY sample_index", (file_id,), ) samples = [ (float(item["signal_value"]), item["second_value"]) for item in cursor.fetchall() if item["signal_value"] is not None ] finally: connection.close() if len(samples) < MIN_SAMPLES: return None pressure = np.asarray([item[0] for item in samples], dtype=float) trigger = np.asarray( [float(item[1]) if item[1] is not None else np.nan for item in samples], dtype=float, ) angle = piecewise_angle(trigger, len(pressure)) phase = phase_degrees(unit, cylinder, row.get("keyphase_info")) if phase: angle = np.mod(angle - phase, 360.0) volume = volume_from_angle(cylinder, side, unit, angle) sample_time = row["sample_time"] return { "side": side, "point_name": point, "wave_file_id": file_id, "sample_time": sample_time.strftime("%Y-%m-%d %H:%M:%S") if sample_time else None, "rpm": float(row["rpm"]) if row["rpm"] is not None else None, "pressure": [round(float(value), 4) for value in pressure.tolist()], "volume": [round(float(value), 5) for value in volume.tolist()], "angle": [round(float(value), 2) for value in angle.tolist()], } def query_indicator(device_code: str, cylinder: str) -> dict[str, Any]: unit = unit_number(device_code) if unit is None: raise ValueError(f"不支持的机组：{device_code}") if cylinder not in CYLINDERS: raise ValueError(f"不支持的缸号：{cylinder}") sides = [item for item in (_side(unit, cylinder, side) for side in SIDES) if item] return {"device_code": device_code, "unit": unit, "cylinder": cylinder, "sides": sides} __all__ = ["CYLINDERS", "SIDES", "query_indicator"]