deploy_pickup/core/pump_state_monitor.py

# -*- coding: utf-8 -*-
"""
pump_state_monitor.py - 泵状态监控模块


================================================================================
使用示例
================================================================================

# 初始化
from pump_state_monitor import PumpStateMonitor

monitor = PumpStateMonitor(
    scada_url="http://47.96.12.136:8788/api/v1/jinke-cloud/device/current-data",
    scada_jwt="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJJRCI6NywiVXNlcm5hbWUiOiJhZG1pbiIsIkRlcCI6IjEzNSIsImV4cCI6MTc3NjExOTExNCwiaXNzIjoiZ2luLWJsb2cifQ.0HTtzHZjyd2mHo8VCy8icYROxmntRMuQhyoZsAYRL_M",
    project_id=92,
    transition_window_minutes=15
)

# 单泵查询
is_running, _ = monitor.update_pump_state("pump_1", "C.M.RO1_GYB@run", "RO1高压泵")
in_transition = monitor.is_in_transition("pump_1")

# 多泵批量查询
pump_configs = [
    {"point": "C.M.RO1_GYB@run", "name": "RO1高压泵"},
    {"point": "C.M.RO2_GYB@run", "name": "RO2高压泵"}
]
in_transition, pumps = monitor.check_pumps_transition(pump_configs)

# 业务逻辑
for audio_file in get_audio_files():
    is_running, _ = monitor.update_pump_state("pump_1", "C.M.RO1_GYB@run", "RO1高压泵")
    if not is_running:
        skip_audio(audio_file)     # 泵停机，跳过
        continue
    if monitor.is_in_transition("pump_1"):
        skip_audio(audio_file)     # 过渡期，跳过
        continue
    process_audio(audio_file)      # 泵运行且稳定，正常处理

================================================================================
时间线案例 (过渡期=15分钟)
================================================================================
10:00:00  首次调用    -> is_running=True,  in_transition=False  (初始化)
10:05:00  泵停机      -> is_running=False, in_transition=True   (进入过渡期)
10:20:00  过渡期结束  -> is_running=False, in_transition=False  (15分钟后)
10:25:00  泵启动      -> is_running=True,  in_transition=True   (再次进入过渡期)
10:40:00  过渡期结束  -> is_running=True,  in_transition=False  (可正常处理)

================================================================================
内部机制
================================================================================
1. 查询缓存: 30秒内同一泵只查询一次SCADA，调用方无需控制频率
2. 状态变化: 本地比较前后状态，变化时自动记录时间
3. 过渡期判定: 基于状态变化时间，默认15分钟窗口
4. 首次查询: 不视为过渡期（假设泵已稳定运行）

日志: 泵状态初始化/变化时输出 INFO 级别日志
"""


import requests
import logging
from datetime import datetime, timedelta
from collections import defaultdict


logger = logging.getLogger(__name__)


