周期功图：7/8/9号机按文档相位角旋转并修正轴侧杆径容积

9号机_3缸_角度问题初步分析.md

@@ -0,0 +1,137 @@
+# 9 号机 3 缸 2026-04-15 23:15 功图角度问题初步分析
+
+## 1. 原始文件清单
+
+已按 `wave_file.file_name` + `compressor_data_import.directory_path` 生成：
+
+- `9号机_20260415_2315_原始文件清单.csv`（含 1/2/3/4/5/6 缸盖侧、轴侧共 12 个文件）
+
+用户要求重点关注的 1、3、5 缸：
+
+| 缸 | 测点 | wave_file_id | 键相 |
+|---|---|---|---|
+| 一缸 | 压力盖侧 | 222589 | 键相8-0-0 |
+| 一缸 | 压力轴侧 | 224154 | 键相8-0-0 |
+| 三缸 | 压力盖侧 | 239529 | 键相8-0-135 |
+| 三缸 | 压力轴侧 | 242439 | 键相8-0-135 |
+| 五缸 | 压力盖侧 | 257776 | 键相8-0-270 |
+| 五缸 | 压力轴侧 | 260757 | 键相8-0-270 |
+
+完整路径见 CSV 的“完整路径”列。
+
+## 2. 当前算法口径
+
+`backend/app/services/data_service.py` 与 `viewer/backend/services/indicator_service.py`：
+
+1. 用 `detect_cycles` 找 `second_value >= 30` 的长脉冲作为 0° 标记；
+2. `wave_file.cycle_start` 取该长脉冲的**上升沿**，`cycle_end` 取下一个 0° 脉冲上升沿；
+3. `wave_sample_one` 保存 `[cycle_start, cycle_end)` 的一个完整周期；
+4. 对周期内样本做 `linspace(0, 360)`，**直接把周期起点当作 0°**；
+5. 整个过程**没有读取 `wave_file.keyphase_info`**。
+
+`statistic_pressure_angle` 也是用同一口径生成（angle=0 的压力与 `wave_sample_one` 首值完全一致），所以数据库内的角度表与前端功图是同一套相位。
+
+## 3. 本次核对结果
+
+同一时刻 1/3/5 缸的周期边界完全一致：
+
+- `cycle_start = 1412`，`cycle_end = 2958`，周期长度 1546 点；
+- 995.4 rpm、25600 Hz 下一转理论长度 `25600*60/995.4 ≈ 1543` 点，1546 点合理；
+- 周期内正好 8 个键相脉冲，第一个为长脉冲（0° 标记）。
+
+也就是说：**不是每个缸取到了不同的圈**，周期长度和键相分割本身是对的。
+
+当前“存储 0°”处（即周期起点）的压力：
+
+| 缸 | 键相 | 盖侧 0° | 轴侧 0° |
+|---|---|---|---|
+| 一缸 | 0 | 4.876 | 3.421 |
+| 二缸 | 0 | 1.906 | 0.683 |
+| 三缸 | 135 | 2.075 | 2.374 |
+| 四缸 | 135 | 0.810 | 1.245 |
+| 五缸 | 270 | 1.686 | 3.882 |
+| 六缸 | 265 | 0.642 | 1.917 |
+
+用户给出的 3 缸实际 0° 约为盖侧 1.9、轴侧 3.3。当前 3 缸存储 0° 是 2.075 / 2.374，轴侧明显偏低，方向对不上。
+
+在三缸存储周期内搜索 `(盖≈1.9, 轴≈3.3)` 的最佳位置：
+
+- 约在存储角 **337°** 处：盖 1.851、轴 3.299，与用户给出的数值最接近。
+
+如果按 `键相8-0-135` 做相位平移，令物理 0° 对应存储角 225°（=360-135）：
+
+- 盖 1.696、轴 3.741，方向正确（盖下降、轴上升），但数值仍有偏差。
+
+五缸按 `键相8-0-270` 平移到存储角 90°：
+
+- 盖 1.865、轴 3.492，同样落在“盖约 1.9、轴约 3.5”附近。
+
+## 4. 初步判断
+
+更可能是**角度参考/相位问题**，而不是采样数据丢失或取错圈：
+
+1. 1、2 缸键相为 `0`，当前算法等于不做平移，所以用户看到“1 缸完全符合、2 缸大方向符合”；
+2. 3、4 缸键相为 `135`，5、6 缸为 `270/265`，而代码完全忽略该字段，所以这几缸的功图相对真实 0° 被整体旋转；
+3. “2/4/6 同级缸”不能解释现象：2 缸对是因为键相 0，4 缸不对是因为键相 135，与缸径级别无关；
+4. “首周期取错”目前证据不足：三个缸的 `cycle_start/cycle_end` 完全相同、周期长度正确、8 个键相脉冲完整。更像是**首周期的 0° 定义（键相偏置）没对齐**，而不是取到了另一圈。
+
+## 5. 需要原始文件确认的点
+
+请把 CSV 中 1/3/5 缸的原始 CSV 拷到本地后，重点核对：
+
+1. 原始文件表头/列名：是否已有“角度”“曲轴转角”“键相”列；
+2. 原始文件前几行（sample_index 0 附近）盖侧/轴侧压力是否就是用户所说的 1.9 / 3.3；
+3. 原始文件里 0° 到底对应长键相脉冲的上升沿、下降沿还是中心；
+4. `键相8-0-135` 中 135 的含义，是否就是该测点相对气缸 0° 的安装偏置；
+5. 原始软件的 P-θ 图在 0° 处取值与 `statistic_pressure_angle` 的差异。
+
+拿到原始文件后，可以直接对比：
+
+- 原始 CSV 的时间列/采样序号 → 确认 `cycle_start=1412` 是否就是原始软件认定的 0°；
+- 原始 CSV 的 0° 行压力 → 与 `wave_sample_one` 首值对比，判断是“数据取错”还是“相位差”。
+
+## 6. 附带发现
+
+- `6 缸盖侧` 的 `wave_file_id=1162517` 来自 `test_20260324111644_20260622111644` 批次，而其余文件来自 `exp_all_device_2026.3.16-2026.4.18...`。同一时刻可能存在重复导入，后续分析时建议先确认是否应排除测试批次。
+- 键相偏置是**按测点**配置的，不是按缸固定：例如 `backend/预训练之文件头.csv` 中 7 号机三缸盖侧为 `键相8-0-270`、轴侧为 `键相8-0-0`。修复时必须逐测点读取 `keyphase_info`。
+
+## 7. 原始文件验证（决定性）
+
+原始 CSV 放在 `cache/9号机组一三五缸压力_20260415231501/`。文件无表头，两列：`压力, 键相`。
+
+逐点比对 `wave_sample_one` 与原始文件 `raw[1412:2958]`：
+
+| 文件 | wave_file_id | 样本区间 | max\|raw-db\| |
+|---|---|---|---|
+| 一缸盖 | 222589 | 1412:2958 | 0.000000 |
+| 一缸轴 | 224154 | 1412:2958 | 0.000000 |
+| 三缸盖 | 239529 | 1412:2958 | 0.000000 |
+| 三缸轴 | 242439 | 1412:2958 | 0.000000 |
+| 五缸盖 | 257776 | 1412:2958 | 0.000000 |
+| 五缸轴 | 260757 | 1412:2958 | 0.000000 |
+
+结论：**数据库里的首周期与原始文件逐点完全一致，取数没有错。**
+
+6 个文件的键相长脉冲起点全部是 `1412`，脉冲序列也完全相同（79, 276, 468, ...），说明它们用的是同一个键相信号、同一个物理圈，`cycle_start/cycle_end` 正确。
+
+原始文件中，3 缸 `(盖≈1.9, 轴≈3.3)` 出现在共享键相坐标的 **≈337°**（raw 第 2859 点，1.856 / 3.292）；按键相 135° 旋转则出现在 225°（1.696 / 3.741）。两者都说明：**真实 0° 不在共享长脉冲起点，而是存在固定相位偏置。**
+
+## 8. 最终结论
+
+**是角度相位/键相偏置问题，不是数据问题，也不是首周期取错圈。**
+
+- 数据：原始文件与库内逐点一致（误差 0）。
+- 圈：6 个文件同一键相脉冲、同一圈、周期长度 1546 点正确。
+- 形态：1/3/5 缸归一化后是同一形状，互相关 0.95~0.99，正向优于反向（不是采样方向问题）。
+- 相位：相对旋转量长期稳定，且按键相分组：
+  - 1/2 缸 `键相0`：基本同相（2 缸约 -20°，所以“大方向符合”）；
+  - 3/4 缸 `键相135`：约 220°（≈360-135）；
+  - 5/6 缸 `键相270/265`：约 100°（≈360-270）。
+- 代码在 `_stored_first_cycle`、`indicator_service.py`、`statistic_pressure_angle` 全链路都直接 `linspace(0,360)`，**没有使用 `keyphase_info`**。
+
+## 9. 修复建议
+
+1. 逐测点解析 `keyphase_info` 末位角度 `K`，把共享键相角换算成该测点自身角：`display_angle = (shared_angle - K) mod 360`（方向需用 1 缸/3 缸实测标定一次）。
+2. 用 8 个键相脉冲做分段角度，替代均匀 `linspace`，消除周期内小角度畸变。
+3. 用 1 缸做基线回归：修正后 1 缸应与现在一致；3/4/5/6 缸 0° 应落到用户给出的 `≈1.9/3.3` 附近。
+4. 修复只影响显示/角度分箱，不改原始采样与阶段规则。

