from __future__ import annotations 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", } PHASES = ( ("排气", "#7b1fa2", 0.0, 120.0), ("压缩", "#c62828", 120.0, 195.0), ("膨胀", "#1565c0", 195.0, 270.0), ("进气", "#2e7d32", 270.0, 330.0), ) ANNOTATION_LABELS = ("正常", "异常") 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_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 _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_measurement_types(values: list[str] | tuple[str, ...] | None) -> list[str]: selected = list(values or MEASUREMENT_TYPES) invalid = [value for value in selected if value not in MEASUREMENT_TYPES] if invalid: raise ValueError(f"不支持的数据名称：{'、'.join(invalid)}") return [value for value in MEASUREMENT_TYPES if value in selected] 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(): placeholders = ", ".join(["%s"] * len(MEASUREMENT_TYPES)) with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( f""" SELECT point_name, measurement_type, MAX(sample_time) AS max_time, MIN(sample_time) AS min_time, COUNT(*) AS file_count FROM wave_file WHERE measurement_type IN ({placeholders}) GROUP BY point_name, measurement_type ORDER BY point_name, measurement_type """, MEASUREMENT_TYPES, ) rows = cursor.fetchall() return [ { "pointName": row["point_name"], "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) point_names = list(dict.fromkeys(row["pointName"] for row in rows)) return { "source": source, "measurementTypes": list(MEASUREMENT_TYPES), "pointNames": point_names, "options": rows, "notice": self._source_notice(source), } def time_points( self, point_name: str, measurement_types: list[str] | None, min_time: str | None, max_time: str | None, ) -> dict[str, Any]: if not point_name.strip(): raise ValueError("机组与部位不能为空") types = _validate_measurement_types(measurement_types) start = _parse_time(min_time) end = _parse_time(max_time) if start and end and start > end: raise ValueError("开始时间不能晚于结束时间") def database_query(): type_placeholders = ", ".join(["%s"] * len(types)) clauses = [ "point_name = %s", f"measurement_type IN ({type_placeholders})", ] params: list[Any] = [point_name, *types] 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, measurement_type, sample_time, sample_count, sample_frequency_hz, rpm FROM wave_file WHERE {' AND '.join(clauses)} ORDER BY sample_time ASC, id ASC """, params, ) rows = cursor.fetchall() return self._group_time_points(rows) def demo_query(): return self._demo_time_points(point_name, types, start, end) points, source = self._run_with_fallback(database_query, demo_query) return { "source": source, "pointName": point_name, "measurementTypes": types, "total": len(points), "points": points, "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: key = row["sample_time"] point = grouped.setdefault( key, { "sampleTime": _time_string(key), "files": {}, }, ) point["files"][row["measurement_type"]] = { "id": int(row["id"]), "sampleCount": int(row["sample_count"] or 0), "sampleFrequencyHz": int(row["sample_frequency_hz"] or 0), "rpm": float(row["rpm"] or 0), } return [ {"index": index, **point} for index, point in enumerate(grouped.values()) ] def wave_window( self, point_name: str, measurement_types: list[str], points: list[dict[str, Any]], max_points: int, no_sampling: bool = False, ) -> dict[str, Any]: types = _validate_measurement_types(measurement_types) if not points: raise ValueError("至少选择一个时间点") if len(points) > 200: raise ValueError("单次最多预览 200 个时间点，请缩小时间窗口") max_points = min(max(int(max_points), 256), 200000) def database_query(): return self._build_wave_window( point_name, types, points, max_points, self._load_db_wave, no_sampling, ) def demo_query(): return self._build_wave_window( point_name, types, points, max_points, self._load_demo_wave, no_sampling, ) result, source = self._run_with_fallback(database_query, demo_query) 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, point_name: str, types: 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]]] = {measurement_type: [] for measurement_type in types} 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 source_type = "压力" if "压力" in types else types[0] load_cache: dict[int, tuple[dict[str, Any], np.ndarray]] = {} for slot, point in enumerate(points): source_measurement_type = source_type source_file = point.get("files", {}).get(source_measurement_type) if not source_file: available = [ (measurement_type, file_info) for measurement_type, file_info in point.get("files", {}).items() if measurement_type in types and file_info ] if available: source_measurement_type, source_file = available[0] else: continue source_id = int(source_file["id"] if isinstance(source_file, dict) else source_file) try: source_metadata, source_samples = self._load_for_window( loader, source_id, source_measurement_type, point_name, point["sampleTime"], load_cache, ) except ValueError: # A selected file may legitimately have no samples. A database # connection error must escape and activate the demo fallback. continue detected, diagnostic = detect_cycles(source_samples) angle_vector = build_angle_vector(len(source_samples), detected) full_angle_vector = np.full(len(source_samples), np.nan, dtype=float) for detected_cycle in detected: full_angle_vector[detected_cycle.start_offset:detected_cycle.end_offset] = detected_cycle.angle current_volume, current_volume_info = self._build_volume_vector( len(source_samples), detected, point_name, ) if current_volume_info is not None: volume_info = current_volume_info sample_indices = source_samples[:, 0].astype(np.int64) sample_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) required = {0, len(source_samples) - 1} for cycle in detected: required.update( { cycle.start_offset, max(cycle.end_offset - 1, cycle.start_offset), *cycle.trigger_offsets, }, ) start_x = slot + cycle.start_offset / sample_span end_x = slot + cycle.end_offset / sample_span start_sample_index = int(sample_indices[cycle.start_offset]) end_sample_index = int(sample_indices[max(cycle.end_offset - 1, cycle.start_offset)]) 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": start_sample_index, "endSampleIndex": end_sample_index, "sourceType": source_measurement_type, }, ) for trigger_offset in sorted(required): if trigger_offset in required and any( trigger_offset == run_offset for cycle in detected for run_offset in cycle.trigger_offsets ): triggers.append(slot + trigger_offset / sample_span) diagnostics.append( { "waveFileId": source_id, "measurementType": source_measurement_type, "sampleTime": point["sampleTime"], **diagnostic, }, ) for measurement_type in types: file_info = point.get("files", {}).get(measurement_type) if not file_info: continue file_id = int(file_info["id"] if isinstance(file_info, dict) else file_info) if file_id == source_id: metadata, samples = source_metadata, source_samples else: try: metadata, samples = self._load_for_window( loader, file_id, measurement_type, point_name, 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} if file_id == source_id: file_required.update(required) if no_sampling: target_per_file = sample_count else: target_per_file = max(256, int(np.ceil(max_points / max(len(points), 1)))) chosen = downsample_indices(samples[:, 1], target_per_file, file_required) for offset in chosen: x = slot + (int(sample_indices_for_file[offset]) - file_base_index) / file_span second = _safe_float(samples[offset, 2]) raw_value = float(samples[offset, 1]) series_data[measurement_type].append( { "value": [x, raw_value], "x": x, "rawValue": raw_value, "sampleIndex": int(sample_indices_for_file[offset]), "waveFileId": file_id, "sampleTime": point["sampleTime"], "secondValue": second, "volume": ( _safe_float(current_volume[offset]) if file_id == source_id else None ), "angle360": ( _safe_float(full_angle_vector[offset]) if file_id == source_id else None ), }, ) if file_id == source_id: second_chosen = downsample_indices(samples[:, 2], target_per_file, file_required) for offset in second_chosen: second_value = _safe_float(samples[offset, 2]) if second_value is None: continue x = slot + (int(sample_indices_for_file[offset]) - file_base_index) / file_span second_series_data.append( { "value": [x, second_value], "x": x, "rawValue": second_value, "sampleIndex": int(sample_indices_for_file[offset]), "waveFileId": file_id, "sampleTime": point["sampleTime"], }, ) angle_chosen = downsample_indices(angle_vector, target_per_file, file_required) for offset in angle_chosen: value = _safe_float(angle_vector[offset]) if value is None: continue x = slot + (int(sample_indices_for_file[offset]) - file_base_index) / file_span angle_data.append( { "value": [x, value], "x": x, "angle": value, "sampleIndex": int(sample_indices_for_file[offset]), "waveFileId": file_id, "sampleTime": point["sampleTime"], }, ) volume_chosen = downsample_indices(current_volume, target_per_file, file_required) for offset in volume_chosen: value = _safe_float(current_volume[offset]) if value is None: continue x = slot + (int(sample_indices_for_file[offset]) - file_base_index) / file_span volume_data.append( { "value": [x, value], "x": x, "volume": value, "sampleIndex": int(sample_indices_for_file[offset]), "waveFileId": file_id, "sampleTime": point["sampleTime"], }, ) files.append( { "id": file_id, "pointIndex": slot, "sampleTime": point["sampleTime"], "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 point_name), "rpm": float(metadata.get("rpm") or 0), "fileName": str(metadata.get("file_name") or ""), }, ) # Keep at most the current time point's raw arrays resident. The # response retains only downsampled values and period metadata. load_cache.clear() for measurement_type in series_data: series_data[measurement_type].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"]) return { "pointName": point_name, "measurementTypes": types, "points": points, "xMin": 0, "xMax": len(points), "series": [ { "measurementType": measurement_type, "color": MEASUREMENT_COLORS[measurement_type], "data": series_data[measurement_type], } for measurement_type in types ], "secondSeries": { "name": "周期数据", "color": "#f56c6c", "sourceMeasurementType": source_type, "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, } @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 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 [ { "pointName": DEMO_POINT_NAME, "measurementType": measurement_type, "minTime": _time_string(DEMO_START), "maxTime": _time_string(end), "fileCount": DEMO_POINT_COUNT, } for measurement_type in MEASUREMENT_TYPES ] @staticmethod def _demo_time_points( point_name: str, types: 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 = { measurement_type: { "id": DEMO_ID_BY_TYPE[measurement_type] + index, "sampleCount": DEMO_SAMPLE_COUNT, "sampleFrequencyHz": 25600, "rpm": 998.0, } for measurement_type in types } points.append( { "index": len(points), "sampleTime": _time_string(timestamp), "files": files, }, ) return points data_service = DataService()