class PumpStateMonitor:
    """
    泵状态监控器
    
    功能:
        - 查询泵运行状态
        - 记录状态变化历史
        - 判断是否处于启停过渡期
    """
    
    def __init__(self, scada_url, scada_jwt, project_id, 
                 timeout=10, transition_window_minutes=15):
        """
        初始化泵状态监控器
        
        参数:
            scada_url: SCADA API地址
            scada_jwt: JWT认证Token
            project_id: 项目ID (用于API查询)
            timeout: 请求超时秒数
            transition_window_minutes: 启停过渡期窗口(默认 15分钟)
        """
        self.scada_url = scada_url
        self.scada_jwt = scada_jwt
        self.project_id = project_id
        self.timeout = timeout
        self.transition_window_minutes = transition_window_minutes
        
        # 状态缓存: {pump_id: True/False}
        self.current_states = {}
        
        # 最后状态变化时间缓存: {pump_id: datetime}
        # 注意: 由本地记录，不依赖接口返回的 htime（那是数据采集时间，不是状态变化时间）
        self.state_change_time = {}
        
        # 上次查询时间（避免频繁查询）
        self.last_query_time = {}
        self.min_query_interval_seconds = 30  # 30 秒查询一次
    
    def query_pump_status(self, point, pump_name=""):
        """
        查询单个泵的运行状态（使用实时数据接口）
        
        使用 current-data 接口直接获取最新一条数据，无需时间窗口查询。
        
        参数:
            point: 点位标识, 如 "C.M.RO1_GYB@run"
            pump_name: 泵名称, 用于日志
        
        返回:
            (is_running, last_change_time):
                is_running: True=运行中, False=停机
                last_change_time: 最后一次状态变化时间 (来自 htime 字段)
        """
        headers = {
            "Content-Type": "application/json",
            "JWT-TOKEN": self.scada_jwt
        }
        
        # 当前时间戳（毫秒）
        now_ms = int(datetime.now().timestamp() * 1000)
        
        # 请求参数
        params = {"time": now_ms}
        
        # 请求体：使用实时数据接口格式
        request_body = [
            {
                "deviceId": "1",
                "deviceItems": point,
                "deviceName": pump_name or point,
                "project_id": self.project_id
            }
        ]
        
        try:
            response = requests.post(
                self.scada_url,
                params=params,
                json=request_body,
                headers=headers,
                timeout=self.timeout
            )
            
            if response.status_code == 200:
                data = response.json()
                if data.get("code") == 200 and data.get("data"):
                    # 获取第一条数据（实时接口只返回最新一条）
                    latest = data["data"][0]
                    if "val" in latest:
                        val = int(float(latest["val"]))
                        is_running = val == 1
                        
                        # 解析 htime 时间字段（实时接口返回的是北京时间字符串）
                        htime_str = latest.get("htime", "")
                        last_change_time = None
                        if htime_str:
                            try:
                                # 直接解析，不做时区转换（按北京时间处理）
                                last_change_time = datetime.strptime(htime_str, "%Y-%m-%d %H:%M:%S")
                            except ValueError:
                                logger.warning(f"无法解析 htime: {htime_str}")
                        
                        logger.debug(f"泵状态查询: {pump_name or point} = {'运行' if is_running else '停机'} (变化时间={htime_str})")
                        return is_running, last_change_time
                
                # 接口返回成功但无数据
                logger.warning(f"泵状态查询无数据: {pump_name or point}")
            else:
                logger.warning(f"泵状态查询HTTP错误: {pump_name or point} | status={response.status_code}")
                
        except Exception as e:
            logger.warning(f"泵状态查询失败: {pump_name or point} | {e}")
        
        # 查询失败时默认返回停机状态
        logger.warning(f"泵状态查询: {pump_name or point} | 查询失败，默认视为停机")
        return False, None
    
    def update_pump_state(self, pump_id, point, pump_name=""):
        """
        更新并检测泵状态变化
        
        状态变化时间由本地记录，不依赖接口返回的 htime（那是数据采集时间）。
        
        参数:
            pump_id: 泵唯一标识
            point: SCADA点位
            pump_name: 泵名称
        
        返回:
            (当前状态 True/False, 最后状态变化时间)
        """
        # 限制查询频率
        now = datetime.now()
        last_query = self.last_query_time.get(pump_id)
        if last_query:
            elapsed = (now - last_query).total_seconds()
            if elapsed < self.min_query_interval_seconds:
                # 使用缓存状态
                cached_state = self.current_states.get(pump_id)
                cached_time = self.state_change_time.get(pump_id)
                return cached_state, cached_time
        
        # 查询状态（实时接口只返回当前状态，不返回状态变化时间）
        new_state, _ = self.query_pump_status(point, pump_name)
        
        self.last_query_time[pump_id] = now
        
        # 检测状态变化
        old_state = self.current_states.get(pump_id)
        state_change_time = self.state_change_time.get(pump_id)
        
        if old_state is None:
            # 首次查询：初始化缓存状态
            # 假设泵已经稳定运行，不设置过渡期（避免程序启动时所有泵都被认为在过渡期）
            state_change_time = None  # None 表示不在过渡期
            logger.info(f"泵状态初始化: {pump_name or pump_id} = {'运行' if new_state else '停机'}")
        elif new_state != old_state:
            # 状态变化：记录当前时间为变化时间
            state_change_time = now
            event = "启动" if new_state else "停机"
            logger.info(f"泵状态变化: {pump_name or pump_id} {event} | 进入过渡期")
        
        # 更新缓存
        self.current_states[pump_id] = new_state
        self.state_change_time[pump_id] = state_change_time
        
        return new_state, state_change_time
    
    def is_in_transition(self, pump_id):
        """
        检查泵是否处于启停过渡期
        
        基于本地记录的状态变化时间判断。
        
        参数:
            pump_id: 泵唯一标识
        
        返回:
            True=过渡期内(最近N分钟有状态变化), False=稳定状态
        """
        last_change_time = self.state_change_time.get(pump_id)
        if not last_change_time:
            # None 表示首次查询或无状态变化记录，视为稳定状态
            return False
        
        # 使用本地时间比较（变化时间是本地记录的）
        elapsed_minutes = (datetime.now() - last_change_time).total_seconds() / 60
        
        return elapsed_minutes < self.transition_window_minutes
    
    def check_pumps_transition(self, pump_configs):
        """
        检查多个泵是否有任何处于启停过渡期
        
        参数:
            pump_configs: 泵配置列表 [{"point": "...", "name": "..."}, ...]
        
        返回:
            (是否有泵在过渡期, 过渡期泵名称列表)
        """
        in_transition = False
        transition_pumps = []
        
        for pump_cfg in pump_configs:
            point = pump_cfg.get("point", "")
            name = pump_cfg.get("name", point)
            pump_id = point  # 使用点位作为唯一ID
            
            # 更新状态
            self.update_pump_state(pump_id, point, name)
            
            # 检查是否过渡期
            if self.is_in_transition(pump_id):
                in_transition = True
                transition_pumps.append(name)
        
        return in_transition, transition_pumps