backend/app/services/data_service.py

@@ -14,7 +14,6 @@ import numpy as np
 from ..algorithms.cycles import (
     DetectedCycle,
     PULSES_PER_REVOLUTION,
-    build_angle_vector,
     detect_cycles,
     downsample_indices,
 )
@@ -91,11 +90,23 @@ CYLINDER_BORE_MM = {
 }
 PISTON_STROKE_MM = 148.0
 CONNECTING_ROD_LENGTH_MM = 460.0
+PISTON_ROD_DIAMETER_MM = 78.0
 CLEARANCE_VOLUME_L_BY_BORE = {
     490.0: 1.51,
     390.0: 0.74,
     360.0: 0.62,
 }
+# 7/8/9 号机各缸相位角（月度报告/沁水压缩机参数20250102.docx）。
+# 4/5/6 号机暂无资料，保持原逻辑（相位 0，不做旋转）。
+CYLINDER_PHASE_DEGREE = {
+    "一缸": 0.0,
+    "二缸": 0.0,
+    "三缸": 120.0,
+    "四缸": 240.0,
+    "五缸": 240.0,
+    "六缸": 120.0,
+}
+PHASE_ROTATION_UNITS = frozenset({"7", "8", "9"})
 
 DEMO_POINT_NAME = "7号机组一缸压力盖侧"
 DEMO_DEVICE_PART = "7号机组一缸"
