PKS 支持 4/5/6 号机：单位正则放开至4-9、批次改按 compressor_data_import 动态关联（多批次合并去重）、site_point 排除序号按机组区分；query_options 补纯 PKS 机组入口（5号机组）；前端放开4-9并处理纯PKS机组

backend/app/services/data_service.py

@@ -54,16 +54,27 @@ OIL_PRESSURE_DEVICE_POINT = "压力"
 VIBRATION_DEVICE_POINT = "振动"
 OIL_PRESSURE_SCAN_DAYS = 14
 
-# PKS 全场点位：机组号 -> pks_long_sample.import_batch_id
-# 7号机=30、8号机=31、9号机=32。
-UNIT_BATCH = {"7": 30, "8": 31, "9": 32}
+# PKS 全场点位：机组号 -> compressor_data_import 中 directory_path 为 "{号}#"
+# 且 data_type='PKS' 的所有导入批次（同一机组可能有多个时间段批次）。
+UNIT_PKS_DIRECTORY_SUFFIX = "#"
 # 全场点位(PKS)每个时间点的曲线窗口：以采样时刻为中心的前后各 7.5 分钟。
 PKS_WINDOW_SECONDS = 15 * 60
 # 仅 PKS 时间点返回给时间条的最大锚点数：超过后按等步长均匀采样（首尾必保）。
 PKS_STRIP_MAX_ANCHORS = 5000
-_SITE_POINT_PATTERN = re.compile(r"^YSJ([789])_([1-9]|1[0-9]|2[0-9]|3[0-9]|4[0-1])$")
-# 不作为全场点位(PKS)下拉候选的序号（对应机组自己的运行状态/转速信号）。
-_EXCLUDED_SITE_INDEXES = frozenset({33, 34, 35, 41})
+_SITE_POINT_PATTERN = re.compile(r"^YSJ([4-9])_([1-9]|[1-3][0-9]|4[0-4])$")
+_SITE_INDEX_MAX = 44
+# 不作为全场点位(PKS)下拉候选的序号（各机组不同：备用/累计运行等信号）。
+_EXCLUDED_SITE_INDEXES: dict[str, frozenset[int]] = {
+    "4": frozenset({31, 32, 33, 34, 35, 36}),
+    "5": frozenset(),
+    "6": frozenset(),
+    "7": frozenset({33, 34, 35, 41}),
+    "8": frozenset({33, 34, 35, 41}),
+    "9": frozenset({33, 34, 35, 41}),
+}
+# 仅这些机组的 YSJ_41 是转速/运行信号，可用于停机判断；4/5/6 的 YSJ_41 非转速，
+# 无法据此判断停机，一律按运转处理。
+_RUNNING_SIGNAL_UNITS = frozenset({"7", "8", "9"})
 
 # Extra samples fetched past cycle_end so detect_cycles can see the zero marker
 # that closes the first cycle (cycle_end is exclusive, the marker sits at it).
@@ -234,8 +245,8 @@ def _primary_device_point(points: list[str]) -> str:
 
 
 def _unit_number(device_part: str) -> str | None:
-    """从 机组与部位 提取机组号（7/8/9），非 7/8/9 机组返回 None。"""
-    match = re.match(r"^([789])号机组", device_part.strip())
+    """从 机组与部位 提取机组号（4~9），非 4~9 机组返回 None。"""
+    match = re.match(r"^([4-9])号机组", device_part.strip())
     return match.group(1) if match else None
 
 
@@ -258,8 +269,10 @@ class DataService:
         self._last_db_error = ""
         self._demo_annotations: dict[int, dict[str, Any]] = {}
         self._demo_annotation_seq = 1
-        # 机组号 -> pks_long_sample 该批次 [min_time, max_time]，进程内只查一次。
-        self._pks_batch_bounds: dict[str, tuple[datetime, datetime]] = {}
+        # 机组号 -> pks_long_sample 该机组所有批次 [min_time, max_time]，进程内只查一次。
+        self._pks_batch_bounds: dict[str, tuple[datetime, datetime] | None] = {}
+        # 机组号 -> compressor_data_import 中该机组 PKS 批次 id 列表，进程内只查一次。
+        self._pks_batch_ids_cache: dict[str, list[int]] = {}
 
     @property
     def source(self) -> str:
