|
|
@@ -50,7 +50,11 @@ ANNOTATION_LABELS = ("正常", "异常")
|
|
|
UNIT_BATCH = {"7": 30, "8": 31, "9": 32}
|
|
|
# 全场点位(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})
|
|
|
|
|
|
# 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).
|
|
|
@@ -123,14 +127,20 @@ def _annotation_dict(row: dict[str, Any]) -> dict[str, Any]:
|
|
|
}
|
|
|
|
|
|
|
|
|
-def _validate_device_points(values: list[str] | tuple[str, ...] | None) -> list[str]:
|
|
|
- selected = list(values or DEVICE_POINTS)
|
|
|
- if not selected:
|
|
|
- raise ValueError("请至少选择一个测试点位")
|
|
|
+def _validate_device_points(
|
|
|
+ values: list[str] | tuple[str, ...] | None,
|
|
|
+ *,
|
|
|
+ allow_empty: bool = False,
|
|
|
+) -> list[str]:
|
|
|
+ """校验波形点位。None 表示未传参 → 回退为全部点位;显式空列表仅在 allow_empty 时放行。"""
|
|
|
+ selected = list(DEVICE_POINTS) if values is None else list(values)
|
|
|
invalid = [value for value in selected if value not in DEVICE_POINTS]
|
|
|
if invalid:
|
|
|
raise ValueError(f"不支持的测试点位:{'、'.join(invalid)}")
|
|
|
- return [value for value in DEVICE_POINTS if value in selected]
|
|
|
+ result = [value for value in DEVICE_POINTS if value in selected]
|
|
|
+ if not result and not allow_empty:
|
|
|
+ raise ValueError("请至少选择一个测试点位")
|
|
|
+ return result
|
|
|
|
|
|
|
|
|
def _primary_device_point(points: list[str]) -> str:
|
|
|
@@ -166,6 +176,8 @@ 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]] = {}
|
|
|
|
|
|
@property
|
|
|
def source(self) -> str:
|
|
|
@@ -290,15 +302,40 @@ class DataService:
|
|
|
result["source"] = source
|
|
|
return result
|
|
|
|
|
|
+ def _pks_batch_range(self, unit: str) -> tuple[datetime, datetime] | None:
|
|
|
+ """机组 pks 批次的时间覆盖范围(仅查一次并缓存)。"""
|
|
|
+ cached = self._pks_batch_bounds.get(unit)
|
|
|
+ if cached is not None:
|
|
|
+ return cached
|
|
|
+ batch = UNIT_BATCH[unit]
|
|
|
+ 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,),
|
|
|
+ )
|
|
|
+ row = cursor.fetchone()
|
|
|
+ bounds = (row["lo"], row["hi"]) if row and row["lo"] is not None else None
|
|
|
+ self._pks_batch_bounds[unit] = bounds
|
|
|
+ return bounds
|
|
|
+
|
|
|
def site_points(self, device_part: str) -> dict[str, Any]:
|
|
|
- """机组(7/8/9)的全场点位记录(site_point 表中 YSJ{机组号}_1..41)。"""
|
|
|
+ """机组(7/8/9)的全场点位记录(site_point 表中 YSJ{机组号}_1..41)。
|
|
|
|
|
|
+ 附带该机组 pks 批次的时间覆盖范围,供“仅 PKS 点位”组合自动赋值
|
|
|
+ 开始/结束时间。
|
|
|
+ """
|
|
|
unit = _unit_number(device_part)
|
|
|
|
|
|
def database_query():
|
|
|
if unit is None:
|
|
|
- return []
|
|
|
- allowed = {f"YSJ{unit}_{n}" for n in range(1, 42)}
|
|
|
+ return {"items": [], "minTime": None, "maxTime": None}
|
|
|
+ allowed = {
|
|
|
+ f"YSJ{unit}_{n}"
|
|
|
+ for n in range(1, 42)
|
|
|
+ if n not in _EXCLUDED_SITE_INDEXES
|
|
|
+ }
|
|
|
with get_connection() as connection:
|
|
|
with connection.cursor() as cursor:
|
|
|
cursor.execute(
|
|
|
@@ -312,16 +349,23 @@ class DataService:
|
|
|
if row["ItemName"] in allowed
|
|
|
]
|
|
|
items.sort(key=lambda item: int(item["itemName"].rsplit("_", 1)[1]))
|
|
|
- return items
|
|
|
+ bounds = self._pks_batch_range(unit)
|
|
|
+ return {
|
|
|
+ "items": items,
|
|
|
+ "minTime": _time_string(bounds[0]) if bounds else None,
|
|
|
+ "maxTime": _time_string(bounds[1]) if bounds else None,
|
|
|
+ }
|
|
|
|
|
|
def demo_query():
|
|
|
- return []
|
|
|
+ return {"items": [], "minTime": None, "maxTime": None}
|
|
|
|
|
|
- items, source = self._run_with_fallback(database_query, demo_query)
|
|
|
+ result, source = self._run_with_fallback(database_query, demo_query)
|
|
|
return {
|
|
|
"source": source,
|
|
|
"unit": unit,
|
|
|
- "items": items,
|
|
|
+ "items": result["items"],
|
|
|
+ "minTime": result["minTime"],
|
|
|
+ "maxTime": result["maxTime"],
|
|
|
"notice": self._source_notice(source),
|
|
|
}
|
|
|
|
|
|
@@ -346,22 +390,31 @@ class DataService:
|
|
|
for point in points:
|
|
|
timestamp = datetime.strptime(point["sampleTime"], "%Y-%m-%d %H:%M:%S")
|
|
|
rounded.append(datetime.fromtimestamp(round(timestamp.timestamp() / 5.0) * 5))
|
|
|
- placeholders = ", ".join(["%s"] * len(rounded))
|
|
|
- for item_name, column in columns:
|
|
|
+ chunk_size = 1000
|
|
|
+ select_expr = ", ".join(f"`{column}`" for _item, column in columns)
|
|
|
+ for index in range(0, len(rounded), chunk_size):
|
|
|
+ chunk = rounded[index:index + chunk_size]
|
|
|
+ placeholders = ", ".join(["%s"] * len(chunk))
|
|
|
with get_connection() as connection:
|
|
|
with connection.cursor() as cursor:
|
|
|
cursor.execute(
|
|
|
- f"SELECT sample_time, `{column}` AS value FROM pks_long_sample "
|
|
|
+ f"SELECT sample_time, {select_expr} FROM pks_long_sample "
|
|
|
f"WHERE import_batch_id = %s AND sample_time IN ({placeholders})",
|
|
|
- (UNIT_BATCH[unit], *rounded),
|
|
|
+ (UNIT_BATCH[unit], *chunk),
|
|
|
)
|
|
|
rows = cursor.fetchall()
|
|
|
- by_time = {row["sample_time"]: row["value"] for row in rows}
|
|
|
- for index, target_time in enumerate(rounded):
|
|
|
- value = by_time.get(target_time)
|
|
|
- points[index].setdefault("siteValues", {})[item_name] = (
|
|
|
- float(value) if value is not None else None
|
|
|
- )
|
|
|
+ by_time: dict[datetime, dict[str, Any]] = {
|
|
|
+ row["sample_time"]: row for row in rows
|
|
|
+ }
|
|
|
+ for point_index in range(index, min(index + chunk_size, len(rounded))):
|
|
|
+ target_time = rounded[point_index]
|
|
|
+ row = by_time.get(target_time)
|
|
|
+ if row is None:
|
|
|
+ continue
|
|
|
+ site_values = points[point_index].setdefault("siteValues", {})
|
|
|
+ for item_name, column in columns:
|
|
|
+ value = row[column]
|
|
|
+ site_values[item_name] = float(value) if value is not None else None
|
|
|
|
|
|
@staticmethod
|
|
|
def _build_site_series(
|
|
|
@@ -394,19 +447,20 @@ class DataService:
|
|
|
half = timedelta(seconds=PKS_WINDOW_SECONDS // 2)
|
|
|
query_start = min(timestamp for _index, timestamp in centers) - half
|
|
|
query_end = max(timestamp for _index, timestamp in centers) + half
|
|
|
+ select_expr = ", ".join(f"`{column}`" for _item, column in columns)
|
|
|
+ with get_connection() as connection:
|
|
|
+ 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 "
|
|
|
+ "ORDER BY sample_time",
|
|
|
+ (UNIT_BATCH[unit], query_start, query_end),
|
|
|
+ )
|
|
|
+ rows = cursor.fetchall()
|
|
|
+ sample_times = [row["sample_time"] for row in rows]
|
|
|
series = []
|
|
|
for item_name, column in columns:
|
|
|
- with get_connection() as connection:
|
|
|
- with connection.cursor() as cursor:
|
|
|
- cursor.execute(
|
|
|
- f"SELECT sample_time, `{column}` AS value FROM pks_long_sample "
|
|
|
- "WHERE import_batch_id = %s AND sample_time >= %s AND sample_time < %s "
|
|
|
- "ORDER BY sample_time",
|
|
|
- (UNIT_BATCH[unit], query_start, query_end),
|
|
|
- )
|
|
|
- rows = cursor.fetchall()
|
|
|
- sample_times = [row["sample_time"] for row in rows]
|
|
|
- values = [row["value"] for row in rows]
|
|
|
+ values = [row[column] for row in rows]
|
|
|
data: list[dict[str, Any]] = []
|
|
|
for index, center in centers:
|
|
|
low = bisect_left(sample_times, center - half)
|
|
|
@@ -443,8 +497,6 @@ class DataService:
|
|
|
) -> dict[str, Any]:
|
|
|
if not device_part.strip():
|
|
|
raise ValueError("机组与部位不能为空")
|
|
|
- selected_points = _validate_device_points(device_points)
|
|
|
- point_names = [f"{device_part}{point}" for point in selected_points]
|
|
|
status_filters = {value for value in (status_filter or [])}
|
|
|
unknown = status_filters - {"abnormal", "no_cycle"}
|
|
|
if unknown:
|
|
|
@@ -453,6 +505,40 @@ class DataService:
|
|
|
end = _parse_time(max_time)
|
|
|
if start and end and start > end:
|
|
|
raise ValueError("开始时间不能晚于结束时间")
|
|
|
+ unit = _unit_number(device_part)
|
|
|
+ site_list = list(site_points or [])
|
|
|
+ site_columns: list[tuple[str, str]] = []
|
|
|
+ if unit:
|
|
|
+ site_columns = [
|
|
|
+ (item, column)
|
|
|
+ for item in site_list
|
|
|
+ if (column := _site_point_column(item, unit)) is not None
|
|
|
+ ]
|
|
|
+ if device_points is None or device_points == []:
|
|
|
+ selected_points = (
|
|
|
+ [] if (unit and site_columns) else _validate_device_points(None)
|
|
|
+ )
|
|
|
+ else:
|
|
|
+ selected_points = _validate_device_points(device_points, allow_empty=True)
|
|
|
+ pks_only = bool(unit and site_columns and not selected_points)
|
|
|
+ if not selected_points and not pks_only:
|
|
|
+ raise ValueError("请至少选择一个测试点位")
|
|
|
+ point_names = [f"{device_part}{point}" for point in selected_points]
|
|
|
+
|
|
|
+ if pks_only:
|
|
|
+ points, source = self._run_with_fallback(
|
|
|
+ lambda: self._pks_time_points(unit, site_columns, start, end, include_stopped),
|
|
|
+ lambda: [],
|
|
|
+ )
|
|
|
+ return {
|
|
|
+ "source": source,
|
|
|
+ "devicePart": device_part,
|
|
|
+ "devicePoints": [],
|
|
|
+ "total": len(points),
|
|
|
+ "points": points,
|
|
|
+ "referencePoints": [],
|
|
|
+ "notice": self._source_notice(source),
|
|
|
+ }
|
|
|
|
|
|
def database_query():
|
|
|
point_placeholders = ", ".join(["%s"] * len(point_names))
|
|
|
@@ -581,6 +667,141 @@ class DataService:
|
|
|
for index, point in enumerate(grouped.values())
|
|
|
]
|
|
|
|
|
|
+ def _pks_time_points(
|
|
|
+ self,
|
|
|
+ unit: str,
|
|
|
+ columns: list[tuple[str, str]],
|
|
|
+ start: datetime | None,
|
|
|
+ end: datetime | None,
|
|
|
+ include_stopped: bool,
|
|
|
+ ) -> list[dict[str, Any]]:
|
|
|
+ """PKS-only 时间点:按 15 分钟槽从 pks_long_sample 生成骨架。
|
|
|
+
|
|
|
+ columns 为 [(item_name, pks 列名), ...],siteValues 直挂所选各列
|
|
|
+ 原值。停机/运转只由该机组批次内的 YSJ_41 决定(>0=运转、=0=停机),
|
|
|
+ 未勾选停机时仅保留运转锚点;勾选停机则停机锚点一并显示。
|
|
|
+ """
|
|
|
+ batch = UNIT_BATCH[unit]
|
|
|
+ clauses = ["import_batch_id = %s"]
|
|
|
+ params: list[Any] = [batch]
|
|
|
+ if start:
|
|
|
+ clauses.append("sample_time >= %s")
|
|
|
+ params.append(start)
|
|
|
+ if end:
|
|
|
+ clauses.append("sample_time <= %s")
|
|
|
+ params.append(end)
|
|
|
+ with get_connection() as connection:
|
|
|
+ with connection.cursor() as cursor:
|
|
|
+ cursor.execute(
|
|
|
+ f"""
|
|
|
+ SELECT MIN(sample_time) AS anchor
|
|
|
+ FROM pks_long_sample
|
|
|
+ WHERE {' AND '.join(clauses)}
|
|
|
+ GROUP BY FLOOR(UNIX_TIMESTAMP(sample_time) / 900)
|
|
|
+ ORDER BY anchor
|
|
|
+ """,
|
|
|
+ params,
|
|
|
+ )
|
|
|
+ anchors = [row["anchor"] for row in cursor.fetchall()]
|
|
|
+ if not anchors:
|
|
|
+ return []
|
|
|
+ if len(anchors) > PKS_STRIP_MAX_ANCHORS:
|
|
|
+ step = (len(anchors) + PKS_STRIP_MAX_ANCHORS - 1) // PKS_STRIP_MAX_ANCHORS
|
|
|
+ sampled = anchors[::step]
|
|
|
+ if sampled[-1] != anchors[-1]:
|
|
|
+ sampled.append(anchors[-1])
|
|
|
+ anchors = sampled
|
|
|
+ chunk_size = 1000
|
|
|
+ select_expr = ", ".join(f"`{column}`" for _item, column in columns)
|
|
|
+ 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):
|
|
|
+ chunk = anchors[index:index + chunk_size]
|
|
|
+ placeholders = ", ".join(["%s"] * len(chunk))
|
|
|
+ with get_connection() as connection:
|
|
|
+ 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),
|
|
|
+ )
|
|
|
+ rows = cursor.fetchall()
|
|
|
+ for row in rows:
|
|
|
+ anchor_time = row["sample_time"]
|
|
|
+ values = values_by_time[anchor_time]
|
|
|
+ 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
|
|
|
+ )
|
|
|
+ points: list[dict[str, Any]] = []
|
|
|
+ for anchor in anchors:
|
|
|
+ running = running_by_time.get(anchor, False)
|
|
|
+ if not include_stopped and not running:
|
|
|
+ continue
|
|
|
+ points.append(
|
|
|
+ {
|
|
|
+ "sampleTime": _time_string(anchor),
|
|
|
+ "files": {},
|
|
|
+ "siteValues": values_by_time[anchor],
|
|
|
+ "machineRunning": running,
|
|
|
+ },
|
|
|
+ )
|
|
|
+ for index, point in enumerate(points):
|
|
|
+ point["index"] = index
|
|
|
+ return points
|
|
|
+
|
|
|
+ def _pks_only_wave_window(
|
|
|
+ self,
|
|
|
+ device_part: str,
|
|
|
+ points: list[dict[str, Any]],
|
|
|
+ site_points: list[str],
|
|
|
+ unit: str,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ """PKS-only 波形窗口:不取 wave_file,仅构建所选 PKS 点位的微曲线。"""
|
|
|
+
|
|
|
+ def database_query():
|
|
|
+ return {
|
|
|
+ "points": points,
|
|
|
+ "siteSeries": self._build_site_series(site_points, unit, points),
|
|
|
+ }
|
|
|
+
|
|
|
+ def demo_query():
|
|
|
+ return {"points": points, "siteSeries": {"points": [], "series": []}}
|
|
|
+
|
|
|
+ result, source = self._run_with_fallback(database_query, demo_query)
|
|
|
+ return {
|
|
|
+ "devicePart": device_part,
|
|
|
+ "devicePoints": [],
|
|
|
+ "primaryPoint": None,
|
|
|
+ "points": points,
|
|
|
+ "xMin": 0,
|
|
|
+ "xMax": len(points),
|
|
|
+ "series": [],
|
|
|
+ "secondSeries": {
|
|
|
+ "name": "周期数据",
|
|
|
+ "color": "#f56c6c",
|
|
|
+ "sourceMeasurementType": None,
|
|
|
+ "sourceDevicePoint": None,
|
|
|
+ "data": [],
|
|
|
+ "finiteCount": 0,
|
|
|
+ "nonZeroCount": 0,
|
|
|
+ "min": None,
|
|
|
+ "max": None,
|
|
|
+ },
|
|
|
+ "angleSeries": {"color": "#d59b2b", "data": []},
|
|
|
+ "volumeSeries": {"color": "#4d9e6f", "data": [], "info": None},
|
|
|
+ "cycles": [],
|
|
|
+ "triggerXs": [],
|
|
|
+ "files": [],
|
|
|
+ "diagnostics": [],
|
|
|
+ "siteSeries": result["siteSeries"],
|
|
|
+ "source": source,
|
|
|
+ "notice": self._source_notice(source),
|
|
|
+ }
|
|
|
+
|
|
|
def wave_window(
|
|
|
self,
|
|
|
device_part: str,
|
|
|
@@ -591,12 +812,27 @@ class DataService:
|
|
|
first_cycle_only: bool = False,
|
|
|
site_points: list[str] | None = None,
|
|
|
) -> dict[str, Any]:
|
|
|
- selected_points = _validate_device_points(device_points)
|
|
|
+ selected_points = _validate_device_points(device_points, allow_empty=True)
|
|
|
if not points:
|
|
|
raise ValueError("至少选择一个时间点")
|
|
|
if len(points) > 200:
|
|
|
raise ValueError("单次最多预览 200 个时间点,请缩小时间窗口")
|
|
|
max_points = min(max(int(max_points), 256), 200000)
|
|
|
+ unit = _unit_number(device_part)
|
|
|
+ site_list = list(site_points or [])
|
|
|
+ site_columns: list[tuple[str, str]] = []
|
|
|
+ if unit:
|
|
|
+ site_columns = [
|
|
|
+ (item, column)
|
|
|
+ for item in site_list
|
|
|
+ if (column := _site_point_column(item, unit)) is not None
|
|
|
+ ]
|
|
|
+ pks_only = bool(unit and site_columns and not selected_points)
|
|
|
+ if not selected_points and not pks_only:
|
|
|
+ raise ValueError("请至少选择一个测试点位")
|
|
|
+
|
|
|
+ if pks_only:
|
|
|
+ return self._pks_only_wave_window(device_part, points, site_list, unit)
|
|
|
|
|
|
def database_query():
|
|
|
if first_cycle_only:
|