@@ -125,13 +136,58 @@ def _safe_float(value: Any) -> float | None:
     return number if np.isfinite(number) else None
 
 
+def _point_cylinder(point_name: str) -> str | None:
+    text = point_name or ""
+    return next((name for name in CYLINDER_BORE_MM if name in text), None)
+
+
+def _phase_degrees(point_name: str) -> float:
+    """Per-cylinder phase angle for 7/8/9; 0 for units without reference data."""
+    unit = _unit_number(point_name or "")
+    cylinder = _point_cylinder(point_name or "")
+    if unit in PHASE_ROTATION_UNITS and cylinder is not None:
+        return CYLINDER_PHASE_DEGREE[cylinder]
+    return 0.0
+
+
+def _rotate_angle(angle: np.ndarray, point_name: str) -> np.ndarray:
+    """Convert shared-keyphase angle to the cylinder's own 0-360 angle."""
+    values = np.asarray(angle, dtype=float)
+    phase = _phase_degrees(point_name)
+    if not phase:
+        return values
+    return np.mod(values - phase, 360.0)
+
+
+def _piecewise_angle(count: int, trigger_offsets: tuple[int, ...]) -> np.ndarray:
+    """Crank angle from the 8 keyphase pulses; falls back to uniform sampling."""
+    offsets = sorted({int(offset) for offset in trigger_offsets if 0 <= int(offset) < count})
+    if len(offsets) < PULSES_PER_REVOLUTION or offsets[0] != 0:
+        return np.linspace(0.0, 360.0, count, endpoint=False)
+    offsets = offsets[:PULSES_PER_REVOLUTION]
+    bounds = offsets + [count]
+    angle = np.full(count, np.nan, dtype=float)
+    for pulse in range(PULSES_PER_REVOLUTION):
+        start, end = bounds[pulse], min(bounds[pulse + 1], count)
+        if end <= start:
+            continue
+        angle[start:end] = np.linspace(
+            pulse * 45.0, (pulse + 1) * 45.0, end - start, endpoint=False,
+        )
+    if np.isnan(angle).any():
+        good = np.flatnonzero(~np.isnan(angle))
+        if not len(good):
+            return np.linspace(0.0, 360.0, count, endpoint=False)
+        angle = np.interp(np.arange(count), good, angle[good])
+    return angle
+
+
 def _stored_first_cycle(samples: np.ndarray) -> DetectedCycle | None:
     """Build a cycle from wave_sample_one, which contains one cycle only."""
     if samples.ndim != 2 or samples.shape[1] < 3 or len(samples) < 2:
         return None
     start = 0
     end = len(samples)
