data_service.py 63 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528
  1. from __future__ import annotations
  2. from collections import OrderedDict
  3. from datetime import datetime, timedelta
  4. from functools import lru_cache
  5. from time import monotonic
  6. from typing import Any, Callable
  7. import numpy as np
  8. from ..algorithms.cycles import (
  9. DetectedCycle,
  10. PULSES_PER_REVOLUTION,
  11. build_angle_vector,
  12. detect_cycles,
  13. downsample_indices,
  14. )
  15. from ..config import settings
  16. from ..db import get_connection
  17. MEASUREMENT_TYPES = ("位移", "加速度", "压力")
  18. MEASUREMENT_COLORS = {
  19. "压力": "#e4572e",
  20. "位移": "#1f7a8c",
  21. "加速度": "#7b61a8",
  22. }
  23. PHASES = (
  24. ("排气", "#7b1fa2", 0.0, 120.0),
  25. ("压缩", "#c62828", 120.0, 195.0),
  26. ("膨胀", "#1565c0", 195.0, 270.0),
  27. ("进气", "#2e7d32", 270.0, 330.0),
  28. )
  29. ANNOTATION_LABELS = ("正常", "异常")
  30. # Extra samples fetched past cycle_end so detect_cycles can see the zero marker
  31. # that closes the first cycle (cycle_end is exclusive, the marker sits at it).
  32. FIRST_CYCLE_PAD = 64
  33. CYLINDER_BORE_MM = {
  34. "一缸": 360.0,
  35. "二缸": 490.0,
  36. "三缸": 390.0,
  37. "四缸": 490.0,
  38. "五缸": 390.0,
  39. "六缸": 490.0,
  40. }
  41. PISTON_STROKE_MM = 148.0
  42. CONNECTING_ROD_LENGTH_MM = 460.0
  43. CLEARANCE_VOLUME_L_BY_BORE = {
  44. 490.0: 1.51,
  45. 390.0: 0.74,
  46. 360.0: 0.62,
  47. }
  48. DEMO_POINT_NAME = "7号机组一缸压力盖侧"
  49. DEMO_START = datetime(2026, 4, 12, 8, 0, 0)
  50. DEMO_POINT_COUNT = 72
  51. DEMO_SAMPLE_COUNT = 32768
  52. DEMO_REVOLUTION_SAMPLES = 800
  53. DEMO_ID_BY_TYPE = {name: 100000 + index * 1000 for index, name in enumerate(MEASUREMENT_TYPES)}
  54. def _time_string(value: Any) -> str:
  55. if isinstance(value, datetime):
  56. return value.strftime("%Y-%m-%d %H:%M:%S")
  57. return str(value)
  58. def _parse_time(value: str | None) -> datetime | None:
  59. if not value:
  60. return None
  61. return datetime.fromisoformat(value.replace("Z", "+00:00").replace("T", " "))
  62. def _safe_float(value: Any) -> float | None:
  63. if value is None:
  64. return None
  65. number = float(value)
  66. return number if np.isfinite(number) else None
  67. def _annotation_dict(row: dict[str, Any]) -> dict[str, Any]:
  68. return {
  69. "id": int(row["id"]),
  70. "waveFileId": int(row["wave_file_id"]),
  71. "label": row["label"],
  72. "periodStart": int(row["period_start"]),
  73. "periodEnd": int(row["period_end"]),
  74. "sampleIndexStart": int(row["sample_index_start"]),
  75. "sampleIndexEnd": int(row["sample_index_end"]),
  76. }
  77. def _validate_measurement_types(values: list[str] | tuple[str, ...] | None) -> list[str]:
  78. selected = list(values or MEASUREMENT_TYPES)
  79. invalid = [value for value in selected if value not in MEASUREMENT_TYPES]
  80. if invalid:
  81. raise ValueError(f"不支持的数据名称:{'、'.join(invalid)}")
  82. return [value for value in MEASUREMENT_TYPES if value in selected]
  83. class DataService:
  84. # 数据库失败后的重试冷却时间(秒)。超过该时间后自动重连数据库,
  85. # 避免一次网络抖动就把服务永久锁死在演示数据模式。
  86. DB_RETRY_COOLDOWN = 30.0
  87. def __init__(self) -> None:
  88. self._db_failed = settings.demo_mode == "always"
  89. self._db_failed_at = monotonic() if self._db_failed else 0.0
  90. self._last_db_error = ""
  91. self._demo_annotations: dict[int, dict[str, Any]] = {}
  92. self._demo_annotation_seq = 1
  93. @property
  94. def source(self) -> str:
  95. return "demo" if self._db_failed else "database"
  96. @property
  97. def last_db_error(self) -> str:
  98. return self._last_db_error
  99. def _run_with_fallback(
  100. self,
  101. database_function: Callable[[], Any],
  102. demo_function: Callable[[], Any],
  103. ) -> tuple[Any, str]:
  104. if self._db_failed:
  105. if settings.demo_mode == "always":
  106. return demo_function(), "demo"
  107. if monotonic() - self._db_failed_at < self.DB_RETRY_COOLDOWN:
  108. return demo_function(), "demo"
  109. # 冷却结束,重新尝试数据库,数据库恢复后可自动切回真实数据。
  110. try:
  111. result = database_function()
  112. self._db_failed = False
  113. self._last_db_error = ""
  114. return result, "database"
  115. except Exception as error:
  116. if settings.demo_mode == "never":
  117. raise
  118. self._db_failed = True
  119. self._db_failed_at = monotonic()
  120. self._last_db_error = str(error)
  121. return demo_function(), "demo"
  122. def query_options(self) -> dict[str, Any]:
  123. def database_query():
  124. placeholders = ", ".join(["%s"] * len(MEASUREMENT_TYPES))
  125. with get_connection() as connection:
  126. with connection.cursor() as cursor:
  127. cursor.execute(
  128. f"""
  129. SELECT point_name, measurement_type,
  130. MAX(sample_time) AS max_time,
  131. MIN(sample_time) AS min_time,
  132. COUNT(*) AS file_count
  133. FROM wave_file
  134. WHERE measurement_type IN ({placeholders})
  135. GROUP BY point_name, measurement_type
  136. ORDER BY point_name, measurement_type
  137. """,
  138. MEASUREMENT_TYPES,
  139. )
  140. rows = cursor.fetchall()
  141. return [
  142. {
  143. "pointName": row["point_name"],
  144. "measurementType": row["measurement_type"],
  145. "minTime": _time_string(row["min_time"]),
  146. "maxTime": _time_string(row["max_time"]),
  147. "fileCount": int(row["file_count"]),
  148. }
  149. for row in rows
  150. ]
  151. rows, source = self._run_with_fallback(database_query, self._demo_options)
  152. point_names = list(dict.fromkeys(row["pointName"] for row in rows))
  153. return {
  154. "source": source,
  155. "measurementTypes": list(MEASUREMENT_TYPES),
  156. "pointNames": point_names,
  157. "options": rows,
  158. "notice": self._source_notice(source),
  159. }
  160. def abnormal_counts(self) -> dict[str, int]:
  161. """每个 point_name 的异常文件数量(tspluse_status > 0)。"""
  162. def database_query():
  163. with get_connection() as connection:
  164. with connection.cursor() as cursor:
  165. cursor.execute(
  166. """
  167. SELECT point_name, COUNT(*) AS cnt
  168. FROM wave_file
  169. WHERE tspluse_status > 0
  170. GROUP BY point_name
  171. """,
  172. )
  173. rows = cursor.fetchall()
  174. return {row["point_name"]: int(row["cnt"]) for row in rows}
  175. def demo_query():
  176. return {}
  177. result, _source = self._run_with_fallback(database_query, demo_query)
  178. return result
  179. def tspluse_ruler(self) -> dict[str, Any]:
  180. """压力部位 tspluse_status 的全局标尺,进入页面时只查询一次。"""
  181. def database_query():
  182. with get_connection() as connection:
  183. with connection.cursor() as cursor:
  184. cursor.execute(
  185. """
  186. SELECT MIN(tspluse_status) AS min_status,
  187. MAX(tspluse_status) AS max_status
  188. FROM wave_file
  189. WHERE rpm > 0 AND measurement_type = '压力'
  190. """,
  191. )
  192. row = cursor.fetchone()
  193. return {
  194. "min": int(row["min_status"]) if row and row["min_status"] is not None else 0,
  195. "max": int(row["max_status"]) if row and row["max_status"] is not None else 0,
  196. }
  197. def demo_query():
  198. return {"min": 0, "max": 0}
  199. result, source = self._run_with_fallback(database_query, demo_query)
  200. result["source"] = source
  201. return result
  202. def time_points(
  203. self,
  204. point_name: str,
  205. measurement_types: list[str] | None,
  206. min_time: str | None,
  207. max_time: str | None,
  208. include_stopped: bool = False,
  209. min_status: int | None = None,
  210. status_filter: list[str] | None = None,
  211. ) -> dict[str, Any]:
  212. if not point_name.strip():
  213. raise ValueError("机组与部位不能为空")
  214. types = _validate_measurement_types(measurement_types)
  215. status_filters = {value for value in (status_filter or [])}
  216. unknown = status_filters - {"abnormal", "no_cycle"}
  217. if unknown:
  218. raise ValueError(f"不支持的状态筛选:{'、'.join(sorted(unknown))}")
  219. start = _parse_time(min_time)
  220. end = _parse_time(max_time)
  221. if start and end and start > end:
  222. raise ValueError("开始时间不能晚于结束时间")
  223. def database_query():
  224. type_placeholders = ", ".join(["%s"] * len(types))
  225. clauses = [
  226. "point_name = %s",
  227. f"measurement_type IN ({type_placeholders})",
  228. ]
  229. params: list[Any] = [point_name, *types]
  230. if not include_stopped:
  231. clauses.append("rpm > 0")
  232. if min_status is not None and min_status > 0 and "no_cycle" not in status_filters:
  233. clauses.append("tspluse_status >= %s")
  234. params.append(min_status)
  235. if "abnormal" in status_filters and "no_cycle" in status_filters:
  236. clauses.append("(tspluse_status > 0 OR tspluse_status = -1)")
  237. elif "abnormal" in status_filters:
  238. clauses.append("tspluse_status > 0")
  239. elif "no_cycle" in status_filters:
  240. clauses.append("tspluse_status = -1")
  241. if start:
  242. clauses.append("sample_time >= %s")
  243. params.append(start)
  244. if end:
  245. clauses.append("sample_time <= %s")
  246. params.append(end)
  247. with get_connection() as connection:
  248. with connection.cursor() as cursor:
  249. cursor.execute(
  250. f"""
  251. SELECT id, point_name, measurement_type, sample_time,
  252. sample_count, sample_frequency_hz, rpm, tspluse_status
  253. FROM wave_file
  254. WHERE {' AND '.join(clauses)}
  255. ORDER BY sample_time ASC, id ASC
  256. """,
  257. params,
  258. )
  259. rows = cursor.fetchall()
  260. reference_rows: list[dict[str, Any]] = []
  261. if status_filters and rows:
  262. reference_where = [
  263. "point_name = %s",
  264. "measurement_type = '压力'",
  265. "tspluse_status = 0",
  266. ]
  267. reference_params: list[Any] = [point_name]
  268. if not include_stopped:
  269. reference_where.append("rpm > 0")
  270. if start:
  271. reference_where.append("sample_time >= %s")
  272. reference_params.append(start)
  273. if end:
  274. reference_where.append("sample_time <= %s")
  275. reference_params.append(end)
  276. reference_params.append(rows[0]["sample_time"])
  277. with get_connection() as connection:
  278. with connection.cursor() as cursor:
  279. cursor.execute(
  280. f"""
  281. SELECT id, point_name, measurement_type, sample_time,
  282. sample_count, sample_frequency_hz, rpm, tspluse_status
  283. FROM wave_file
  284. WHERE {' AND '.join(reference_where)}
  285. ORDER BY ABS(TIMESTAMPDIFF(SECOND, sample_time, %s)) ASC
  286. LIMIT 1
  287. """,
  288. reference_params,
  289. )
  290. reference_rows = cursor.fetchall()
  291. return self._group_time_points(rows), self._group_time_points(reference_rows)
  292. def demo_query():
  293. return self._demo_time_points(point_name, types, start, end), []
  294. result, source = self._run_with_fallback(database_query, demo_query)
  295. points, reference = result
  296. return {
  297. "source": source,
  298. "pointName": point_name,
  299. "measurementTypes": types,
  300. "total": len(points),
  301. "points": points,
  302. "referencePoints": reference,
  303. "notice": self._source_notice(source),
  304. }
  305. @staticmethod
  306. def _group_time_points(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
  307. grouped: OrderedDict[Any, dict[str, Any]] = OrderedDict()
  308. for row in rows:
  309. key = row["sample_time"]
  310. point = grouped.setdefault(
  311. key,
  312. {
  313. "sampleTime": _time_string(key),
  314. "files": {},
  315. },
  316. )
  317. point["files"][row["measurement_type"]] = {
  318. "id": int(row["id"]),
  319. "sampleCount": int(row["sample_count"] or 0),
  320. "sampleFrequencyHz": int(row["sample_frequency_hz"] or 0),
  321. "rpm": float(row["rpm"] or 0),
  322. "status": int(row.get("tspluse_status") or 0),
  323. }
  324. return [
  325. {"index": index, **point}
  326. for index, point in enumerate(grouped.values())
  327. ]
  328. def wave_window(
  329. self,
  330. point_name: str,
  331. measurement_types: list[str],
  332. points: list[dict[str, Any]],
  333. max_points: int,
  334. no_sampling: bool = False,
  335. first_cycle_only: bool = False,
  336. ) -> dict[str, Any]:
  337. types = _validate_measurement_types(measurement_types)
  338. if not points:
  339. raise ValueError("至少选择一个时间点")
  340. if len(points) > 200:
  341. raise ValueError("单次最多预览 200 个时间点,请缩小时间窗口")
  342. max_points = min(max(int(max_points), 256), 200000)
  343. def database_query():
  344. if first_cycle_only:
  345. return self._build_first_cycle_window(
  346. point_name,
  347. types,
  348. points,
  349. self._load_db_wave,
  350. )
  351. return self._build_wave_window(
  352. point_name,
  353. types,
  354. points,
  355. max_points,
  356. self._load_db_wave,
  357. no_sampling,
  358. )
  359. def demo_query():
  360. if first_cycle_only:
  361. return self._build_first_cycle_window(
  362. point_name,
  363. types,
  364. points,
  365. self._load_demo_wave,
  366. )
  367. return self._build_wave_window(
  368. point_name,
  369. types,
  370. points,
  371. max_points,
  372. self._load_demo_wave,
  373. no_sampling,
  374. )
  375. result, source = self._run_with_fallback(database_query, demo_query)
  376. result["source"] = source
  377. result["notice"] = self._source_notice(source)
  378. return result
  379. def period_detail(self, wave_file_id: int, period_number: int) -> dict[str, Any]:
  380. if wave_file_id <= 0 or period_number <= 0:
  381. raise ValueError("wave_file_id 和周期编号必须为正整数")
  382. def database_query():
  383. metadata, samples = self._load_db_wave(wave_file_id)
  384. return self._build_period_detail(metadata, samples, period_number)
  385. def demo_query():
  386. metadata, samples = self._load_demo_wave(
  387. wave_file_id,
  388. "压力",
  389. DEMO_POINT_NAME,
  390. DEMO_START,
  391. )
  392. return self._build_period_detail(metadata, samples, period_number)
  393. result, source = self._run_with_fallback(database_query, demo_query)
  394. result["source"] = source
  395. result["notice"] = self._source_notice(source)
  396. return result
  397. def annotation_config(self) -> dict[str, Any]:
  398. return {
  399. "source": self.source,
  400. "annotationWidth": settings.annotation_width,
  401. "notice": self._source_notice(self.source),
  402. }
  403. def list_annotations(self, wave_file_ids: list[int]) -> dict[str, Any]:
  404. ids = sorted({int(value) for value in wave_file_ids if value})
  405. if not ids:
  406. return {
  407. "source": self.source,
  408. "annotations": [],
  409. "notice": self._source_notice(self.source),
  410. }
  411. def database_query():
  412. placeholders = ", ".join(["%s"] * len(ids))
  413. with get_connection() as connection:
  414. with connection.cursor() as cursor:
  415. cursor.execute(
  416. f"""
  417. SELECT id, wave_file_id, label, period_start, period_end,
  418. sample_index_start, sample_index_end
  419. FROM wave_annotation
  420. WHERE wave_file_id IN ({placeholders})
  421. ORDER BY id ASC
  422. """,
  423. ids,
  424. )
  425. rows = cursor.fetchall()
  426. return [_annotation_dict(row) for row in rows]
  427. def demo_query():
  428. return [
  429. _annotation_dict(annotation)
  430. for annotation in self._demo_annotations.values()
  431. if annotation["wave_file_id"] in ids
  432. ]
  433. annotations, source = self._run_with_fallback(database_query, demo_query)
  434. return {
  435. "source": source,
  436. "annotations": annotations,
  437. "notice": self._source_notice(source),
  438. }
  439. def create_annotation(self, payload: dict[str, Any]) -> dict[str, Any]:
  440. self._validate_annotation(payload)
  441. wave_file_id = int(payload["wave_file_id"])
  442. label = payload["label"]
  443. period_start = int(payload["period_start"])
  444. period_end = int(payload["period_end"])
  445. sample_index_start = int(payload["sample_index_start"])
  446. sample_index_end = int(payload["sample_index_end"])
  447. def database_query():
  448. with get_connection() as connection:
  449. with connection.cursor() as cursor:
  450. cursor.execute(
  451. """
  452. INSERT INTO wave_annotation
  453. (wave_file_id, label, period_start, period_end,
  454. sample_index_start, sample_index_end)
  455. VALUES (%s, %s, %s, %s, %s, %s)
  456. """,
  457. (
  458. wave_file_id,
  459. label,
  460. period_start,
  461. period_end,
  462. sample_index_start,
  463. sample_index_end,
  464. ),
  465. )
  466. annotation_id = cursor.lastrowid
  467. cursor.execute(
  468. """
  469. SELECT id, wave_file_id, label, period_start, period_end,
  470. sample_index_start, sample_index_end
  471. FROM wave_annotation
  472. WHERE id = %s
  473. """,
  474. (annotation_id,),
  475. )
  476. return _annotation_dict(cursor.fetchone())
  477. def demo_query():
  478. annotation_id = self._demo_annotation_seq
  479. self._demo_annotation_seq += 1
  480. annotation = {
  481. "id": annotation_id,
  482. "wave_file_id": wave_file_id,
  483. "label": label,
  484. "period_start": period_start,
  485. "period_end": period_end,
  486. "sample_index_start": sample_index_start,
  487. "sample_index_end": sample_index_end,
  488. }
  489. self._demo_annotations[annotation_id] = annotation
  490. return _annotation_dict(annotation)
  491. result, source = self._run_with_fallback(database_query, demo_query)
  492. result["source"] = source
  493. result["notice"] = self._source_notice(source)
  494. return result
  495. def delete_annotation(self, annotation_id: int) -> dict[str, Any]:
  496. if annotation_id <= 0:
  497. raise ValueError("标注 id 必须为正整数")
  498. def database_query():
  499. with get_connection() as connection:
  500. with connection.cursor() as cursor:
  501. cursor.execute(
  502. "DELETE FROM wave_annotation WHERE id = %s",
  503. (annotation_id,),
  504. )
  505. return int(cursor.rowcount)
  506. def demo_query():
  507. if annotation_id not in self._demo_annotations:
  508. return 0
  509. del self._demo_annotations[annotation_id]
  510. return 1
  511. deleted, source = self._run_with_fallback(database_query, demo_query)
  512. if not deleted:
  513. raise ValueError(f"标注 id={annotation_id} 不存在")
  514. return {
  515. "deleted": annotation_id,
  516. "source": source,
  517. "notice": self._source_notice(source),
  518. }
  519. @staticmethod
  520. def _validate_annotation(payload: dict[str, Any]) -> None:
  521. label = payload.get("label")
  522. if label not in ANNOTATION_LABELS:
  523. raise ValueError("样本类型只能是 正常 或 异常")
  524. for field in ("wave_file_id", "period_start", "period_end", "sample_index_start", "sample_index_end"):
  525. if payload.get(field) is None:
  526. raise ValueError(f"{field} 不能为空")
  527. if int(payload["wave_file_id"]) <= 0:
  528. raise ValueError("wave_file_id 必须为正整数")
  529. if int(payload["period_start"]) <= 0 or int(payload["period_end"]) <= 0:
  530. raise ValueError("周期编号必须为正整数")
  531. if int(payload["period_start"]) > int(payload["period_end"]):
  532. raise ValueError("起始周期不能大于结束周期")
  533. if int(payload["sample_index_start"]) < 0 or int(payload["sample_index_end"]) < 0:
  534. raise ValueError("采样点索引不能为负")
  535. if int(payload["sample_index_start"]) > int(payload["sample_index_end"]):
  536. raise ValueError("起始采样点不能大于结束采样点")
  537. def health(self) -> dict[str, Any]:
  538. return {
  539. "status": "ok",
  540. "source": self.source,
  541. "databaseError": self._last_db_error or None,
  542. }
  543. def _source_notice(self, source: str) -> str | None:
  544. if source == "demo":
  545. if self._last_db_error:
  546. return f"当前为演示数据:数据库暂不可用({self._last_db_error})"
  547. return "当前为演示数据:可设置 DEMO_MODE=never 强制使用数据库"
  548. return "已连接 MySQL 数据库"
  549. def _build_wave_window(
  550. self,
  551. point_name: str,
  552. types: list[str],
  553. points: list[dict[str, Any]],
  554. max_points: int,
  555. loader: Callable[..., tuple[dict[str, Any], np.ndarray]],
  556. no_sampling: bool = False,
  557. ) -> dict[str, Any]:
  558. series_data: dict[str, list[dict[str, Any]]] = {measurement_type: [] for measurement_type in types}
  559. angle_data: list[dict[str, Any]] = []
  560. volume_data: list[dict[str, Any]] = []
  561. volume_info: dict[str, Any] | None = None
  562. cycles: list[dict[str, Any]] = []
  563. triggers: list[float] = []
  564. files: list[dict[str, Any]] = []
  565. diagnostics: list[dict[str, Any]] = []
  566. second_series_data: list[dict[str, Any]] = []
  567. second_finite_count = 0
  568. second_non_zero_count = 0
  569. second_min: float | None = None
  570. second_max: float | None = None
  571. source_type = "压力" if "压力" in types else types[0]
  572. load_cache: dict[int, tuple[dict[str, Any], np.ndarray]] = {}
  573. for slot, point in enumerate(points):
  574. source_measurement_type = source_type
  575. source_file = point.get("files", {}).get(source_measurement_type)
  576. if not source_file:
  577. available = [
  578. (measurement_type, file_info)
  579. for measurement_type, file_info in point.get("files", {}).items()
  580. if measurement_type in types and file_info
  581. ]
  582. if available:
  583. source_measurement_type, source_file = available[0]
  584. else:
  585. continue
  586. source_id = int(source_file["id"] if isinstance(source_file, dict) else source_file)
  587. try:
  588. source_metadata, source_samples = self._load_for_window(
  589. loader,
  590. source_id,
  591. source_measurement_type,
  592. point_name,
  593. point["sampleTime"],
  594. load_cache,
  595. )
  596. except ValueError:
  597. # A selected file may legitimately have no samples. A database
  598. # connection error must escape and activate the demo fallback.
  599. continue
  600. detected, diagnostic = detect_cycles(source_samples)
  601. angle_vector = build_angle_vector(len(source_samples), detected)
  602. full_angle_vector = np.full(len(source_samples), np.nan, dtype=float)
  603. for detected_cycle in detected:
  604. full_angle_vector[detected_cycle.start_offset:detected_cycle.end_offset] = detected_cycle.angle
  605. current_volume, current_volume_info = self._build_volume_vector(
  606. len(source_samples),
  607. detected,
  608. point_name,
  609. )
  610. if current_volume_info is not None:
  611. volume_info = current_volume_info
  612. sample_indices = source_samples[:, 0].astype(np.int64)
  613. sample_span = max(len(source_samples), 1)
  614. finite_second = source_samples[:, 2][np.isfinite(source_samples[:, 2])]
  615. if len(finite_second):
  616. second_finite_count += int(len(finite_second))
  617. second_non_zero_count += int(np.count_nonzero(finite_second != 0))
  618. current_min = float(np.min(finite_second))
  619. current_max = float(np.max(finite_second))
  620. second_min = current_min if second_min is None else min(second_min, current_min)
  621. second_max = current_max if second_max is None else max(second_max, current_max)
  622. required = {0, len(source_samples) - 1}
  623. for cycle in detected:
  624. required.update(
  625. {
  626. cycle.start_offset,
  627. max(cycle.end_offset - 1, cycle.start_offset),
  628. *cycle.trigger_offsets,
  629. },
  630. )
  631. start_x = slot + cycle.start_offset / sample_span
  632. end_x = slot + cycle.end_offset / sample_span
  633. start_sample_index = int(sample_indices[cycle.start_offset])
  634. end_sample_index = int(sample_indices[max(cycle.end_offset - 1, cycle.start_offset)])
  635. cycles.append(
  636. {
  637. "id": f"{source_id}:{cycle.number}",
  638. "waveFileId": source_id,
  639. "periodNo": cycle.number,
  640. "pointIndex": slot,
  641. "sampleTime": point["sampleTime"],
  642. "startX": start_x,
  643. "endX": end_x,
  644. "startSampleIndex": start_sample_index,
  645. "endSampleIndex": end_sample_index,
  646. "sourceType": source_measurement_type,
  647. },
  648. )
  649. for trigger_offset in sorted(required):
  650. if trigger_offset in required and any(
  651. trigger_offset == run_offset
  652. for cycle in detected
  653. for run_offset in cycle.trigger_offsets
  654. ):
  655. triggers.append(slot + trigger_offset / sample_span)
  656. diagnostics.append(
  657. {
  658. "waveFileId": source_id,
  659. "measurementType": source_measurement_type,
  660. "sampleTime": point["sampleTime"],
  661. **diagnostic,
  662. },
  663. )
  664. for measurement_type in types:
  665. file_info = point.get("files", {}).get(measurement_type)
  666. if not file_info:
  667. continue
  668. file_id = int(file_info["id"] if isinstance(file_info, dict) else file_info)
  669. if file_id == source_id:
  670. metadata, samples = source_metadata, source_samples
  671. else:
  672. try:
  673. metadata, samples = self._load_for_window(
  674. loader,
  675. file_id,
  676. measurement_type,
  677. point_name,
  678. point["sampleTime"],
  679. load_cache,
  680. )
  681. except ValueError:
  682. continue
  683. sample_count = len(samples)
  684. if not sample_count:
  685. continue
  686. sample_indices_for_file = samples[:, 0].astype(np.int64)
  687. file_base_index = int(sample_indices_for_file[0])
  688. file_span = max(sample_count, 1)
  689. file_required = {0, sample_count - 1}
  690. if file_id == source_id:
  691. file_required.update(required)
  692. if no_sampling:
  693. target_per_file = sample_count
  694. else:
  695. target_per_file = max(256, int(np.ceil(max_points / max(len(points), 1))))
  696. chosen = downsample_indices(samples[:, 1], target_per_file, file_required)
  697. for offset in chosen:
  698. x = slot + (int(sample_indices_for_file[offset]) - file_base_index) / file_span
  699. second = _safe_float(samples[offset, 2])
  700. raw_value = float(samples[offset, 1])
  701. series_data[measurement_type].append(
  702. {
  703. "value": [x, raw_value],
  704. "x": x,
  705. "rawValue": raw_value,
  706. "sampleIndex": int(sample_indices_for_file[offset]),
  707. "waveFileId": file_id,
  708. "sampleTime": point["sampleTime"],
  709. "secondValue": second,
  710. "volume": (
  711. _safe_float(current_volume[offset])
  712. if file_id == source_id
  713. else None
  714. ),
  715. "angle360": (
  716. _safe_float(full_angle_vector[offset])
  717. if file_id == source_id
  718. else None
  719. ),
  720. },
  721. )
  722. if file_id == source_id:
  723. second_chosen = downsample_indices(samples[:, 2], target_per_file, file_required)
  724. for offset in second_chosen:
  725. second_value = _safe_float(samples[offset, 2])
  726. if second_value is None:
  727. continue
  728. x = slot + (int(sample_indices_for_file[offset]) - file_base_index) / file_span
  729. second_series_data.append(
  730. {
  731. "value": [x, second_value],
  732. "x": x,
  733. "rawValue": second_value,
  734. "sampleIndex": int(sample_indices_for_file[offset]),
  735. "waveFileId": file_id,
  736. "sampleTime": point["sampleTime"],
  737. },
  738. )
  739. angle_chosen = downsample_indices(angle_vector, target_per_file, file_required)
  740. for offset in angle_chosen:
  741. value = _safe_float(angle_vector[offset])
  742. if value is None:
  743. continue
  744. x = slot + (int(sample_indices_for_file[offset]) - file_base_index) / file_span
  745. angle_data.append(
  746. {
  747. "value": [x, value],
  748. "x": x,
  749. "angle": value,
  750. "sampleIndex": int(sample_indices_for_file[offset]),
  751. "waveFileId": file_id,
  752. "sampleTime": point["sampleTime"],
  753. },
  754. )
  755. volume_chosen = downsample_indices(current_volume, target_per_file, file_required)
  756. for offset in volume_chosen:
  757. value = _safe_float(current_volume[offset])
  758. if value is None:
  759. continue
  760. x = slot + (int(sample_indices_for_file[offset]) - file_base_index) / file_span
  761. volume_data.append(
  762. {
  763. "value": [x, value],
  764. "x": x,
  765. "volume": value,
  766. "sampleIndex": int(sample_indices_for_file[offset]),
  767. "waveFileId": file_id,
  768. "sampleTime": point["sampleTime"],
  769. },
  770. )
  771. files.append(
  772. {
  773. "id": file_id,
  774. "pointIndex": slot,
  775. "sampleTime": point["sampleTime"],
  776. "measurementType": measurement_type,
  777. "sampleCount": int(metadata.get("sample_count") or sample_count),
  778. "sampleFrequencyHz": int(metadata.get("sample_frequency_hz") or 0),
  779. "pointName": str(metadata.get("point_name") or point_name),
  780. "rpm": float(metadata.get("rpm") or 0),
  781. "status": int(metadata.get("tspluse_status") or 0),
  782. "fileName": str(metadata.get("file_name") or ""),
  783. },
  784. )
  785. # Keep at most the current time point's raw arrays resident. The
  786. # response retains only downsampled values and period metadata.
  787. load_cache.clear()
  788. for measurement_type in series_data:
  789. series_data[measurement_type].sort(key=lambda item: item["x"])
  790. second_series_data.sort(key=lambda item: item["x"])
  791. angle_data.sort(key=lambda item: item["x"])
  792. volume_data.sort(key=lambda item: item["x"])
  793. return {
  794. "pointName": point_name,
  795. "measurementTypes": types,
  796. "points": points,
  797. "xMin": 0,
  798. "xMax": len(points),
  799. "series": [
  800. {
  801. "measurementType": measurement_type,
  802. "color": MEASUREMENT_COLORS[measurement_type],
  803. "data": series_data[measurement_type],
  804. }
  805. for measurement_type in types
  806. ],
  807. "secondSeries": {
  808. "name": "周期数据",
  809. "color": "#f56c6c",
  810. "sourceMeasurementType": source_type,
  811. "data": second_series_data,
  812. "finiteCount": second_finite_count,
  813. "nonZeroCount": second_non_zero_count,
  814. "min": second_min,
  815. "max": second_max,
  816. },
  817. "angleSeries": {
  818. "color": "#d59b2b",
  819. "data": angle_data,
  820. },
  821. "volumeSeries": {
  822. "color": "#4d9e6f",
  823. "data": volume_data,
  824. "info": volume_info,
  825. },
  826. "cycles": cycles,
  827. "triggerXs": sorted(set(triggers)),
  828. "files": files,
  829. "diagnostics": diagnostics,
  830. }
  831. def _build_first_cycle_window(
  832. self,
  833. point_name: str,
  834. types: list[str],
  835. points: list[dict[str, Any]],
  836. loader: Callable[..., tuple[dict[str, Any], np.ndarray]],
  837. ) -> dict[str, Any]:
  838. """Build a slot-layout window where each file shows only its first cycle.
  839. Uses the ``wave_file.cycle_start / cycle_end`` bounds written by
  840. detect_cycle_index.py and fetches every file's first-cycle slice in a
  841. single JOIN query (range scan on the ``(wave_file_id, sample_index)``
  842. primary key), so 50 files cost one query instead of 50 full-file loads.
  843. Files without a recorded cycle (无周期 / 未回写) contribute no curve but
  844. still appear in the file record list. The curve is continuous: all cycle
  845. samples are kept and each file spans its own x slot.
  846. """
  847. series_data: dict[str, list[dict[str, Any]]] = {measurement_type: [] for measurement_type in types}
  848. second_series_data: list[dict[str, Any]] = []
  849. angle_data: list[dict[str, Any]] = []
  850. volume_data: list[dict[str, Any]] = []
  851. volume_info: dict[str, Any] | None = None
  852. cycles: list[dict[str, Any]] = []
  853. triggers: list[float] = []
  854. files: list[dict[str, Any]] = []
  855. diagnostics: list[dict[str, Any]] = []
  856. second_finite_count = 0
  857. second_non_zero_count = 0
  858. second_min: float | None = None
  859. second_max: float | None = None
  860. source_type = "压力" if "压力" in types else types[0]
  861. slot_sources: list[tuple[int, dict[str, Any], str, int]] = []
  862. all_file_ids: set[int] = set()
  863. for slot, point in enumerate(points):
  864. source_measurement_type = source_type
  865. source_file = point.get("files", {}).get(source_measurement_type)
  866. if not source_file:
  867. available = [
  868. (measurement_type, file_info)
  869. for measurement_type, file_info in point.get("files", {}).items()
  870. if measurement_type in types and file_info
  871. ]
  872. if available:
  873. source_measurement_type, source_file = available[0]
  874. else:
  875. continue
  876. source_id = int(source_file["id"] if isinstance(source_file, dict) else source_file)
  877. slot_sources.append((slot, point, source_measurement_type, source_id))
  878. for measurement_type in types:
  879. file_info = point.get("files", {}).get(measurement_type)
  880. if file_info:
  881. all_file_ids.add(int(file_info["id"] if isinstance(file_info, dict) else file_info))
  882. if not slot_sources:
  883. return self._assemble_first_cycle_window(
  884. point_name, types, points, series_data, second_series_data,
  885. angle_data, volume_data, volume_info,
  886. second_finite_count, second_non_zero_count, second_min, second_max,
  887. cycles, triggers, files, diagnostics, 0,
  888. )
  889. is_demo = loader.__name__ == "_load_demo_wave"
  890. metas: dict[int, dict[str, Any]] = {}
  891. # file_id -> (cycle_start, cycle_end, padded_slice[offset, signal, second])
  892. slices: dict[int, tuple[int, int, np.ndarray]] = {}
  893. if is_demo:
  894. for _slot, point, source_measurement_type, source_id in slot_sources:
  895. metadata, samples = self._load_for_window(
  896. loader, source_id, source_measurement_type, point_name,
  897. point["sampleTime"], {},
  898. )
  899. metas[source_id] = metadata
  900. detected, _ = detect_cycles(samples)
  901. if detected:
  902. cycle = detected[0]
  903. start_si = int(samples[cycle.start_offset, 0])
  904. end_si = int(samples[cycle.end_offset, 0])
  905. pad_end = min(cycle.end_offset + FIRST_CYCLE_PAD, len(samples))
  906. slices[source_id] = (start_si, end_si, samples[cycle.start_offset:pad_end])
  907. else:
  908. source_ids = [source_id for _, _, _, source_id in slot_sources]
  909. with get_connection() as connection:
  910. with connection.cursor() as cursor:
  911. placeholders = ", ".join(["%s"] * len(all_file_ids))
  912. cursor.execute(
  913. f"""
  914. SELECT id, point_name, measurement_type, sample_time,
  915. sample_count, sample_frequency_hz, rpm, tspluse_status,
  916. file_name, cycle_start, cycle_end
  917. FROM wave_file
  918. WHERE id IN ({placeholders})
  919. """,
  920. tuple(all_file_ids),
  921. )
  922. for row in cursor.fetchall():
  923. metas[int(row["id"])] = row
  924. source_placeholders = ", ".join(["%s"] * len(source_ids))
  925. cursor.execute(
  926. f"""
  927. SELECT ws.wave_file_id,
  928. ws.sample_index,
  929. CAST(ws.signal_value AS FLOAT) AS sig,
  930. CAST(ws.second_value AS FLOAT) AS sec
  931. FROM wave_sample ws
  932. JOIN wave_file wf ON wf.id = ws.wave_file_id
  933. WHERE ws.wave_file_id IN ({source_placeholders})
  934. AND wf.cycle_start >= 0
  935. AND ws.sample_index >= wf.cycle_start
  936. AND ws.sample_index < wf.cycle_end + %s
  937. ORDER BY ws.wave_file_id, ws.sample_index
  938. """,
  939. (*tuple(source_ids), FIRST_CYCLE_PAD),
  940. )
  941. raw: dict[int, list[tuple[float, float, float]]] = {}
  942. for row in cursor.fetchall():
  943. raw.setdefault(int(row["wave_file_id"]), []).append(
  944. (
  945. float(row["sample_index"]),
  946. float(row["sig"]),
  947. float(row["sec"]) if row["sec"] is not None else float("nan"),
  948. ),
  949. )
  950. for file_id, rows in raw.items():
  951. meta = metas.get(file_id)
  952. if meta is None or meta.get("cycle_start") is None:
  953. continue
  954. slices[file_id] = (
  955. int(meta["cycle_start"]),
  956. int(meta["cycle_end"]),
  957. np.asarray(rows, dtype=float),
  958. )
  959. for slot, point, source_measurement_type, source_id in slot_sources:
  960. sample_time = point["sampleTime"]
  961. cycle_bounds = slices.get(source_id)
  962. cycle_start = 0
  963. cycle_end = 0
  964. slice_arr: np.ndarray | None = None
  965. if cycle_bounds is not None:
  966. cycle_start, cycle_end, slice_arr = cycle_bounds
  967. cycle_len = max(cycle_end - cycle_start, 0)
  968. has_cycle = slice_arr is not None and 1 < cycle_len <= len(slice_arr)
  969. if has_cycle:
  970. angle: np.ndarray | None = None
  971. volume: np.ndarray | None = None
  972. volume_info: dict[str, Any] | None = None
  973. detected, _ = detect_cycles(slice_arr)
  974. if detected:
  975. cycle = detected[0]
  976. angle = cycle.angle
  977. volume, volume_info = self._build_volume_vector(len(slice_arr), [cycle], point_name)
  978. for offset in range(cycle_len):
  979. sample_index = int(slice_arr[offset, 0])
  980. raw_value = float(slice_arr[offset, 1])
  981. second = _safe_float(slice_arr[offset, 2])
  982. angle360 = _safe_float(angle[offset]) if angle is not None and offset < len(angle) else None
  983. volume_value = _safe_float(volume[offset]) if volume is not None and offset < len(volume) else None
  984. x = slot + (sample_index - cycle_start) / cycle_len
  985. series_data[source_measurement_type].append(
  986. {
  987. "value": [x, raw_value],
  988. "x": x,
  989. "rawValue": raw_value,
  990. "sampleIndex": sample_index,
  991. "waveFileId": source_id,
  992. "sampleTime": sample_time,
  993. "secondValue": second,
  994. "volume": volume_value,
  995. "angle360": angle360,
  996. },
  997. )
  998. if volume_value is not None:
  999. volume_data.append(
  1000. {
  1001. "value": [x, volume_value],
  1002. "x": x,
  1003. "volume": volume_value,
  1004. "sampleIndex": sample_index,
  1005. "waveFileId": source_id,
  1006. "sampleTime": sample_time,
  1007. },
  1008. )
  1009. if angle360 is not None:
  1010. display_angle = angle360 if angle360 <= 180.0 else 360.0 - angle360
  1011. angle_data.append(
  1012. {
  1013. "value": [x, display_angle],
  1014. "x": x,
  1015. "angle": display_angle,
  1016. "sampleIndex": sample_index,
  1017. "waveFileId": source_id,
  1018. "sampleTime": sample_time,
  1019. },
  1020. )
  1021. if second is not None:
  1022. second_finite_count += 1
  1023. if second != 0:
  1024. second_non_zero_count += 1
  1025. second_min = second if second_min is None else min(second_min, second)
  1026. second_max = second if second_max is None else max(second_max, second)
  1027. second_series_data.append(
  1028. {
  1029. "value": [x, second],
  1030. "x": x,
  1031. "rawValue": second,
  1032. "sampleIndex": sample_index,
  1033. "waveFileId": source_id,
  1034. "sampleTime": sample_time,
  1035. },
  1036. )
  1037. if offset > 0:
  1038. prev = _safe_float(slice_arr[offset - 1, 2])
  1039. if second is not None and (prev is None or prev < 30) and second >= 30:
  1040. triggers.append(x)
  1041. cycles.append(
  1042. {
  1043. "id": f"{source_id}:1",
  1044. "waveFileId": source_id,
  1045. "periodNo": 1,
  1046. "pointIndex": slot,
  1047. "sampleTime": sample_time,
  1048. "startX": float(slot),
  1049. "endX": float(slot + 1),
  1050. "startSampleIndex": cycle_start,
  1051. "endSampleIndex": cycle_end - 1,
  1052. "sourceType": source_measurement_type,
  1053. },
  1054. )
  1055. diagnostics.append(
  1056. {
  1057. "waveFileId": source_id,
  1058. "measurementType": source_measurement_type,
  1059. "sampleTime": sample_time,
  1060. "cycleStart": cycle_start,
  1061. "cycleEnd": cycle_end,
  1062. "cycleSampleCount": cycle_len,
  1063. },
  1064. )
  1065. for measurement_type in types:
  1066. file_info = point.get("files", {}).get(measurement_type)
  1067. if not file_info:
  1068. continue
  1069. file_id = int(file_info["id"] if isinstance(file_info, dict) else file_info)
  1070. metadata = metas.get(file_id)
  1071. cycle_start = metadata.get("cycle_start") if metadata else None
  1072. cycle_end = metadata.get("cycle_end") if metadata else None
  1073. files.append(
  1074. {
  1075. "id": file_id,
  1076. "pointIndex": slot,
  1077. "sampleTime": sample_time,
  1078. "measurementType": measurement_type,
  1079. "sampleCount": int(
  1080. (metadata.get("sample_count") if metadata else file_info.get("sampleCount") or 0) or 0,
  1081. ),
  1082. "sampleFrequencyHz": int(
  1083. (metadata.get("sample_frequency_hz") if metadata else file_info.get("sampleFrequencyHz") or 0) or 0,
  1084. ),
  1085. "pointName": str((metadata.get("point_name") if metadata else point_name) or point_name),
  1086. "rpm": float((metadata.get("rpm") if metadata else file_info.get("rpm") or 0) or 0),
  1087. "status": int(
  1088. (metadata.get("tspluse_status") if metadata else file_info.get("status") or 0) or 0,
  1089. ),
  1090. "fileName": str((metadata.get("file_name") if metadata else "") or ""),
  1091. "cycleStart": int(cycle_start) if cycle_start is not None else None,
  1092. "cycleEnd": int(cycle_end) if cycle_end is not None else None,
  1093. },
  1094. )
  1095. return self._assemble_first_cycle_window(
  1096. point_name, types, points, series_data, second_series_data,
  1097. angle_data, volume_data, volume_info,
  1098. second_finite_count, second_non_zero_count, second_min, second_max,
  1099. cycles, triggers, files, diagnostics,
  1100. max(len(slot_sources) - len(cycles), 0),
  1101. )
  1102. @staticmethod
  1103. def _assemble_first_cycle_window(
  1104. point_name: str,
  1105. types: list[str],
  1106. points: list[dict[str, Any]],
  1107. series_data: dict[str, list[dict[str, Any]]],
  1108. second_series_data: list[dict[str, Any]],
  1109. angle_data: list[dict[str, Any]],
  1110. volume_data: list[dict[str, Any]],
  1111. volume_info: dict[str, Any] | None,
  1112. second_finite_count: int,
  1113. second_non_zero_count: int,
  1114. second_min: float | None,
  1115. second_max: float | None,
  1116. cycles: list[dict[str, Any]],
  1117. triggers: list[float],
  1118. files: list[dict[str, Any]],
  1119. diagnostics: list[dict[str, Any]],
  1120. missing_cycles: int = 0,
  1121. ) -> dict[str, Any]:
  1122. for measurement_type in series_data:
  1123. series_data[measurement_type].sort(key=lambda item: item["x"])
  1124. second_series_data.sort(key=lambda item: item["x"])
  1125. angle_data.sort(key=lambda item: item["x"])
  1126. volume_data.sort(key=lambda item: item["x"])
  1127. return {
  1128. "pointName": point_name,
  1129. "measurementTypes": types,
  1130. "points": points,
  1131. "xMin": 0,
  1132. "xMax": len(points),
  1133. "series": [
  1134. {
  1135. "measurementType": measurement_type,
  1136. "color": MEASUREMENT_COLORS[measurement_type],
  1137. "data": series_data[measurement_type],
  1138. }
  1139. for measurement_type in types
  1140. ],
  1141. "secondSeries": {
  1142. "name": "周期数据",
  1143. "color": "#f56c6c",
  1144. "sourceMeasurementType": "压力" if "压力" in types else types[0],
  1145. "data": second_series_data,
  1146. "finiteCount": second_finite_count,
  1147. "nonZeroCount": second_non_zero_count,
  1148. "min": second_min,
  1149. "max": second_max,
  1150. },
  1151. "angleSeries": {
  1152. "color": "#d59b2b",
  1153. "data": angle_data,
  1154. },
  1155. "volumeSeries": {
  1156. "color": "#4d9e6f",
  1157. "data": volume_data,
  1158. "info": volume_info,
  1159. },
  1160. "cycles": cycles,
  1161. "triggerXs": sorted(set(triggers)),
  1162. "files": files,
  1163. "diagnostics": diagnostics,
  1164. "firstCycleMode": True,
  1165. "firstCycleNotice": (
  1166. f"窗口内 {missing_cycles} 个文件尚未回写首个周期索引(cycle_start),"
  1167. "请先运行 detect_cycle_index.py 回写后再查看。"
  1168. if missing_cycles > 0
  1169. else None
  1170. ),
  1171. }
  1172. @staticmethod
  1173. def _build_volume_vector(
  1174. sample_count: int,
  1175. cycles: list[DetectedCycle],
  1176. point_name: str,
  1177. ) -> tuple[np.ndarray, dict[str, Any] | None]:
  1178. volume = np.full(sample_count, np.nan, dtype=float)
  1179. cylinder_name = next(
  1180. (name for name in CYLINDER_BORE_MM if name in point_name),
  1181. None,
  1182. )
  1183. if cylinder_name is None or not cycles:
  1184. return volume, None
  1185. bore_mm = CYLINDER_BORE_MM[cylinder_name]
  1186. clearance = CLEARANCE_VOLUME_L_BY_BORE[bore_mm]
  1187. crank_radius = PISTON_STROKE_MM / 2.0
  1188. piston_area = np.pi * (bore_mm / 2.0) ** 2
  1189. for cycle in cycles:
  1190. angle_rad = np.deg2rad(cycle.angle)
  1191. travel = (
  1192. crank_radius * (1.0 - np.cos(angle_rad))
  1193. + CONNECTING_ROD_LENGTH_MM
  1194. - np.sqrt(
  1195. CONNECTING_ROD_LENGTH_MM**2
  1196. - (crank_radius * np.sin(angle_rad)) ** 2,
  1197. )
  1198. )
  1199. volume[cycle.start_offset : cycle.end_offset] = (
  1200. clearance + piston_area * travel / 1_000_000.0
  1201. )
  1202. finite = volume[np.isfinite(volume)]
  1203. return volume, {
  1204. "cylinder": cylinder_name,
  1205. "boreMm": bore_mm,
  1206. "clearanceVolumeL": clearance,
  1207. "minVolumeL": float(np.min(finite)) if len(finite) else None,
  1208. "maxVolumeL": float(np.max(finite)) if len(finite) else None,
  1209. }
  1210. @staticmethod
  1211. def _build_period_detail(
  1212. metadata: dict[str, Any],
  1213. samples: np.ndarray,
  1214. period_number: int,
  1215. ) -> dict[str, Any]:
  1216. detected, diagnostics = detect_cycles(samples)
  1217. cycle = next((item for item in detected if item.number == period_number), None)
  1218. if cycle is None:
  1219. raise ValueError(f"没有找到周期 {period_number}")
  1220. angles360 = np.linspace(0.0, 359.0, 360)
  1221. pressure = np.interp(angles360, cycle.angle, cycle.signal)
  1222. display_angles = np.where(angles360 <= 180.0, angles360, 360.0 - angles360)
  1223. point_name = str(metadata.get("point_name") or "")
  1224. cylinder_name = next(
  1225. (name for name in CYLINDER_BORE_MM if name in point_name),
  1226. None,
  1227. )
  1228. volume: np.ndarray | None = None
  1229. volume_info: dict[str, Any] | None = None
  1230. if cylinder_name:
  1231. bore_mm = CYLINDER_BORE_MM[cylinder_name]
  1232. clearance_volume = CLEARANCE_VOLUME_L_BY_BORE[bore_mm]
  1233. angle_rad = np.deg2rad(angles360)
  1234. crank_radius = PISTON_STROKE_MM / 2.0
  1235. piston_travel = (
  1236. crank_radius * (1.0 - np.cos(angle_rad))
  1237. + CONNECTING_ROD_LENGTH_MM
  1238. - np.sqrt(
  1239. CONNECTING_ROD_LENGTH_MM**2
  1240. - (crank_radius * np.sin(angle_rad)) ** 2,
  1241. )
  1242. )
  1243. piston_area = np.pi * (bore_mm / 2.0) ** 2
  1244. volume = clearance_volume + piston_area * piston_travel / 1_000_000.0
  1245. volume_info = {
  1246. "cylinder": cylinder_name,
  1247. "boreMm": bore_mm,
  1248. "clearanceVolumeL": clearance_volume,
  1249. "minVolumeL": float(np.min(volume)),
  1250. "maxVolumeL": float(np.max(volume)),
  1251. }
  1252. start_index = int(samples[cycle.start_offset, 0])
  1253. end_offset = min(cycle.end_offset, len(samples) - 1)
  1254. end_index = int(samples[max(cycle.end_offset - 1, cycle.start_offset), 0])
  1255. return {
  1256. "waveFile": {
  1257. "id": int(metadata["id"]),
  1258. "pointName": point_name,
  1259. "measurementType": metadata.get("measurement_type"),
  1260. "sampleTime": _time_string(metadata.get("sample_time")),
  1261. "sampleFrequencyHz": int(metadata.get("sample_frequency_hz") or 0),
  1262. "sampleCount": int(metadata.get("sample_count") or len(samples)),
  1263. },
  1264. "period": {
  1265. "periodNo": cycle.number,
  1266. "startSampleIndex": start_index,
  1267. "endSampleIndex": end_index,
  1268. "sampleCount": int(cycle.end_offset - cycle.start_offset),
  1269. "triggerSampleIndices": [
  1270. int(samples[offset, 0])
  1271. for offset in cycle.trigger_offsets
  1272. if 0 <= offset < len(samples)
  1273. ],
  1274. },
  1275. "angles": display_angles.tolist(),
  1276. "angles360": angles360.tolist(),
  1277. "pressure": pressure.tolist(),
  1278. "volume": volume.tolist() if volume is not None else None,
  1279. "volumeInfo": volume_info,
  1280. "phases": [
  1281. {
  1282. "name": name,
  1283. "color": color,
  1284. "start": start,
  1285. "end": end,
  1286. }
  1287. for name, color, start, end in PHASES
  1288. ],
  1289. "diagnostics": diagnostics,
  1290. }
  1291. @staticmethod
  1292. def _load_for_window(
  1293. loader: Callable[..., tuple[dict[str, Any], np.ndarray]],
  1294. file_id: int,
  1295. measurement_type: str,
  1296. point_name: str,
  1297. sample_time: str,
  1298. load_cache: dict[int, tuple[dict[str, Any], np.ndarray]],
  1299. ) -> tuple[dict[str, Any], np.ndarray]:
  1300. if file_id not in load_cache:
  1301. if loader.__name__ == "_load_demo_wave":
  1302. load_cache[file_id] = loader(file_id, measurement_type, point_name, sample_time)
  1303. else:
  1304. load_cache[file_id] = loader(file_id)
  1305. return load_cache[file_id]
  1306. def _load_db_wave(self, file_id: int) -> tuple[dict[str, Any], np.ndarray]:
  1307. with get_connection() as connection:
  1308. with connection.cursor() as cursor:
  1309. cursor.execute(
  1310. """
  1311. SELECT id, point_name, measurement_type, sample_frequency_hz,
  1312. sample_count, sample_time, rpm, file_name, tspluse_status
  1313. FROM wave_file
  1314. WHERE id = %s
  1315. """,
  1316. (file_id,),
  1317. )
  1318. metadata = cursor.fetchone()
  1319. if metadata is None:
  1320. raise ValueError(f"wave_file.id={file_id} 不存在")
  1321. cursor.execute(
  1322. """
  1323. SELECT sample_index, signal_value, second_value
  1324. FROM wave_sample
  1325. WHERE wave_file_id = %s
  1326. ORDER BY sample_index ASC
  1327. """,
  1328. (file_id,),
  1329. )
  1330. rows = cursor.fetchall()
  1331. if not rows:
  1332. raise ValueError(f"wave_file.id={file_id} 没有采样数据")
  1333. samples = np.asarray(
  1334. [
  1335. (
  1336. float(row["sample_index"]),
  1337. float(row["signal_value"]),
  1338. float(row["second_value"]) if row["second_value"] is not None else np.nan,
  1339. )
  1340. for row in rows
  1341. ],
  1342. dtype=float,
  1343. )
  1344. return metadata, samples
  1345. @staticmethod
  1346. @lru_cache(maxsize=24)
  1347. def _demo_samples(file_id: int, measurement_type: str) -> np.ndarray:
  1348. count = DEMO_SAMPLE_COUNT
  1349. index = np.arange(count, dtype=float)
  1350. revolution = DEMO_REVOLUTION_SAMPLES
  1351. phase = (index % revolution) / revolution * 2 * np.pi
  1352. second = np.zeros(count, dtype=float)
  1353. for revolution_start in range(0, count, revolution):
  1354. for pulse in range(PULSES_PER_REVOLUTION):
  1355. pulse_start = revolution_start + int(round(pulse * revolution / PULSES_PER_REVOLUTION))
  1356. width = 22 if pulse == 0 else 8
  1357. pulse_end = min(count, pulse_start + width)
  1358. second[pulse_start:pulse_end] = 40.0
  1359. variation = (file_id % 17) / 17.0
  1360. if measurement_type == "压力":
  1361. signal = (
  1362. 4.2
  1363. + 1.8 * np.sin(phase - 0.4)
  1364. + 0.55 * np.sin(2 * phase + variation)
  1365. + 0.22 * np.sin(7 * phase)
  1366. )
  1367. signal += 0.2 * np.maximum(np.sin(phase - 0.2), 0) ** 5
  1368. elif measurement_type == "位移":
  1369. signal = 0.5 + 0.18 * np.cos(phase) + 0.035 * np.sin(3 * phase + variation)
  1370. else:
  1371. signal = 0.15 * np.sin(phase * 2 + variation) + 0.04 * np.sin(11 * phase)
  1372. signal += 0.018 * np.cos(index / 37.0)
  1373. return np.column_stack((index, signal, second))
  1374. def _load_demo_wave(
  1375. self,
  1376. file_id: int,
  1377. measurement_type: str,
  1378. point_name: str,
  1379. sample_time: str,
  1380. ) -> tuple[dict[str, Any], np.ndarray]:
  1381. samples = self._demo_samples(file_id, measurement_type)
  1382. metadata = {
  1383. "id": file_id,
  1384. "point_name": point_name,
  1385. "measurement_type": measurement_type,
  1386. "sample_frequency_hz": 25600,
  1387. "sample_count": len(samples),
  1388. "sample_time": sample_time,
  1389. "rpm": 998.0,
  1390. "file_name": f"demo-{file_id}.dat",
  1391. }
  1392. return metadata, samples
  1393. @staticmethod
  1394. def _demo_options() -> list[dict[str, Any]]:
  1395. end = DEMO_START + timedelta(minutes=5 * (DEMO_POINT_COUNT - 1))
  1396. return [
  1397. {
  1398. "pointName": DEMO_POINT_NAME,
  1399. "measurementType": measurement_type,
  1400. "minTime": _time_string(DEMO_START),
  1401. "maxTime": _time_string(end),
  1402. "fileCount": DEMO_POINT_COUNT,
  1403. }
  1404. for measurement_type in MEASUREMENT_TYPES
  1405. ]
  1406. @staticmethod
  1407. def _demo_time_points(
  1408. point_name: str,
  1409. types: list[str],
  1410. start: datetime | None,
  1411. end: datetime | None,
  1412. ) -> list[dict[str, Any]]:
  1413. points = []
  1414. for index in range(DEMO_POINT_COUNT):
  1415. timestamp = DEMO_START + timedelta(minutes=5 * index)
  1416. if start and timestamp < start:
  1417. continue
  1418. if end and timestamp > end:
  1419. continue
  1420. files = {
  1421. measurement_type: {
  1422. "id": DEMO_ID_BY_TYPE[measurement_type] + index,
  1423. "sampleCount": DEMO_SAMPLE_COUNT,
  1424. "sampleFrequencyHz": 25600,
  1425. "rpm": 998.0,
  1426. }
  1427. for measurement_type in types
  1428. }
  1429. points.append(
  1430. {
  1431. "index": len(points),
  1432. "sampleTime": _time_string(timestamp),
  1433. "files": files,
  1434. },
  1435. )
  1436. return points
  1437. data_service = DataService()