@@ -324,6 +337,8 @@ class DataService:
 
         rows, source = self._run_with_fallback(database_query, self._demo_options)
         device_parts = list(dict.fromkeys(row["devicePart"] for row in rows))
+        if source == "database":
+            device_parts.extend(self._pks_only_device_parts(device_parts))
         return {
             "source": source,
             "measurementTypes": list(MEASUREMENT_TYPES),
@@ -334,6 +349,26 @@ class DataService:
             "notice": self._source_notice(source),
         }
 
+    def _pks_only_device_parts(self, existing: list[str]) -> list[str]:
+        """有 PKS 数据但 wave_file 无部位的机组，补一个纯 PKS 的"{号}号机组"入口。"""
+        existing_units = {unit for part in existing if (unit := _unit_number(part)) is not None}
+        try:
+            with get_connection() as connection:
+                with connection.cursor() as cursor:
+                    cursor.execute(
+                        "SELECT DISTINCT directory_path FROM compressor_data_import "
+                        "WHERE data_type = 'PKS' AND directory_path REGEXP '^[4-9]#$'",
+                    )
+                    directories = [str(row["directory_path"]) for row in cursor.fetchall()]
+        except Exception:
+            return []
+        parts: list[str] = []
+        for directory in directories:
+            unit = directory.rstrip(UNIT_PKS_DIRECTORY_SUFFIX).strip()
+            if unit.isdigit() and unit not in existing_units:
+                parts.append(f"{unit}号机组")
+        return parts
+
     def abnormal_counts(self) -> dict[str, int]:
         """每个 point_name 的异常文件数量（tspluse_status > 0）。"""
 
@@ -384,18 +419,41 @@ class DataService:
         result["source"] = source
         return result
 
+    def _pks_batch_ids(self, unit: str) -> list[int]:
+        """机组 PKS 导入批次 id 列表（directory_path = "{号}#" 且 data_type='PKS'）。"""
+        cached = self._pks_batch_ids_cache.get(unit)
+        if cached is not None:
+            return cached
+        with get_connection() as connection:
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    "SELECT id FROM compressor_data_import "
+                    "WHERE data_type = 'PKS' AND directory_path = %s ORDER BY id",
+                    (f"{unit}{UNIT_PKS_DIRECTORY_SUFFIX}",),
+                )
+                batches = [int(row["id"]) for row in cursor.fetchall()]
+        self._pks_batch_ids_cache[unit] = batches
+        return batches
+
+    @staticmethod
+    def _in_clause(values: list[Any]) -> str:
+        return ", ".join(["%s"] * len(values))
+
     def _pks_batch_range(self, unit: str) -> tuple[datetime, datetime] | None:
-        """机组 pks 批次的时间覆盖范围（仅查一次并缓存）。"""
+        """机组 pks 所有批次的时间覆盖范围（仅查一次并缓存）。"""
         cached = self._pks_batch_bounds.get(unit)
         if cached is not None:
             return cached
-        batch = UNIT_BATCH[unit]
+        batches = self._pks_batch_ids(unit)
+        if not batches:
+            self._pks_batch_bounds[unit] = None
+            return None
         with get_connection() as connection:
             with connection.cursor() as cursor:
                 cursor.execute(
                     "SELECT MIN(sample_time) AS lo, MAX(sample_time) AS hi "
-                    "FROM pks_long_sample WHERE import_batch_id = %s",
-                    (batch,),
+                    f"FROM pks_long_sample WHERE import_batch_id IN ({self._in_clause(batches)})",
+                    tuple(batches),
                 )
                 row = cursor.fetchone()
         bounds = (row["lo"], row["hi"]) if row and row["lo"] is not None else None
@@ -403,7 +461,7 @@ class DataService:
         return bounds
 
     def site_points(self, device_part: str) -> dict[str, Any]:
-        """机组(7/8/9)的全场点位记录（site_point 表中 YSJ{机组号}_1..41）。
+        """机组(4~9)的全场点位记录（site_point 表中 YSJ{机组号}_*）。
 
         附带该机组 pks 批次的时间覆盖范围，供“仅 PKS 点位”组合自动赋值
         开始/结束时间。