-    angle = np.linspace(0.0, 360.0, end - start, endpoint=False)
     trigger_offsets = tuple(
         int(offset)
         for offset in np.flatnonzero(
@@ -139,6 +195,7 @@ def _stored_first_cycle(samples: np.ndarray) -> DetectedCycle | None:
         )
         if offset == 0 or samples[offset - 1, 2] < 30.0
     )
+    angle = _piecewise_angle(end - start, trigger_offsets)
     return DetectedCycle(
         number=1,
         start_offset=start,
@@ -1602,6 +1659,9 @@ class DataService:
         if not angles:
             raise ValueError("请至少输入一个角度")
         angles = [int(angle) for angle in angles if 0 <= int(angle) <= 360]
+        # statistic_pressure_angle 存的是共享键相角；7/8/9 查询时换算到该缸自身角。
+        phase = int(round(_phase_degrees(device_part))) % 360
+        stored_angles = sorted({(angle + phase) % 360 for angle in angles})
 
         def database_query():
             with get_connection() as connection:
@@ -1620,9 +1680,9 @@ class DataService:
                     if end:
                         clauses.append("wf.sample_time <= %s")
                         params.append(end)
-                    angle_placeholders = ", ".join(["%s"] * len(angles))
+                    angle_placeholders = ", ".join(["%s"] * len(stored_angles))
                     clauses.append(f"spa.angle IN ({angle_placeholders})")
-                    params.extend(angles)
+                    params.extend(stored_angles)
                     cursor.execute(
                         f"""
                         SELECT wf.sample_time, spa.angle, spa.pressure
@@ -1638,22 +1698,26 @@ class DataService:
                     stop_days = self._query_stop_days(
                         cursor, device_part, device_points, start, end,
                     )
-            series: dict[int, list[dict[str, Any]]] = {}
+            by_stored: dict[int, list[dict[str, Any]]] = {}
             for row in rows:
-                angle = int(round(float(row["angle"])))
-                series.setdefault(angle, []).append({
+                stored_angle = int(round(float(row["angle"])))
+                by_stored.setdefault(stored_angle, []).append({
                     "time": _time_string(row["sample_time"]),
                     "pressure": float(row["pressure"]),
                 })
+            series: dict[int, list[dict[str, Any]]] = {
+                angle: list(by_stored.get((angle + phase) % 360, []))
+                for angle in angles
+            }
             for day in sorted(stop_days):
                 day_time = f"{day} 00:00:00"
                 for angle in angles:
-                    series.setdefault(angle, []).append({
+                    series[angle].append({
                         "time": day_time,
                         "pressure": 0.0,
                     })
             for angle in angles:
-                series.setdefault(angle, []).sort(key=lambda point: point["time"])
+                series[angle].sort(key=lambda point: point["time"])
             return {
                 "angles": angles,
                 "series": [
@@ -2346,14 +2410,22 @@ class DataService:
                     source_samples = None
                 if source_samples is not None and len(source_samples):
                     source_detected, source_diagnostic = _detect_source_cycles(source_samples, single_cycle)
-                    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_angle_full = _rotate_angle(
+                        source_angle_full, device_part + source_point,
+                    )
+                    source_angle_vector = np.where(
+                        source_angle_full <= 180.0,
+                        source_angle_full,
+                        360.0 - source_angle_full,
+                    )
                     source_volume, current_volume_info = self._build_volume_vector(
                         len(source_samples),
                         source_detected,
-                        device_part + primary,
+                        device_part + source_point,
+                        side=source_point,
                     )
                     if current_volume_info is not None:
                         volume_info = current_volume_info
@@ -2523,14 +2595,15 @@ class DataService:
                                 },
                             )
                         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
+                            full_own = _rotate_angle(full_own, device_part + device_point)
                             own_volume, _ = self._build_volume_vector(
                                 len(samples),
                                 detected_own,
                                 device_part + device_point,
+                                side=device_point,
                             )
                             own_angle = full_own
                 if no_sampling:
@@ -2771,11 +2844,13 @@ class DataService:
                 detected = [stored_cycle] if stored_cycle is not None else []
                 if detected:
                     cycle = detected[0]
-                    angle = cycle.angle
+                    point_name = device_part + device_point
+                    angle = _rotate_angle(cycle.angle, point_name)
                     volume, point_volume_info = self._build_volume_vector(
                         len(slice_arr),
                         [cycle],
-                        device_part + primary,
+                        point_name,
+                        side=device_point,
                     )
                 if is_slot_source and point_volume_info is not None:
                     volume_info = point_volume_info
@@ -2998,20 +3073,27 @@ class DataService:
         sample_count: int,
         cycles: list[DetectedCycle],
         point_name: str,
+        side: str | None = None,
     ) -> 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,
-        )
+        cylinder_name = _point_cylinder(point_name)
         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
+        phase = _phase_degrees(point_name)
+        rod_mm = (
+            PISTON_ROD_DIAMETER_MM
+            if side == "压力轴侧" and _unit_number(point_name) in PHASE_ROTATION_UNITS
+            else 0.0
+        )
+        piston_area = np.pi * ((bore_mm / 2.0) ** 2 - (rod_mm / 2.0) ** 2)
         for cycle in cycles:
-            angle_rad = np.deg2rad(cycle.angle)
+            angle = cycle.angle
+            if phase:
+                angle = np.mod(angle - phase, 360.0)
+            angle_rad = np.deg2rad(angle)
             travel = (
                 crank_radius * (1.0 - np.cos(angle_rad))
                 + CONNECTING_ROD_LENGTH_MM
@@ -3027,6 +3109,8 @@ class DataService:
         return volume, {
             "cylinder": cylinder_name,
             "boreMm": bore_mm,
+            "rodDiameterMm": rod_mm,
+            "phaseDegree": phase,
             "clearanceVolumeL": clearance,
             "minVolumeL": float(np.min(finite)) if len(finite) else None,
             "maxVolumeL": float(np.max(finite)) if len(finite) else None,
@@ -3048,12 +3132,13 @@ class DataService:
         if cycle is None:
             raise ValueError(f"没有找到周期 {period_number}")
 
+        point_name = str(metadata.get("point_name") or "")
+        side = metadata.get("device_point")
         # 保留 wave_sample_one 的完整周期原始采样点，不再插值成固定 360 点。
-        angles360 = np.asarray(cycle.angle, dtype=float)
+        # 7/8/9 号机按文档相位角旋转到该缸自身 0°；4/5/6 保持原样。
+        angles360 = _rotate_angle(np.asarray(cycle.angle, dtype=float), point_name)
         pressure = np.asarray(cycle.signal, dtype=float)
         display_angles = np.where(angles360 <= 180.0, angles360, 360.0 - angles360)
-        point_name = str(metadata.get("point_name") or "")
-        side = metadata.get("device_point")
         if side not in PRESSURE_SIDES:
             side = next((item for item in PRESSURE_SIDES if point_name.endswith(item)), None)
         device_part = metadata.get("device_part")
@@ -3078,11 +3163,19 @@ class DataService:
                     - (crank_radius * np.sin(angle_rad)) ** 2,
                 )
             )
-            piston_area = np.pi * (bore_mm / 2.0) ** 2
+            # 轴侧为环形面积，需扣除活塞杆；7/8/9 按文档杆径 78mm。
+            rod_mm = (
+                PISTON_ROD_DIAMETER_MM
+                if side == "压力轴侧" and _unit_number(point_name) in PHASE_ROTATION_UNITS
+                else 0.0
+            )
+            piston_area = np.pi * ((bore_mm / 2.0) ** 2 - (rod_mm / 2.0) ** 2)
             volume = clearance_volume + piston_area * piston_travel / 1_000_000.0
             volume_info = {
                 "cylinder": cylinder_name,
                 "boreMm": bore_mm,
+                "rodDiameterMm": rod_mm,
+                "phaseDegree": _phase_degrees(point_name),
                 "clearanceVolumeL": clearance_volume,
                 "minVolumeL": float(np.min(volume)),
                 "maxVolumeL": float(np.max(volume)),

frontend/src/components/PeriodModal.vue

@@ -107,6 +107,9 @@ function sideSeries(mode: 'pv' | 'angle') {
       if (phaseIndex < 0) continue
       buckets[phaseIndex].push([x, y, Number.isFinite(angle360) ? angle360 : Number.NaN])
     }
+    for (const bucket of buckets) {
+      bucket.sort((left, right) => (left[2] ?? 0) - (right[2] ?? 0))
+    }
     const color = SIDE_COLORS[side.side] ?? '#54717f'
     return side.phases.flatMap((_phase, phaseIndex) => {
       if (!buckets[phaseIndex].length) return []

frontend/src/components/WaveChart.vue

@@ -713,7 +713,7 @@ function drawPvPreviews() {
             point.angle360 != null
             && point.angle360 >= phase.start
             && point.angle360 < phase.end
-          ))
+          )).sort((left, right) => (left.angle360 ?? 0) - (right.angle360 ?? 0))
           if (phasePoints.length < 2) return
           context.strokeStyle = phase.color
           context.beginPath()

viewer/backend/services/indicator_service.py

@@ -4,8 +4,10 @@
 （``wave_sample_one``），按缸径/行程/连杆/余隙把曲柄转角换算为容积（L），
 返回盖侧 / 轴侧两副压力功图（P-V）。
 
-首周期本身即一个完整冲程，样本按时间顺序对应曲柄角 0→360°，直接成闭环，
-无需额外相位旋转（与 backend/app/services/data_service.py 的示功图一致）。
+首周期本身即一个完整冲程。7/8/9 号机按
+``月度报告/沁水压缩机参数20250102.docx`` 的相位角把共享键相角旋转到各缸自身
+0°（显示角 = 共享角 − 相位角）；4/5/6 号机暂无资料，保持原逻辑。
+轴侧容积扣除活塞杆直径 78mm，盖侧用全缸径。
 """
 
 from __future__ import annotations
@@ -22,7 +24,12 @@ SIDES = ("盖侧", "轴侧")
 CYLINDER_BORE_MM = {"一": 360.0, "二": 490.0, "三": 390.0, "四": 490.0, "五": 390.0, "六": 490.0}
 PISTON_STROKE_MM = 148.0
 CONNECTING_ROD_LENGTH_MM = 460.0
+PISTON_ROD_DIAMETER_MM = 78.0
 CLEARANCE_VOLUME_L_BY_BORE = {490.0: 1.51, 390.0: 0.74, 360.0: 0.62}
+# 7/8/9 号机各缸相位角（月度报告/沁水压缩机参数20250102.docx）。
+CYLINDER_PHASE_DEGREE = {"一": 0.0, "二": 0.0, "三": 120.0, "四": 240.0, "五": 240.0, "六": 120.0}
+PHASE_ROTATION_UNITS = frozenset({"7", "8", "9"})
+PULSES_PER_REVOLUTION = 8
 MIN_SAMPLES = 60
 
 
@@ -31,18 +38,59 @@ def unit_number(device_code: str) -> str | None:
     return match.group(1) if match else None
 
 
-def volume_from_angle(cylinder: str, count: int) -> np.ndarray:
+def piecewise_angle(trigger: np.ndarray, count: int) -> np.ndarray:
+    """Crank angle from the 8 keyphase pulses; falls back to uniform sampling."""
+    high = np.nan_to_num(np.asarray(trigger, dtype=float), nan=-np.inf) >= 30.0
+    changes = np.diff(np.r_[False, high, False].astype(np.int8))
+    offsets = [int(offset) for offset in np.flatnonzero(changes == 1)]
+    if len(offsets) < PULSES_PER_REVOLUTION or offsets[0] != 0:
+        return np.linspace(0.0, 360.0, count, endpoint=False)
+    offsets = offsets[:PULSES_PER_REVOLUTION]
+    bounds = offsets + [count]
+    angle = np.full(count, np.nan, dtype=float)
+    for pulse in range(PULSES_PER_REVOLUTION):
+        start, end = bounds[pulse], min(bounds[pulse + 1], count)
+        if end <= start:
+            continue
+        angle[start:end] = np.linspace(
+            pulse * 45.0, (pulse + 1) * 45.0, end - start, endpoint=False,
+        )
+    if np.isnan(angle).any():
+        good = np.flatnonzero(~np.isnan(angle))
+        if not len(good):
+            return np.linspace(0.0, 360.0, count, endpoint=False)
+        angle = np.interp(np.arange(count), good, angle[good])
+    return angle
+
+
+def phase_degrees(unit: str, cylinder: str) -> float:
+    if unit in PHASE_ROTATION_UNITS:
+        return CYLINDER_PHASE_DEGREE[cylinder]
+    return 0.0
+
+
+def volume_from_angle(
+    cylinder: str,
+    side: str,
+    unit: str,
+    angle_degree: np.ndarray,
+) -> np.ndarray:
     bore = CYLINDER_BORE_MM[cylinder]
     clearance = CLEARANCE_VOLUME_L_BY_BORE[bore]
     crank_radius = PISTON_STROKE_MM / 2.0
-    rod = CONNECTING_ROD_LENGTH_MM
-    theta = np.linspace(0.0, 2.0 * np.pi, count, endpoint=False)
+    rod_length = CONNECTING_ROD_LENGTH_MM
+    rod_diameter = (
+        PISTON_ROD_DIAMETER_MM
+        if side == "轴侧" and unit in PHASE_ROTATION_UNITS
+        else 0.0
+    )
+    theta = np.deg2rad(np.asarray(angle_degree, dtype=float))
     travel = (
         crank_radius * (1.0 - np.cos(theta))
-        + rod
-        - np.sqrt(rod**2 - (crank_radius * np.sin(theta)) ** 2)
+        + rod_length
+        - np.sqrt(rod_length**2 - (crank_radius * np.sin(theta)) ** 2)
     )
-    piston_area = np.pi * (bore / 2.0) ** 2
+    piston_area = np.pi * ((bore / 2.0) ** 2 - (rod_diameter / 2.0) ** 2)
     return clearance + piston_area * travel / 1_000_000.0
 
 
@@ -65,22 +113,30 @@ def _side(unit: str, cylinder: str, side: str) -> dict[str, Any] | None:
                 return None
             file_id = int(row["id"])
             cursor.execute(
-                "SELECT signal_value FROM wave_sample_one WHERE wave_file_id = %s ORDER BY sample_index",
+                "SELECT signal_value, second_value FROM wave_sample_one "
+                "WHERE wave_file_id = %s ORDER BY sample_index",
                 (file_id,),
             )
-            values = [
-                float(item["signal_value"])
+            samples = [
+                (float(item["signal_value"]), item["second_value"])
                 for item in cursor.fetchall()
                 if item["signal_value"] is not None
             ]
     finally:
         connection.close()
 
-    if len(values) < MIN_SAMPLES:
+    if len(samples) < MIN_SAMPLES:
         return None
-    pressure = np.asarray(values, dtype=float)
-    volume = volume_from_angle(cylinder, len(pressure))
-    angle = np.linspace(0.0, 360.0, len(pressure), endpoint=False)
+    pressure = np.asarray([item[0] for item in samples], dtype=float)
+    trigger = np.asarray(
+        [float(item[1]) if item[1] is not None else np.nan for item in samples],
+        dtype=float,
+    )
+    angle = piecewise_angle(trigger, len(pressure))
+    phase = phase_degrees(unit, cylinder)
+    if phase:
+        angle = np.mod(angle - phase, 360.0)
+    volume = volume_from_angle(cylinder, side, unit, angle)
     sample_time = row["sample_time"]
     return {
         "side": side,