from __future__ import annotations import re from collections import OrderedDict from datetime import datetime, timedelta from functools import lru_cache from time import monotonic from typing import Any, Callable import numpy as np from ..algorithms.cycles import ( DetectedCycle, PULSES_PER_REVOLUTION, build_angle_vector, detect_cycles, downsample_indices, ) from ..config import settings from ..db import get_connection MEASUREMENT_TYPES = ("位移", "加速度", "压力") MEASUREMENT_COLORS = { "压力": "#e4572e", "位移": "#1f7a8c", "加速度": "#7b61a8", } DEVICE_POINTS = ("压力盖侧", "压力轴侧", "活塞杆沉降", "十字头振动", "自由端振动", "驱动端振动") DEVICE_POINT_TO_TYPE = { "压力盖侧": "压力", "压力轴侧": "压力", "活塞杆沉降": "位移", "十字头振动": "加速度", "自由端振动": "加速度", "驱动端振动": "加速度", } PRIMARY_DEVICE_POINT = "压力盖侧" PHASES = ( ("排气", "#7b1fa2", 0.0, 120.0), ("压缩", "#c62828", 120.0, 195.0), ("膨胀", "#1565c0", 195.0, 270.0), ("进气", "#2e7d32", 270.0, 330.0), ) ANNOTATION_LABELS = ("正常", "异常") # PKS 全场点位：机组号 -> pks_long_sample.import_batch_id # 7号机=30、8号机=31、9号机=32。 UNIT_BATCH = {"7": 30, "8": 31, "9": 32} _SITE_POINT_PATTERN = re.compile(r"^YSJ([789])_([1-9]|1[0-9]|2[0-9]|3[0-9]|4[0-1])$") # Extra samples fetched past cycle_end so detect_cycles can see the zero marker # that closes the first cycle (cycle_end is exclusive, the marker sits at it). FIRST_CYCLE_PAD = 64 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 CLEARANCE_VOLUME_L_BY_BORE = { 490.0: 1.51, 390.0: 0.74, 360.0: 0.62, } DEMO_POINT_NAME = "7号机组一缸压力盖侧" DEMO_DEVICE_PART = "7号机组一缸" DEMO_START = datetime(2026, 4, 12, 8, 0, 0) DEMO_POINT_COUNT = 72 DEMO_SAMPLE_COUNT = 32768 DEMO_REVOLUTION_SAMPLES = 800 DEMO_ID_BY_TYPE = {name: 100000 + index * 1000 for index, name in enumerate(MEASUREMENT_TYPES)} def _time_string(value: Any) -> str: if isinstance(value, datetime): return value.strftime("%Y-%m-%d %H:%M:%S") return str(value) def _parse_time(value: str | None) -> datetime | None: if not value: return None return datetime.fromisoformat(value.replace("Z", "+00:00").replace("T", " ")) def _safe_float(value: Any) -> float | None: if value is None: return None number = float(value) return number if np.isfinite(number) else None def _series_extent(items: list[dict[str, Any]]) -> tuple[float, float]: """Whole-window min/max over a series' finite raw values.""" if not items: return (0.0, 1.0) values = np.fromiter((item["rawValue"] for item in items), dtype=float, count=len(items)) values = values[np.isfinite(values)] if values.size == 0: return (0.0, 1.0) return (float(values.min()), float(values.max())) def _annotation_dict(row: dict[str, Any]) -> dict[str, Any]: return { "id": int(row["id"]), "waveFileId": int(row["wave_file_id"]), "label": row["label"], "periodStart": int(row["period_start"]), "periodEnd": int(row["period_end"]), "sampleIndexStart": int(row["sample_index_start"]), "sampleIndexEnd": int(row["sample_index_end"]), } def _validate_device_points(values: list[str] | tuple[str, ...] | None) -> list[str]: selected = list(values or DEVICE_POINTS) if not selected: raise ValueError("请至少选择一个测试点位") invalid = [value for value in selected if value not in DEVICE_POINTS] if invalid: raise ValueError(f"不支持的测试点位：{'、'.join(invalid)}") return [value for value in DEVICE_POINTS if value in selected] def _primary_device_point(points: list[str]) -> str: """Prefer pressure cap, then pressure shaft, then any selected point.""" for point in ("压力盖侧", "压力轴侧"): if point in points: return point return points[0] def _unit_number(device_part: str) -> str | None: """从 机组与部位 提取机组号（7/8/9），非 7/8/9 机组返回 None。""" match = re.match(r"^([789])号机组", device_part.strip()) return match.group(1) if match else None def _site_point_column(item_name: str, unit: str | None) -> str | None: """校验全场点位名称并映射到 pks_long_sample 的列名（如 YSJ7_3 -> YSJ_3）。""" match = _SITE_POINT_PATTERN.match(item_name) if not match or match.group(1) != unit: return None return f"YSJ_{match.group(2)}" class DataService: # 数据库失败后的重试冷却时间（秒）。超过该时间后自动重连数据库， # 避免一次网络抖动就把服务永久锁死在演示数据模式。 DB_RETRY_COOLDOWN = 30.0 def __init__(self) -> None: self._db_failed = settings.demo_mode == "always" self._db_failed_at = monotonic() if self._db_failed else 0.0 self._last_db_error = "" self._demo_annotations: dict[int, dict[str, Any]] = {} self._demo_annotation_seq = 1 @property def source(self) -> str: return "demo" if self._db_failed else "database" @property def last_db_error(self) -> str: return self._last_db_error def _run_with_fallback( self, database_function: Callable[[], Any], demo_function: Callable[[], Any], ) -> tuple[Any, str]: if self._db_failed: if settings.demo_mode == "always": return demo_function(), "demo" if monotonic() - self._db_failed_at < self.DB_RETRY_COOLDOWN: return demo_function(), "demo" # 冷却结束，重新尝试数据库，数据库恢复后可自动切回真实数据。 try: result = database_function() self._db_failed = False self._last_db_error = "" return result, "database" except Exception as error: if settings.demo_mode == "never": raise self._db_failed = True self._db_failed_at = monotonic() self._last_db_error = str(error) return demo_function(), "demo" def query_options(self) -> dict[str, Any]: def database_query(): with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( """ SELECT device_part, device_point, measurement_type, MAX(sample_time) AS max_time, MIN(sample_time) AS min_time, COUNT(*) AS file_count FROM wave_file WHERE rpm > 0 AND device_part <> '' AND device_point <> '' GROUP BY device_part, device_point, measurement_type ORDER BY device_part, device_point """, ) rows = cursor.fetchall() return [ { "devicePart": row["device_part"], "devicePoint": row["device_point"], "measurementType": row["measurement_type"], "minTime": _time_string(row["min_time"]), "maxTime": _time_string(row["max_time"]), "fileCount": int(row["file_count"]), } for row in rows ] rows, source = self._run_with_fallback(database_query, self._demo_options) device_parts = list(dict.fromkeys(row["devicePart"] for row in rows)) return { "source": source, "measurementTypes": list(MEASUREMENT_TYPES), "deviceParts": device_parts, "devicePoints": list(DEVICE_POINTS), "devicePointToType": dict(DEVICE_POINT_TO_TYPE), "options": rows, "notice": self._source_notice(source), } def abnormal_counts(self) -> dict[str, int]: """每个 point_name 的异常文件数量（tspluse_status > 0）。""" def database_query(): with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( """ SELECT point_name, COUNT(*) AS cnt FROM wave_file WHERE tspluse_status > 0 AND point_name <> '' GROUP BY point_name """, ) rows = cursor.fetchall() return {row["point_name"]: int(row["cnt"]) for row in rows} def demo_query(): return {} result, _source = self._run_with_fallback(database_query, demo_query) return result def tspluse_ruler(self) -> dict[str, Any]: """压力部位 tspluse_status 的全局标尺，进入页面时只查询一次。""" def database_query(): with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( """ SELECT MIN(tspluse_status) AS min_status, MAX(tspluse_status) AS max_status FROM wave_file WHERE rpm > 0 AND measurement_type = '压力' """, ) row = cursor.fetchone() return { "min": int(row["min_status"]) if row and row["min_status"] is not None else 0, "max": int(row["max_status"]) if row and row["max_status"] is not None else 0, } def demo_query(): return {"min": 0, "max": 0} result, source = self._run_with_fallback(database_query, demo_query) result["source"] = source return result def site_points(self, device_part: str) -> dict[str, Any]: """机组(7/8/9)的全场点位记录（site_point 表中 YSJ{机组号}_1..41）。""" unit = _unit_number(device_part) def database_query(): if unit is None: return [] allowed = {f"YSJ{unit}_{n}" for n in range(1, 42)} with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( "SELECT ItemName, ItemDescription FROM site_point WHERE ItemName LIKE %s", (f"YSJ{unit}\\_%",), ) rows = cursor.fetchall() items = [ {"itemName": row["ItemName"], "itemDescription": row["ItemDescription"] or ""} for row in rows if row["ItemName"] in allowed ] items.sort(key=lambda item: int(item["itemName"].rsplit("_", 1)[1])) return items def demo_query(): return [] items, source = self._run_with_fallback(database_query, demo_query) return { "source": source, "unit": unit, "items": items, "notice": self._source_notice(source), } @staticmethod def _attach_site_values(points: list[dict[str, Any]], unit: str, site_points: list[str]) -> None: """把 pks_long_sample 的最近邻值挂到每个时间点的 siteValues 上。 pks 数据 5 秒一条、wave_file 15 分钟一条。按用户口径做分钟/5秒级对齐： 把 wave 采样时刻四舍五入到最近的 5 秒格点，再用一次 ``sample_time IN (...)`` 精确取数（结果行数 = 时间点数），避免把整段 pks 拉出来。 """ columns = [ (item_name, column) for item_name in site_points if (column := _site_point_column(item_name, unit)) is not None ] if not columns or not points: for point in points: point.setdefault("siteValues", {}) return rounded: list[datetime] = [] for point in points: timestamp = datetime.strptime(point["sampleTime"], "%Y-%m-%d %H:%M:%S") rounded.append(datetime.fromtimestamp(round(timestamp.timestamp() / 5.0) * 5)) placeholders = ", ".join(["%s"] * len(rounded)) for item_name, column in columns: with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( f"SELECT sample_time, `{column}` AS value FROM pks_long_sample " f"WHERE import_batch_id = %s AND sample_time IN ({placeholders})", (UNIT_BATCH[unit], *rounded), ) rows = cursor.fetchall() by_time = {row["sample_time"]: row["value"] for row in rows} for index, target_time in enumerate(rounded): value = by_time.get(target_time) points[index].setdefault("siteValues", {})[item_name] = ( float(value) if value is not None else None ) @staticmethod def _build_site_series(site_points: list[str], points: list[dict[str, Any]]) -> dict[str, Any]: """把时间点上的 siteValues 整理成每周期一个点的曲线系列。""" series = [] for item_name in site_points: data = [] for index, point in enumerate(points): value = (point.get("siteValues") or {}).get(item_name) if value is not None: data.append( { "value": [index, float(value)], "x": index, "sampleTime": point["sampleTime"], }, ) series.append({"itemName": item_name, "data": data}) return {"points": site_points, "series": series} def time_points( self, device_part: str, device_points: list[str] | None, min_time: str | None, max_time: str | None, include_stopped: bool = False, min_status: int | None = None, status_filter: list[str] | None = None, site_points: list[str] | None = None, ) -> dict[str, Any]: if not device_part.strip(): raise ValueError("机组与部位不能为空") selected_points = _validate_device_points(device_points) point_names = [f"{device_part}{point}" for point in selected_points] status_filters = {value for value in (status_filter or [])} unknown = status_filters - {"abnormal", "no_cycle"} if unknown: raise ValueError(f"不支持的状态筛选：{'、'.join(sorted(unknown))}") start = _parse_time(min_time) end = _parse_time(max_time) if start and end and start > end: raise ValueError("开始时间不能晚于结束时间") def database_query(): point_placeholders = ", ".join(["%s"] * len(point_names)) clauses = [ f"point_name IN ({point_placeholders})", ] params: list[Any] = list(point_names) if not include_stopped: clauses.append("rpm > 0") if min_status is not None and min_status > 0 and "no_cycle" not in status_filters: clauses.append("tspluse_status >= %s") params.append(min_status) if "abnormal" in status_filters and "no_cycle" in status_filters: clauses.append("(tspluse_status > 0 OR tspluse_status = -1)") elif "abnormal" in status_filters: clauses.append("tspluse_status > 0") elif "no_cycle" in status_filters: clauses.append("tspluse_status = -1") if start: clauses.append("sample_time >= %s") params.append(start) if end: clauses.append("sample_time <= %s") params.append(end) with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( f""" SELECT id, point_name, device_point, measurement_type, sample_time, sample_count, sample_frequency_hz, rpm, tspluse_status FROM wave_file WHERE {' AND '.join(clauses)} ORDER BY sample_time ASC, id ASC """, params, ) rows = cursor.fetchall() reference_rows: list[dict[str, Any]] = [] if status_filters and rows: primary = _primary_device_point(selected_points) reference_where = [ "point_name = %s", "measurement_type = '压力'", "tspluse_status = 0", ] reference_params: list[Any] = [f"{device_part}{primary}"] if not include_stopped: reference_where.append("rpm > 0") if start: reference_where.append("sample_time >= %s") reference_params.append(start) if end: reference_where.append("sample_time <= %s") reference_params.append(end) reference_params.append(rows[0]["sample_time"]) with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( f""" SELECT id, point_name, device_point, measurement_type, sample_time, sample_count, sample_frequency_hz, rpm, tspluse_status FROM wave_file WHERE {' AND '.join(reference_where)} ORDER BY ABS(TIMESTAMPDIFF(SECOND, sample_time, %s)) ASC LIMIT 1 """, reference_params, ) reference_rows = cursor.fetchall() return self._group_time_points(rows), self._group_time_points(reference_rows) def demo_query(): return self._demo_time_points(device_part, selected_points, start, end), [] result, source = self._run_with_fallback(database_query, demo_query) points, reference = result # 全场点位(PKS)数据是辅助层：任意失败都静默跳过，不影响主查询。 try: unit = _unit_number(device_part) if unit and site_points: self._attach_site_values(points, unit, site_points) except Exception: pass for point in points: point.setdefault("siteValues", {}) return { "source": source, "devicePart": device_part, "devicePoints": selected_points, "total": len(points), "points": points, "referencePoints": reference, "notice": self._source_notice(source), } @staticmethod def _group_time_points(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: grouped: OrderedDict[Any, dict[str, Any]] = OrderedDict() for row in rows: # Channels from one acquisition batch can be stamped a few seconds # apart (e.g. 10:30:02 vs 10:30:05), yet the acquisition cadence is # 15 minutes. Align on the minute so one batch is one time point. sample_time = row["sample_time"] key = str(sample_time)[:16] point = grouped.setdefault( key, { "sampleTime": _time_string(sample_time), "files": {}, }, ) device_point = row["device_point"] point["files"].setdefault(device_point, { "id": int(row["id"]), "devicePoint": device_point, "measurementType": row["measurement_type"], "sampleCount": int(row["sample_count"] or 0), "sampleFrequencyHz": int(row["sample_frequency_hz"] or 0), "rpm": float(row["rpm"] or 0), "status": int(row.get("tspluse_status") or 0), }) return [ {"index": index, **point} for index, point in enumerate(grouped.values()) ] def wave_window( self, device_part: str, device_points: list[str], points: list[dict[str, Any]], max_points: int, no_sampling: bool = False, first_cycle_only: bool = False, site_points: list[str] | None = None, ) -> dict[str, Any]: selected_points = _validate_device_points(device_points) if not points: raise ValueError("至少选择一个时间点") if len(points) > 200: raise ValueError("单次最多预览 200 个时间点，请缩小时间窗口") max_points = min(max(int(max_points), 256), 200000) def database_query(): if first_cycle_only: return self._build_first_cycle_window( device_part, selected_points, points, self._load_db_wave, ) return self._build_wave_window( device_part, selected_points, points, max_points, self._load_db_wave, no_sampling, ) def demo_query(): if first_cycle_only: return self._build_first_cycle_window( device_part, selected_points, points, self._load_demo_wave, ) return self._build_wave_window( device_part, selected_points, points, max_points, self._load_demo_wave, no_sampling, ) result, source = self._run_with_fallback(database_query, demo_query) # 全场点位(PKS)系列：仅首个周期模式叠加，每周期一个点。 if first_cycle_only and site_points: try: result["siteSeries"] = self._build_site_series(site_points, result["points"]) except Exception: result["siteSeries"] = {"points": [], "series": []} else: result["siteSeries"] = {"points": [], "series": []} result["source"] = source result["notice"] = self._source_notice(source) return result def period_detail(self, wave_file_id: int, period_number: int) -> dict[str, Any]: if wave_file_id <= 0 or period_number <= 0: raise ValueError("wave_file_id 和周期编号必须为正整数") def database_query(): metadata, samples = self._load_db_wave(wave_file_id) return self._build_period_detail(metadata, samples, period_number) def demo_query(): metadata, samples = self._load_demo_wave( wave_file_id, "压力", DEMO_POINT_NAME, DEMO_START, ) return self._build_period_detail(metadata, samples, period_number) result, source = self._run_with_fallback(database_query, demo_query) result["source"] = source result["notice"] = self._source_notice(source) return result def annotation_config(self) -> dict[str, Any]: return { "source": self.source, "annotationWidth": settings.annotation_width, "notice": self._source_notice(self.source), } def list_annotations(self, wave_file_ids: list[int]) -> dict[str, Any]: ids = sorted({int(value) for value in wave_file_ids if value}) if not ids: return { "source": self.source, "annotations": [], "notice": self._source_notice(self.source), } def database_query(): placeholders = ", ".join(["%s"] * len(ids)) with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( f""" SELECT id, wave_file_id, label, period_start, period_end, sample_index_start, sample_index_end FROM wave_annotation WHERE wave_file_id IN ({placeholders}) ORDER BY id ASC """, ids, ) rows = cursor.fetchall() return [_annotation_dict(row) for row in rows] def demo_query(): return [ _annotation_dict(annotation) for annotation in self._demo_annotations.values() if annotation["wave_file_id"] in ids ] annotations, source = self._run_with_fallback(database_query, demo_query) return { "source": source, "annotations": annotations, "notice": self._source_notice(source), } def create_annotation(self, payload: dict[str, Any]) -> dict[str, Any]: self._validate_annotation(payload) wave_file_id = int(payload["wave_file_id"]) label = payload["label"] period_start = int(payload["period_start"]) period_end = int(payload["period_end"]) sample_index_start = int(payload["sample_index_start"]) sample_index_end = int(payload["sample_index_end"]) def database_query(): with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( """ INSERT INTO wave_annotation (wave_file_id, label, period_start, period_end, sample_index_start, sample_index_end) VALUES (%s, %s, %s, %s, %s, %s) """, ( wave_file_id, label, period_start, period_end, sample_index_start, sample_index_end, ), ) annotation_id = cursor.lastrowid cursor.execute( """ SELECT id, wave_file_id, label, period_start, period_end, sample_index_start, sample_index_end FROM wave_annotation WHERE id = %s """, (annotation_id,), ) return _annotation_dict(cursor.fetchone()) def demo_query(): annotation_id = self._demo_annotation_seq self._demo_annotation_seq += 1 annotation = { "id": annotation_id, "wave_file_id": wave_file_id, "label": label, "period_start": period_start, "period_end": period_end, "sample_index_start": sample_index_start, "sample_index_end": sample_index_end, } self._demo_annotations[annotation_id] = annotation return _annotation_dict(annotation) result, source = self._run_with_fallback(database_query, demo_query) result["source"] = source result["notice"] = self._source_notice(source) return result def delete_annotation(self, annotation_id: int) -> dict[str, Any]: if annotation_id <= 0: raise ValueError("标注 id 必须为正整数") def database_query(): with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( "DELETE FROM wave_annotation WHERE id = %s", (annotation_id,), ) return int(cursor.rowcount) def demo_query(): if annotation_id not in self._demo_annotations: return 0 del self._demo_annotations[annotation_id] return 1 deleted, source = self._run_with_fallback(database_query, demo_query) if not deleted: raise ValueError(f"标注 id={annotation_id} 不存在") return { "deleted": annotation_id, "source": source, "notice": self._source_notice(source), } @staticmethod def _validate_annotation(payload: dict[str, Any]) -> None: label = payload.get("label") if label not in ANNOTATION_LABELS: raise ValueError("样本类型只能是 正常 或 异常") for field in ("wave_file_id", "period_start", "period_end", "sample_index_start", "sample_index_end"): if payload.get(field) is None: raise ValueError(f"{field} 不能为空") if int(payload["wave_file_id"]) <= 0: raise ValueError("wave_file_id 必须为正整数") if int(payload["period_start"]) <= 0 or int(payload["period_end"]) <= 0: raise ValueError("周期编号必须为正整数") if int(payload["period_start"]) > int(payload["period_end"]): raise ValueError("起始周期不能大于结束周期") if int(payload["sample_index_start"]) < 0 or int(payload["sample_index_end"]) < 0: raise ValueError("采样点索引不能为负") if int(payload["sample_index_start"]) > int(payload["sample_index_end"]): raise ValueError("起始采样点不能大于结束采样点") def health(self) -> dict[str, Any]: return { "status": "ok", "source": self.source, "databaseError": self._last_db_error or None, } def _source_notice(self, source: str) -> str | None: if source == "demo": if self._last_db_error: return f"当前为演示数据：数据库暂不可用（{self._last_db_error}）" return "当前为演示数据：可设置 DEMO_MODE=never 强制使用数据库" return "已连接 MySQL 数据库" def _build_wave_window( self, device_part: str, device_points: list[str], points: list[dict[str, Any]], max_points: int, loader: Callable[..., tuple[dict[str, Any], np.ndarray]], no_sampling: bool = False, ) -> dict[str, Any]: series_data: dict[str, list[dict[str, Any]]] = {point: [] for point in device_points} angle_data: list[dict[str, Any]] = [] volume_data: list[dict[str, Any]] = [] volume_info: dict[str, Any] | None = None cycles: list[dict[str, Any]] = [] triggers: list[float] = [] files: list[dict[str, Any]] = [] diagnostics: list[dict[str, Any]] = [] second_series_data: list[dict[str, Any]] = [] second_finite_count = 0 second_non_zero_count = 0 second_min: float | None = None second_max: float | None = None primary = _primary_device_point(device_points) primary_type = DEVICE_POINT_TO_TYPE[primary] pressure_points = [point for point in device_points if DEVICE_POINT_TO_TYPE[point] == "压力"] load_cache: dict[int, tuple[dict[str, Any], np.ndarray]] = {} for slot, point in enumerate(points): target_per_file = max(256, int(np.ceil(max_points / max(len(points), 1)))) slot_files: dict[str, dict[str, Any]] = {} for device_point in device_points: file_info = point.get("files", {}).get(device_point) if file_info: slot_files[device_point] = file_info # The per-slot "period source" supplies the 周期数据/体积/角度 series # and the background cycle bands. Primary point is preferred; when a # selected point's timestamp differs (e.g. 10:30:02 vs 10:30:05), a # slot may only contain another device point, which is used instead # so the period data and cycle bands do not disappear there. source_point = primary if primary in slot_files else (next(iter(slot_files)) if slot_files else None) source_samples: np.ndarray | None = None source_detected: list[Any] = [] source_angle_full: np.ndarray | None = None source_volume: np.ndarray | None = None source_required: set[int] | None = None source_id: int | None = None source_type = "" if source_point is not None: source_info = slot_files[source_point] source_id = int(source_info["id"] if isinstance(source_info, dict) else source_info) source_type = DEVICE_POINT_TO_TYPE[source_point] try: source_meta, source_samples = self._load_for_window( loader, source_id, source_type, device_part + source_point, point["sampleTime"], load_cache, ) except ValueError: source_samples = None if source_samples is not None and len(source_samples): source_detected, source_diagnostic = detect_cycles(source_samples) source_angle_vector = build_angle_vector(len(source_samples), source_detected) source_angle_full = np.full(len(source_samples), np.nan, dtype=float) for detected_cycle in source_detected: source_angle_full[detected_cycle.start_offset:detected_cycle.end_offset] = detected_cycle.angle source_volume, current_volume_info = self._build_volume_vector( len(source_samples), source_detected, device_part + primary, ) if current_volume_info is not None: volume_info = current_volume_info source_indices = source_samples[:, 0].astype(np.int64) source_span = max(len(source_samples), 1) finite_second = source_samples[:, 2][np.isfinite(source_samples[:, 2])] if len(finite_second): second_finite_count += int(len(finite_second)) second_non_zero_count += int(np.count_nonzero(finite_second != 0)) current_min = float(np.min(finite_second)) current_max = float(np.max(finite_second)) second_min = current_min if second_min is None else min(second_min, current_min) second_max = current_max if second_max is None else max(second_max, current_max) source_required = {0, len(source_samples) - 1} for cycle in source_detected: source_required.update( { cycle.start_offset, max(cycle.end_offset - 1, cycle.start_offset), *cycle.trigger_offsets, }, ) start_x = slot + cycle.start_offset / source_span end_x = slot + cycle.end_offset / source_span cycles.append( { "id": f"{source_id}:{cycle.number}", "waveFileId": source_id, "periodNo": cycle.number, "pointIndex": slot, "sampleTime": point["sampleTime"], "startX": start_x, "endX": end_x, "startSampleIndex": int(source_indices[cycle.start_offset]), "endSampleIndex": int( source_indices[max(cycle.end_offset - 1, cycle.start_offset)], ), "sourceType": source_type, "devicePoint": source_point, "background": True, }, ) for trigger_offset in sorted(source_required): if any( trigger_offset == run_offset for cycle in source_detected for run_offset in cycle.trigger_offsets ): triggers.append(slot + trigger_offset / source_span) diagnostics.append( { "waveFileId": source_id, "measurementType": source_type, "devicePoint": source_point, "sampleTime": point["sampleTime"], **source_diagnostic, }, ) source_base_index = int(source_indices[0]) second_chosen = downsample_indices(source_samples[:, 2], target_per_file, source_required) for offset in second_chosen: second_value = _safe_float(source_samples[offset, 2]) if second_value is None: continue x = slot + (int(source_indices[offset]) - source_base_index) / source_span second_series_data.append( { "value": [x, second_value], "x": x, "rawValue": second_value, "sampleIndex": int(source_indices[offset]), "waveFileId": source_id, "sampleTime": point["sampleTime"], }, ) angle_chosen = downsample_indices(source_angle_vector, target_per_file, source_required) for offset in angle_chosen: value = _safe_float(source_angle_vector[offset]) if value is None: continue x = slot + (int(source_indices[offset]) - source_base_index) / source_span angle_data.append( { "value": [x, value], "x": x, "angle": value, "sampleIndex": int(source_indices[offset]), "waveFileId": source_id, "sampleTime": point["sampleTime"], }, ) volume_chosen = downsample_indices(source_volume, target_per_file, source_required) for offset in volume_chosen: value = _safe_float(source_volume[offset]) if value is None: continue x = slot + (int(source_indices[offset]) - source_base_index) / source_span volume_data.append( { "value": [x, value], "x": x, "volume": value, "sampleIndex": int(source_indices[offset]), "waveFileId": source_id, "sampleTime": point["sampleTime"], }, ) for device_point in device_points: file_info = slot_files.get(device_point) if not file_info: continue file_id = int(file_info["id"] if isinstance(file_info, dict) else file_info) measurement_type = DEVICE_POINT_TO_TYPE[device_point] try: metadata, samples = self._load_for_window( loader, file_id, measurement_type, device_part + device_point, point["sampleTime"], load_cache, ) except ValueError: continue sample_count = len(samples) if not sample_count: continue sample_indices_for_file = samples[:, 0].astype(np.int64) file_base_index = int(sample_indices_for_file[0]) file_span = max(sample_count, 1) file_required = {0, sample_count - 1} own_volume: np.ndarray | None = None own_angle: np.ndarray | None = None if device_point in pressure_points: if source_point == device_point and source_samples is not None: own_volume = source_volume own_angle = source_angle_full if source_required is not None: file_required.update(source_required) else: detected_own, _ = detect_cycles(samples) for cycle in detected_own: file_required.update( { cycle.start_offset, max(cycle.end_offset - 1, cycle.start_offset), *cycle.trigger_offsets, }, ) cycles.append( { "id": f"{file_id}:{cycle.number}", "waveFileId": file_id, "periodNo": cycle.number, "pointIndex": slot, "sampleTime": point["sampleTime"], "startX": slot + cycle.start_offset / file_span, "endX": slot + cycle.end_offset / file_span, "startSampleIndex": int(sample_indices_for_file[cycle.start_offset]), "endSampleIndex": int( sample_indices_for_file[max(cycle.end_offset - 1, cycle.start_offset)], ), "sourceType": measurement_type, "devicePoint": device_point, "background": False, }, ) if detected_own: angle_own = build_angle_vector(len(samples), detected_own) full_own = np.full(len(samples), np.nan, dtype=float) for detected_cycle in detected_own: full_own[detected_cycle.start_offset:detected_cycle.end_offset] = detected_cycle.angle own_volume, _ = self._build_volume_vector( len(samples), detected_own, device_part + device_point, ) own_angle = full_own if no_sampling: target_per_file_own = sample_count else: target_per_file_own = target_per_file chosen = downsample_indices(samples[:, 1], target_per_file_own, file_required) for offset in chosen: x = slot + (int(sample_indices_for_file[offset]) - file_base_index) / file_span raw_value = float(samples[offset, 1]) series_data[device_point].append( { "value": [x, raw_value], "x": x, "rawValue": raw_value, "sampleIndex": int(sample_indices_for_file[offset]), "waveFileId": file_id, "sampleTime": point["sampleTime"], "secondValue": _safe_float(samples[offset, 2]), "volume": _safe_float(own_volume[offset]) if own_volume is not None else None, "angle360": _safe_float(own_angle[offset]) if own_angle is not None else None, }, ) files.append( { "id": file_id, "pointIndex": slot, "sampleTime": point["sampleTime"], "devicePoint": device_point, "measurementType": measurement_type, "sampleCount": int(metadata.get("sample_count") or sample_count), "sampleFrequencyHz": int(metadata.get("sample_frequency_hz") or 0), "pointName": str(metadata.get("point_name") or device_part + device_point), "rpm": float(metadata.get("rpm") or 0), "status": int(metadata.get("tspluse_status") or 0), "fileName": str(metadata.get("file_name") or ""), }, ) load_cache.clear() for device_point in series_data: series_data[device_point].sort(key=lambda item: item["x"]) second_series_data.sort(key=lambda item: item["x"]) angle_data.sort(key=lambda item: item["x"]) volume_data.sort(key=lambda item: item["x"]) extents = {device_point: _series_extent(series_data[device_point]) for device_point in device_points} return { "devicePart": device_part, "devicePoints": device_points, "primaryPoint": primary, "points": points, "xMin": 0, "xMax": len(points), "series": [ { "devicePoint": device_point, "measurementType": DEVICE_POINT_TO_TYPE[device_point], "color": MEASUREMENT_COLORS[DEVICE_POINT_TO_TYPE[device_point]], "data": series_data[device_point], "min": extents[device_point][0], "max": extents[device_point][1], } for device_point in device_points ], "secondSeries": { "name": "周期数据", "color": "#f56c6c", "sourceMeasurementType": primary_type, "sourceDevicePoint": primary, "data": second_series_data, "finiteCount": second_finite_count, "nonZeroCount": second_non_zero_count, "min": second_min, "max": second_max, }, "angleSeries": { "color": "#d59b2b", "data": angle_data, }, "volumeSeries": { "color": "#4d9e6f", "data": volume_data, "info": volume_info, }, "cycles": cycles, "triggerXs": sorted(set(triggers)), "files": files, "diagnostics": diagnostics, } def _build_first_cycle_window( self, device_part: str, device_points: list[str], points: list[dict[str, Any]], loader: Callable[..., tuple[dict[str, Any], np.ndarray]], ) -> dict[str, Any]: """Build a slot-layout window where each file shows only its first cycle. The primary point is selected in pressure-cap, pressure-shaft, then selected-point order. It is the point used for the first-cycle slice; files without recorded bounds still appear in the file list. The slice for every primary file is fetched in a single JOIN query (range scan on the ``(wave_file_id, sample_index)`` primary key). Other selected device points contribute no curve but still appear in the file list. The curve is continuous: all cycle samples are kept and each file spans its own x slot. """ primary = _primary_device_point(device_points) primary_type = DEVICE_POINT_TO_TYPE[primary] series_data: dict[str, list[dict[str, Any]]] = {point: [] for point in device_points} second_series_data: list[dict[str, Any]] = [] angle_data: list[dict[str, Any]] = [] volume_data: list[dict[str, Any]] = [] volume_info: dict[str, Any] | None = None cycles: list[dict[str, Any]] = [] triggers: list[float] = [] files: list[dict[str, Any]] = [] diagnostics: list[dict[str, Any]] = [] second_finite_count = 0 second_non_zero_count = 0 second_min: float | None = None second_max: float | None = None slot_sources: list[tuple[int, dict[str, Any], str, int]] = [] all_file_ids: set[int] = set() for slot, point in enumerate(points): for device_point in device_points: file_info = point.get("files", {}).get(device_point) if not file_info: continue file_id = int(file_info["id"] if isinstance(file_info, dict) else file_info) slot_sources.append((slot, point, device_point, file_id)) all_file_ids.add(file_id) if not slot_sources: return self._assemble_first_cycle_window( device_part, device_points, primary, points, series_data, second_series_data, angle_data, volume_data, volume_info, second_finite_count, second_non_zero_count, second_min, second_max, cycles, triggers, files, diagnostics, 0, ) is_demo = loader.__name__ == "_load_demo_wave" metas: dict[int, dict[str, Any]] = {} # file_id -> (cycle_start, cycle_end, padded_slice[offset, signal, second]) slices: dict[int, tuple[int, int, np.ndarray]] = {} if is_demo: for _slot, point, device_point, source_id in slot_sources: metadata, samples = self._load_for_window( loader, source_id, DEVICE_POINT_TO_TYPE[device_point], device_part + device_point, point["sampleTime"], {}, ) metas[source_id] = metadata detected, _ = detect_cycles(samples) if detected: cycle = detected[0] start_si = int(samples[cycle.start_offset, 0]) end_si = int(samples[cycle.end_offset, 0]) pad_end = min(cycle.end_offset + FIRST_CYCLE_PAD, len(samples)) slices[source_id] = (start_si, end_si, samples[cycle.start_offset:pad_end]) else: source_ids = [source_id for _, _, _, source_id in slot_sources] with get_connection() as connection: with connection.cursor() as cursor: placeholders = ", ".join(["%s"] * len(all_file_ids)) cursor.execute( f""" SELECT id, point_name, measurement_type, sample_time, sample_count, sample_frequency_hz, rpm, tspluse_status, file_name, cycle_start, cycle_end FROM wave_file WHERE id IN ({placeholders}) """, tuple(all_file_ids), ) for row in cursor.fetchall(): metas[int(row["id"])] = row source_placeholders = ", ".join(["%s"] * len(source_ids)) cursor.execute( f""" SELECT ws.wave_file_id, ws.sample_index, CAST(ws.signal_value AS FLOAT) AS sig, CAST(ws.second_value AS FLOAT) AS sec FROM wave_sample ws JOIN wave_file wf ON wf.id = ws.wave_file_id WHERE ws.wave_file_id IN ({source_placeholders}) AND wf.cycle_start >= 0 AND ws.sample_index >= wf.cycle_start AND ws.sample_index < wf.cycle_end + %s ORDER BY ws.wave_file_id, ws.sample_index """, (*tuple(source_ids), FIRST_CYCLE_PAD), ) raw: dict[int, list[tuple[float, float, float]]] = {} for row in cursor.fetchall(): raw.setdefault(int(row["wave_file_id"]), []).append( ( float(row["sample_index"]), float(row["sig"]), float(row["sec"]) if row["sec"] is not None else float("nan"), ), ) for file_id, rows in raw.items(): meta = metas.get(file_id) if meta is None or meta.get("cycle_start") is None: continue slices[file_id] = ( int(meta["cycle_start"]), int(meta["cycle_end"]), np.asarray(rows, dtype=float), ) # Per-slot period source: primary preferred, else the first selected # point with a file at that slot (timestamps may differ by seconds). source_by_slot: dict[int, str] = {} for _slot, _point, device_point, _file_id in slot_sources: if _slot not in source_by_slot: source_by_slot[_slot] = device_point for _slot, _point, device_point, _file_id in slot_sources: if device_point == primary: source_by_slot[_slot] = primary for slot, point, device_point, source_id in slot_sources: sample_time = point["sampleTime"] is_slot_source = device_point == source_by_slot.get(slot, device_point) cycle_bounds = slices.get(source_id) cycle_start = 0 cycle_end = 0 slice_arr: np.ndarray | None = None if cycle_bounds is not None: cycle_start, cycle_end, slice_arr = cycle_bounds cycle_len = max(cycle_end - cycle_start, 0) has_cycle = slice_arr is not None and 1 < cycle_len <= len(slice_arr) if has_cycle: angle: np.ndarray | None = None volume: np.ndarray | None = None point_volume_info: dict[str, Any] | None = None detected, _ = detect_cycles(slice_arr) if detected: cycle = detected[0] angle = cycle.angle volume, point_volume_info = self._build_volume_vector( len(slice_arr), [cycle], device_part + primary, ) if is_slot_source and point_volume_info is not None: volume_info = point_volume_info for offset in range(cycle_len): sample_index = int(slice_arr[offset, 0]) raw_value = float(slice_arr[offset, 1]) second = _safe_float(slice_arr[offset, 2]) angle360 = _safe_float(angle[offset]) if angle is not None and offset < len(angle) else None volume_value = _safe_float(volume[offset]) if volume is not None and offset < len(volume) else None x = slot + (sample_index - cycle_start) / cycle_len series_data[device_point].append( { "value": [x, raw_value], "x": x, "rawValue": raw_value, "sampleIndex": sample_index, "waveFileId": source_id, "sampleTime": sample_time, "secondValue": second, "volume": volume_value, "angle360": angle360, }, ) if is_slot_source: if volume_value is not None: volume_data.append( { "value": [x, volume_value], "x": x, "volume": volume_value, "sampleIndex": sample_index, "waveFileId": source_id, "sampleTime": sample_time, }, ) if angle360 is not None: display_angle = angle360 if angle360 <= 180.0 else 360.0 - angle360 angle_data.append( { "value": [x, display_angle], "x": x, "angle": display_angle, "sampleIndex": sample_index, "waveFileId": source_id, "sampleTime": sample_time, }, ) if second is not None: second_finite_count += 1 if second != 0: second_non_zero_count += 1 second_min = second if second_min is None else min(second_min, second) second_max = second if second_max is None else max(second_max, second) second_series_data.append( { "value": [x, second], "x": x, "rawValue": second, "sampleIndex": sample_index, "waveFileId": source_id, "sampleTime": sample_time, }, ) if offset > 0: prev = _safe_float(slice_arr[offset - 1, 2]) if second is not None and (prev is None or prev < 30) and second >= 30: triggers.append(x) cycles.append( { "id": f"{source_id}:1", "waveFileId": source_id, "periodNo": 1, "pointIndex": slot, "sampleTime": sample_time, "startX": float(slot), "endX": float(slot + 1), "startSampleIndex": cycle_start, "endSampleIndex": cycle_end - 1, "sourceType": DEVICE_POINT_TO_TYPE[device_point], "devicePoint": device_point, "background": is_slot_source, }, ) diagnostics.append( { "waveFileId": source_id, "measurementType": DEVICE_POINT_TO_TYPE[device_point], "devicePoint": device_point, "sampleTime": sample_time, "cycleStart": cycle_start, "cycleEnd": cycle_end, "cycleSampleCount": cycle_len, }, ) for slot, point in enumerate(points): for device_point in device_points: file_info = point.get("files", {}).get(device_point) if not file_info: continue file_id = int(file_info["id"] if isinstance(file_info, dict) else file_info) metadata = metas.get(file_id) cycle_start = metadata.get("cycle_start") if metadata else None cycle_end = metadata.get("cycle_end") if metadata else None files.append( { "id": file_id, "pointIndex": slot, "sampleTime": point["sampleTime"], "devicePoint": device_point, "measurementType": DEVICE_POINT_TO_TYPE[device_point], "sampleCount": int( (metadata.get("sample_count") if metadata else file_info.get("sampleCount") or 0) or 0, ), "sampleFrequencyHz": int( (metadata.get("sample_frequency_hz") if metadata else file_info.get("sampleFrequencyHz") or 0) or 0, ), "pointName": str( (metadata.get("point_name") if metadata else device_part + device_point) or device_part + device_point, ), "rpm": float((metadata.get("rpm") if metadata else file_info.get("rpm") or 0) or 0), "status": int( (metadata.get("tspluse_status") if metadata else file_info.get("status") or 0) or 0, ), "fileName": str((metadata.get("file_name") if metadata else "") or ""), "cycleStart": int(cycle_start) if cycle_start is not None else None, "cycleEnd": int(cycle_end) if cycle_end is not None else None, }, ) return self._assemble_first_cycle_window( device_part, device_points, primary, points, series_data, second_series_data, angle_data, volume_data, volume_info, second_finite_count, second_non_zero_count, second_min, second_max, cycles, triggers, files, diagnostics, max(len(slot_sources) - len(cycles), 0), ) @staticmethod def _assemble_first_cycle_window( device_part: str, device_points: list[str], primary: str, points: list[dict[str, Any]], series_data: dict[str, list[dict[str, Any]]], second_series_data: list[dict[str, Any]], angle_data: list[dict[str, Any]], volume_data: list[dict[str, Any]], volume_info: dict[str, Any] | None, second_finite_count: int, second_non_zero_count: int, second_min: float | None, second_max: float | None, cycles: list[dict[str, Any]], triggers: list[float], files: list[dict[str, Any]], diagnostics: list[dict[str, Any]], missing_cycles: int = 0, ) -> dict[str, Any]: for device_point in series_data: series_data[device_point].sort(key=lambda item: item["x"]) second_series_data.sort(key=lambda item: item["x"]) angle_data.sort(key=lambda item: item["x"]) volume_data.sort(key=lambda item: item["x"]) extents = {device_point: _series_extent(series_data[device_point]) for device_point in device_points} return { "devicePart": device_part, "devicePoints": device_points, "primaryPoint": primary, "points": points, "xMin": 0, "xMax": len(points), "series": [ { "devicePoint": device_point, "measurementType": DEVICE_POINT_TO_TYPE[device_point], "color": MEASUREMENT_COLORS[DEVICE_POINT_TO_TYPE[device_point]], "data": series_data[device_point], "min": extents[device_point][0], "max": extents[device_point][1], } for device_point in device_points ], "secondSeries": { "name": "周期数据", "color": "#f56c6c", "sourceMeasurementType": DEVICE_POINT_TO_TYPE[primary], "sourceDevicePoint": primary, "data": second_series_data, "finiteCount": second_finite_count, "nonZeroCount": second_non_zero_count, "min": second_min, "max": second_max, }, "angleSeries": { "color": "#d59b2b", "data": angle_data, }, "volumeSeries": { "color": "#4d9e6f", "data": volume_data, "info": volume_info, }, "cycles": cycles, "triggerXs": sorted(set(triggers)), "files": files, "diagnostics": diagnostics, "firstCycleMode": True, "firstCycleNotice": ( f"窗口内 {missing_cycles} 个文件尚未回写首个周期索引（cycle_start），" "请先运行 detect_cycle_index.py 回写后再查看。" if missing_cycles > 0 else None ), } @staticmethod def _build_volume_vector( sample_count: int, cycles: list[DetectedCycle], point_name: str, ) -> tuple[np.ndarray, dict[str, Any] | None]: volume = np.full(sample_count, np.nan, dtype=float) cylinder_name = next( (name for name in CYLINDER_BORE_MM if name in point_name), None, ) if cylinder_name is None or not cycles: return volume, None bore_mm = CYLINDER_BORE_MM[cylinder_name] clearance = CLEARANCE_VOLUME_L_BY_BORE[bore_mm] crank_radius = PISTON_STROKE_MM / 2.0 piston_area = np.pi * (bore_mm / 2.0) ** 2 for cycle in cycles: angle_rad = np.deg2rad(cycle.angle) travel = ( crank_radius * (1.0 - np.cos(angle_rad)) + CONNECTING_ROD_LENGTH_MM - np.sqrt( CONNECTING_ROD_LENGTH_MM**2 - (crank_radius * np.sin(angle_rad)) ** 2, ) ) volume[cycle.start_offset : cycle.end_offset] = ( clearance + piston_area * travel / 1_000_000.0 ) finite = volume[np.isfinite(volume)] return volume, { "cylinder": cylinder_name, "boreMm": bore_mm, "clearanceVolumeL": clearance, "minVolumeL": float(np.min(finite)) if len(finite) else None, "maxVolumeL": float(np.max(finite)) if len(finite) else None, } @staticmethod def _build_period_detail( metadata: dict[str, Any], samples: np.ndarray, period_number: int, ) -> dict[str, Any]: detected, diagnostics = detect_cycles(samples) cycle = next((item for item in detected if item.number == period_number), None) if cycle is None: raise ValueError(f"没有找到周期 {period_number}") angles360 = np.linspace(0.0, 359.0, 360) pressure = np.interp(angles360, cycle.angle, cycle.signal) display_angles = np.where(angles360 <= 180.0, angles360, 360.0 - angles360) point_name = str(metadata.get("point_name") or "") cylinder_name = next( (name for name in CYLINDER_BORE_MM if name in point_name), None, ) volume: np.ndarray | None = None volume_info: dict[str, Any] | None = None if cylinder_name: bore_mm = CYLINDER_BORE_MM[cylinder_name] clearance_volume = CLEARANCE_VOLUME_L_BY_BORE[bore_mm] angle_rad = np.deg2rad(angles360) crank_radius = PISTON_STROKE_MM / 2.0 piston_travel = ( crank_radius * (1.0 - np.cos(angle_rad)) + CONNECTING_ROD_LENGTH_MM - np.sqrt( CONNECTING_ROD_LENGTH_MM**2 - (crank_radius * np.sin(angle_rad)) ** 2, ) ) piston_area = np.pi * (bore_mm / 2.0) ** 2 volume = clearance_volume + piston_area * piston_travel / 1_000_000.0 volume_info = { "cylinder": cylinder_name, "boreMm": bore_mm, "clearanceVolumeL": clearance_volume, "minVolumeL": float(np.min(volume)), "maxVolumeL": float(np.max(volume)), } start_index = int(samples[cycle.start_offset, 0]) end_offset = min(cycle.end_offset, len(samples) - 1) end_index = int(samples[max(cycle.end_offset - 1, cycle.start_offset), 0]) return { "waveFile": { "id": int(metadata["id"]), "pointName": point_name, "measurementType": metadata.get("measurement_type"), "sampleTime": _time_string(metadata.get("sample_time")), "sampleFrequencyHz": int(metadata.get("sample_frequency_hz") or 0), "sampleCount": int(metadata.get("sample_count") or len(samples)), }, "period": { "periodNo": cycle.number, "startSampleIndex": start_index, "endSampleIndex": end_index, "sampleCount": int(cycle.end_offset - cycle.start_offset), "triggerSampleIndices": [ int(samples[offset, 0]) for offset in cycle.trigger_offsets if 0 <= offset < len(samples) ], }, "angles": display_angles.tolist(), "angles360": angles360.tolist(), "pressure": pressure.tolist(), "volume": volume.tolist() if volume is not None else None, "volumeInfo": volume_info, "phases": [ { "name": name, "color": color, "start": start, "end": end, } for name, color, start, end in PHASES ], "diagnostics": diagnostics, } @staticmethod def _load_for_window( loader: Callable[..., tuple[dict[str, Any], np.ndarray]], file_id: int, measurement_type: str, point_name: str, sample_time: str, load_cache: dict[int, tuple[dict[str, Any], np.ndarray]], ) -> tuple[dict[str, Any], np.ndarray]: if file_id not in load_cache: if loader.__name__ == "_load_demo_wave": load_cache[file_id] = loader(file_id, measurement_type, point_name, sample_time) else: load_cache[file_id] = loader(file_id) return load_cache[file_id] def _load_db_wave(self, file_id: int) -> tuple[dict[str, Any], np.ndarray]: with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( """ SELECT id, point_name, measurement_type, sample_frequency_hz, sample_count, sample_time, rpm, file_name, tspluse_status FROM wave_file WHERE id = %s """, (file_id,), ) metadata = cursor.fetchone() if metadata is None: raise ValueError(f"wave_file.id={file_id} 不存在") cursor.execute( """ SELECT sample_index, signal_value, second_value FROM wave_sample WHERE wave_file_id = %s ORDER BY sample_index ASC """, (file_id,), ) rows = cursor.fetchall() if not rows: raise ValueError(f"wave_file.id={file_id} 没有采样数据") samples = np.asarray( [ ( float(row["sample_index"]), float(row["signal_value"]), float(row["second_value"]) if row["second_value"] is not None else np.nan, ) for row in rows ], dtype=float, ) return metadata, samples @staticmethod @lru_cache(maxsize=24) def _demo_samples(file_id: int, measurement_type: str) -> np.ndarray: count = DEMO_SAMPLE_COUNT index = np.arange(count, dtype=float) revolution = DEMO_REVOLUTION_SAMPLES phase = (index % revolution) / revolution * 2 * np.pi second = np.zeros(count, dtype=float) for revolution_start in range(0, count, revolution): for pulse in range(PULSES_PER_REVOLUTION): pulse_start = revolution_start + int(round(pulse * revolution / PULSES_PER_REVOLUTION)) width = 22 if pulse == 0 else 8 pulse_end = min(count, pulse_start + width) second[pulse_start:pulse_end] = 40.0 variation = (file_id % 17) / 17.0 if measurement_type == "压力": signal = ( 4.2 + 1.8 * np.sin(phase - 0.4) + 0.55 * np.sin(2 * phase + variation) + 0.22 * np.sin(7 * phase) ) signal += 0.2 * np.maximum(np.sin(phase - 0.2), 0) ** 5 elif measurement_type == "位移": signal = 0.5 + 0.18 * np.cos(phase) + 0.035 * np.sin(3 * phase + variation) else: signal = 0.15 * np.sin(phase * 2 + variation) + 0.04 * np.sin(11 * phase) signal += 0.018 * np.cos(index / 37.0) return np.column_stack((index, signal, second)) def _load_demo_wave( self, file_id: int, measurement_type: str, point_name: str, sample_time: str, ) -> tuple[dict[str, Any], np.ndarray]: samples = self._demo_samples(file_id, measurement_type) metadata = { "id": file_id, "point_name": point_name, "measurement_type": measurement_type, "sample_frequency_hz": 25600, "sample_count": len(samples), "sample_time": sample_time, "rpm": 998.0, "file_name": f"demo-{file_id}.dat", } return metadata, samples @staticmethod def _demo_options() -> list[dict[str, Any]]: end = DEMO_START + timedelta(minutes=5 * (DEMO_POINT_COUNT - 1)) return [ { "devicePart": DEMO_DEVICE_PART, "devicePoint": device_point, "measurementType": DEVICE_POINT_TO_TYPE[device_point], "minTime": _time_string(DEMO_START), "maxTime": _time_string(end), "fileCount": DEMO_POINT_COUNT, } for device_point in DEVICE_POINTS ] @staticmethod def _demo_time_points( device_part: str, device_points: list[str], start: datetime | None, end: datetime | None, ) -> list[dict[str, Any]]: points = [] for index in range(DEMO_POINT_COUNT): timestamp = DEMO_START + timedelta(minutes=5 * index) if start and timestamp < start: continue if end and timestamp > end: continue files = {} for device_point in device_points: measurement_type = DEVICE_POINT_TO_TYPE[device_point] files[device_point] = { "id": DEMO_ID_BY_TYPE[measurement_type] + index, "devicePoint": device_point, "measurementType": measurement_type, "sampleCount": DEMO_SAMPLE_COUNT, "sampleFrequencyHz": 25600, "rpm": 998.0, } points.append( { "index": len(points), "sampleTime": _time_string(timestamp), "files": files, }, ) return points data_service = DataService()