@@ -415,8 +473,8 @@ class DataService:
                 return {"items": [], "minTime": None, "maxTime": None}
             allowed = {
                 f"YSJ{unit}_{n}"
-                for n in range(1, 42)
-                if n not in _EXCLUDED_SITE_INDEXES
+                for n in range(1, _SITE_INDEX_MAX + 1)
+                if n not in _EXCLUDED_SITE_INDEXES.get(unit, frozenset())
             }
             with get_connection() as connection:
                 with connection.cursor() as cursor:
@@ -451,8 +509,7 @@ class DataService:
             "notice": self._source_notice(source),
         }
 
-    @staticmethod
-    def _attach_site_values(points: list[dict[str, Any]], unit: str, site_points: list[str]) -> None:
+    def _attach_site_values(self, points: list[dict[str, Any]], unit: str, site_points: list[str]) -> None:
         """把 pks_long_sample 的最近邻值挂到每个时间点的 siteValues 上。
 
         pks 数据 5 秒一条、wave_file 15 分钟一条。按用户口径做分钟/5秒级对齐：
@@ -464,7 +521,8 @@ class DataService:
             for item_name in site_points
             if (column := _site_point_column(item_name, unit)) is not None
         ]
-        if not columns or not points:
+        batches = self._pks_batch_ids(unit)
+        if not columns or not points or not batches:
             for point in points:
                 point.setdefault("siteValues", {})
             return
@@ -481,8 +539,9 @@ class DataService:
                 with connection.cursor() as cursor:
                     cursor.execute(
                         f"SELECT sample_time, {select_expr} FROM pks_long_sample "
-                        f"WHERE import_batch_id = %s AND sample_time IN ({placeholders})",
-                        (UNIT_BATCH[unit], *chunk),
+                        f"WHERE import_batch_id IN ({self._in_clause(batches)}) "
+                        f"AND sample_time IN ({placeholders})",
+                        (*batches, *chunk),
                     )
                     rows = cursor.fetchall()
             by_time: dict[datetime, dict[str, Any]] = {
@@ -498,8 +557,8 @@ class DataService:
                     value = row[column]
                     site_values[item_name] = float(value) if value is not None else None
 
-    @staticmethod
     def _build_site_series(
+        self,
         site_points: list[str],
         unit: str | None,
         points: list[dict[str, Any]],
@@ -524,7 +583,8 @@ class DataService:
             except (TypeError, ValueError):
                 continue
             centers.append((index, timestamp))
-        if not columns or not centers:
+        batches = self._pks_batch_ids(unit) if unit else []
+        if not columns or not centers or not batches:
             return {"points": site_points, "series": []}
         half = timedelta(seconds=PKS_WINDOW_SECONDS // 2)
         query_start = min(timestamp for _index, timestamp in centers) - half
@@ -534,15 +594,21 @@ class DataService:
             with connection.cursor() as cursor:
                 cursor.execute(
                     f"SELECT sample_time, {select_expr} FROM pks_long_sample "
-                    "WHERE import_batch_id = %s AND sample_time >= %s AND sample_time < %s "
+                    f"WHERE import_batch_id IN ({self._in_clause(batches)}) "
+                    "AND sample_time >= %s AND sample_time < %s "
                     "ORDER BY sample_time",
-                    (UNIT_BATCH[unit], query_start, query_end),
+                    (*batches, query_start, query_end),
                 )
                 rows = cursor.fetchall()
-        sample_times = [row["sample_time"] for row in rows]
+        # 同一机组可能有多个时间段批次，按 sample_time 去重，避免边界重叠重复绘制。
+        dedup: dict[datetime, dict[str, Any]] = {}
+        for row in rows:
+            dedup.setdefault(row["sample_time"], row)
+        ordered = [dedup[timestamp] for timestamp in sorted(dedup)]
+        sample_times = [row["sample_time"] for row in ordered]
         series = []
         for item_name, column in columns:
-            values = [row[column] for row in rows]
+            values = [row[column] for row in ordered]
             data: list[dict[str, Any]] = []
             for index, center in centers:
                 low = bisect_left(sample_times, center - half)
@@ -1697,12 +1763,16 @@ class DataService:
         """PKS-only 时间点：按 15 分钟槽从 pks_long_sample 生成骨架。
 
         columns 为 [(item_name, pks 列名), ...]，siteValues 直挂所选各列
-        原值。停机/运转只由该机组批次内的 YSJ_41 决定（>0=运转、=0=停机），
-        未勾选停机时仅保留运转锚点；勾选停机则停机锚点一并显示。
+        原值。7/8/9 机组的停机/运转由该机组批次内的 YSJ_41（转速）决定
+        （>0=运转、=0=停机）；4/5/6 无转速信号，按运转处理。未勾选停机时
+        仅保留运转锚点；勾选停机则停机锚点一并显示。
         """
-        batch = UNIT_BATCH[unit]
-        clauses = ["import_batch_id = %s"]
-        params: list[Any] = [batch]
+        batches = self._pks_batch_ids(unit)
+        if not batches:
+            return []
+        batch_clause = f"import_batch_id IN ({self._in_clause(batches)})"
+        clauses = [batch_clause]
+        params: list[Any] = list(batches)
         if start:
             clauses.append("sample_time >= %s")
             params.append(start)
@@ -1732,6 +1802,7 @@ class DataService:
             anchors = sampled
         chunk_size = 1000
         select_expr = ", ".join(f"`{column}`" for _item, column in columns)
+        use_running_signal = unit in _RUNNING_SIGNAL_UNITS
         values_by_time: dict[datetime, dict[str, float | None]] = {anchor: {} for anchor in anchors}
         running_by_time: dict[datetime, bool] = {}
         for index in range(0, len(anchors), chunk_size):
@@ -1741,8 +1812,8 @@ class DataService:
                 with connection.cursor() as cursor:
                     cursor.execute(
                         f"SELECT sample_time, {select_expr}, `YSJ_41` FROM pks_long_sample "
-                        f"WHERE import_batch_id = %s AND sample_time IN ({placeholders})",
-                        (batch, *chunk),
+                        f"WHERE {batch_clause} AND sample_time IN ({placeholders})",
+                        (*batches, *chunk),
                     )
                     rows = cursor.fetchall()
             for row in rows:
@@ -1751,10 +1822,13 @@ class DataService:
                 for item_name, column in columns:
                     value = row[column]
                     values[item_name] = float(value) if value is not None else None
-                running_value = row["YSJ_41"]
-                running_by_time[anchor_time] = bool(
-                    running_value is not None and running_value > 0
-                )
+                if use_running_signal:
+                    running_value = row["YSJ_41"]
+                    running_by_time[anchor_time] = bool(
+                        running_value is not None and running_value > 0
+                    )
+                else:
+                    running_by_time.setdefault(anchor_time, True)
         points: list[dict[str, Any]] = []
         for anchor in anchors:
             running = running_by_time.get(anchor, False)

frontend/src/App.vue

@@ -63,7 +63,7 @@ const deviceParts = computed(() => queryMeta.value?.deviceParts ?? [])
 const devicePoints = computed(() => queryMeta.value?.devicePoints ?? DEVICE_POINTS)
 
 const unitNumber = computed(() => {
-  const match = /^([789])号机组/.exec(selectedDevicePart.value)
+  const match = /^([4-9])号机组/.exec(selectedDevicePart.value)
   return match ? match[1] : null
 })
 const isPksMode = computed(() => !!unitNumber.value)
@@ -210,14 +210,14 @@ function rememberPart(part: string) {
 }
 
 function unitOfPart(part: string): string | null {
-  const match = /^([789])号机组/.exec(part)
+  const match = /^([4-9])号机组/.exec(part)
   return match ? match[1] : null
 }
 
 function mirrorSiteAcrossUnits(site: string[], toUnit: string, available: Set<string>): string[] {
   const result: string[] = []
   for (const item of site) {
-    const match = /^YSJ([789])_(\d+)$/.exec(item)
+    const match = /^YSJ([4-9])_(\d+)$/.exec(item)
     if (!match || match[1] === toUnit) continue
     const mirrored = `YSJ${toUnit}_${match[2]}`
     if (available.has(mirrored)) result.push(mirrored)
@@ -471,7 +471,8 @@ async function onDevicePartChange(previous?: string) {
     if (currentUnit) site = mirrorSiteAcrossUnits(source.site, currentUnit, available)
   }
   if (!wave.length && !(currentUnit && site.length)) {
-    wave = [PRIMARY_DEVICE_POINT]
+    const hasWaveOptions = (queryMeta.value?.options ?? []).some((option) => option.devicePart === current)
+    wave = hasWaveOptions ? [PRIMARY_DEVICE_POINT] : []
   }
   selectedDevicePoints.value = wave
   selectedSitePoints.value = site