223 lines
7.8 KiB
Python
223 lines
7.8 KiB
Python
|
# metric_device.py
|
|||
|
|
import socket
|
|||
|
|
import threading
|
|||
|
|
import time
|
|||
|
|
import statistics
|
|||
|
|
from abc import ABC, abstractmethod
|
|||
|
|
from typing import Optional, Dict, Any, List
|
|||
|
|
from dataclasses import dataclass, asdict
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class DeviceReading:
|
|||
|
|
"""标准化设备读数"""
|
|||
|
|
weight: Optional[int] # 重量（None 表示 OL/ER）
|
|||
|
|
raw_data: bytes # 原始字节
|
|||
|
|
status: str # 'ST', 'US', 'OL', 'ER'
|
|||
|
|
timestamp: float # 本地接收时间（秒，time.time()）
|
|||
|
|
rtt_ms: float # 网络往返延迟（毫秒）
|
|||
|
|
is_valid: bool # 是否含有效重量
|
|||
|
|
|
|||
|
|
def to_dict(self) -> Dict[str, Any]:
|
|||
|
|
return {
|
|||
|
|
**asdict(self),
|
|||
|
|
"timestamp_iso": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(self.timestamp)),
|
|||
|
|
"weight_kg": self.weight / 1000.0 if self.weight is not None else None,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
class MetricDevice(ABC):
|
|||
|
|
"""抽象计量设备基类 —— 支持 get() + RTT 统计"""
|
|||
|
|
def __init__(self, ip: str, port: int, name: str = ""):
|
|||
|
|
self.ip = ip
|
|||
|
|
self.port = port
|
|||
|
|
self.name = name or f"{ip}:{port}"
|
|||
|
|
self._latest_reading: Optional[DeviceReading] = None
|
|||
|
|
self._lock = threading.Lock()
|
|||
|
|
|
|||
|
|
# 📊 RTT 统计（滑动窗口）
|
|||
|
|
self._rtt_window: List[float] = []
|
|||
|
|
self._rtt_max_len = 100 # 保留最近 100 次 RTT
|
|||
|
|
self._rtt_stats = {
|
|||
|
|
"min_ms": float('inf'),
|
|||
|
|
"max_ms": 0.0,
|
|||
|
|
"avg_ms": 0.0,
|
|||
|
|
"count": 0,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
def _update_rtt(self, rtt_ms: float):
|
|||
|
|
"""更新 RTT 统计"""
|
|||
|
|
self._rtt_window.append(rtt_ms)
|
|||
|
|
if len(self._rtt_window) > self._rtt_max_len:
|
|||
|
|
self._rtt_window.pop(0)
|
|||
|
|
|
|||
|
|
rtt_list = self._rtt_window
|
|||
|
|
self._rtt_stats.update({
|
|||
|
|
"min_ms": min(rtt_list) if rtt_list else 0,
|
|||
|
|
"max_ms": max(rtt_list) if rtt_list else 0,
|
|||
|
|
"avg_ms": statistics.mean(rtt_list) if rtt_list else 0,
|
|||
|
|
"count": len(rtt_list),
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
def get_rtt_stats(self) -> Dict[str, float]:
|
|||
|
|
"""获取 RTT 统计信息"""
|
|||
|
|
with self._lock:
|
|||
|
|
return self._rtt_stats.copy()
|
|||
|
|
|
|||
|
|
def get(self) -> Optional[Dict[str, Any]]:
|
|||
|
|
"""✅ 统一 get() 接口：返回最新读数（含 RTT）"""
|
|||
|
|
with self._lock:
|
|||
|
|
if self._latest_reading is None:
|
|||
|
|
return None
|
|||
|
|
return self._latest_reading.to_dict()
|
|||
|
|
|
|||
|
|
@abstractmethod
|
|||
|
|
def _connect_and_run(self):
|
|||
|
|
"""子类实现连接与数据接收循环"""
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
def start(self):
|
|||
|
|
"""启动后台线程"""
|
|||
|
|
thread = threading.Thread(target=self._connect_and_run, daemon=True, name=f"Dev-{self.name}")
|
|||
|
|
thread.start()
|
|||
|
|
|
|||
|
|
def _update_reading(self, reading: DeviceReading):
|
|||
|
|
"""线程安全更新最新读数"""
|
|||
|
|
with self._lock:
|
|||
|
|
self._latest_reading = reading
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TCPScaleDevice(MetricDevice):
|
|||
|
|
"""TCP ASCII 协议称重设备（支持您的两种设备）"""
|
|||
|
|
def __init__(self, ip: str, port: int, name: str = ""):
|
|||
|
|
super().__init__(ip, port, name)
|
|||
|
|
self._buffer = b""
|
|||
|
|
self._socket: Optional[socket.socket] = None
|
|||
|
|
self._running = True
|
|||
|
|
|
|||
|
|
def _parse_line(self, line: bytes) -> Optional[DeviceReading]:
|
|||
|
|
try:
|
|||
|
|
clean = line.strip()
|
|||
|
|
if not clean:
|
|||
|
|
return None
|
|||
|
|
parts = clean.split(b',')
|
|||
|
|
if len(parts) < 3:
|
|||
|
|
return DeviceReading(
|
|||
|
|
weight=None, raw_data=clean, status="??",
|
|||
|
|
timestamp=time.time(), rtt_ms=0.0, is_valid=False
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
status = parts[0].decode('ascii', errors='replace').upper()
|
|||
|
|
mode = parts[1].decode('ascii', errors='replace').upper()
|
|||
|
|
weight_str = parts[2].decode('ascii', errors='replace')
|
|||
|
|
|
|||
|
|
weight = None
|
|||
|
|
is_valid = False
|
|||
|
|
|
|||
|
|
if weight_str.replace('+', '').replace('-', '').isdigit():
|
|||
|
|
try:
|
|||
|
|
weight = int(weight_str)
|
|||
|
|
is_valid = True
|
|||
|
|
except ValueError:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
return DeviceReading(
|
|||
|
|
weight=weight,
|
|||
|
|
raw_data=clean,
|
|||
|
|
status=status,
|
|||
|
|
timestamp=time.time(),
|
|||
|
|
rtt_ms=0.0, # 稍后更新
|
|||
|
|
is_valid=is_valid,
|
|||
|
|
)
|
|||
|
|
except Exception:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def _extract_lines(self, data: bytes) -> List[bytes]:
|
|||
|
|
"""自适应行分割"""
|
|||
|
|
self._buffer += data
|
|||
|
|
lines = []
|
|||
|
|
if b'\r\n' in self._buffer or b'\n' in self._buffer:
|
|||
|
|
norm = self._buffer.replace(b'\r\n', b'\n')
|
|||
|
|
parts = norm.split(b'\n')
|
|||
|
|
*complete, self._buffer = parts
|
|||
|
|
lines = [line for line in complete if line]
|
|||
|
|
else:
|
|||
|
|
# 无换行符：整包作为单行
|
|||
|
|
if self._buffer:
|
|||
|
|
lines = [self._buffer]
|
|||
|
|
self._buffer = b""
|
|||
|
|
return lines
|
|||
|
|
|
|||
|
|
def _connect_and_run(self):
|
|||
|
|
while self._running:
|
|||
|
|
try:
|
|||
|
|
# 🔁 重连逻辑
|
|||
|
|
if self._socket is None:
|
|||
|
|
self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|||
|
|
self._socket.settimeout(5.0)
|
|||
|
|
|
|||
|
|
# ⏱️ 测量连接 RTT
|
|||
|
|
t0 = time.perf_counter()
|
|||
|
|
self._socket.connect((self.ip, self.port))
|
|||
|
|
rtt_connect = (time.perf_counter() - t0) * 1000 # ms
|
|||
|
|
self._update_rtt(rtt_connect)
|
|||
|
|
print(f"[{self.name}] Connected (RTT={rtt_connect:.1f}ms)")
|
|||
|
|
|
|||
|
|
# 接收数据
|
|||
|
|
t0 = time.perf_counter()
|
|||
|
|
data = self._socket.recv(1024)
|
|||
|
|
recv_rtt = (time.perf_counter() - t0) * 1000
|
|||
|
|
|
|||
|
|
if not data:
|
|||
|
|
break
|
|||
|
|
|
|||
|
|
# 更新 RTT（粗略：recv 耗时 ≈ 半 RTT）
|
|||
|
|
self._update_rtt(recv_rtt * 2)
|
|||
|
|
|
|||
|
|
# 解析
|
|||
|
|
lines = self._extract_lines(data)
|
|||
|
|
for line in lines:
|
|||
|
|
reading = self._parse_line(line)
|
|||
|
|
if reading:
|
|||
|
|
reading.rtt_ms = recv_rtt * 2 # 近似 RTT
|
|||
|
|
self._update_reading(reading)
|
|||
|
|
|
|||
|
|
except socket.timeout:
|
|||
|
|
continue
|
|||
|
|
except (OSError, socket.error) as e:
|
|||
|
|
print(f"[{self.name}] Error: {e}")
|
|||
|
|
if self._socket:
|
|||
|
|
self._socket.close()
|
|||
|
|
self._socket = None
|
|||
|
|
time.sleep(2)
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"[{self.name}] Fatal: {e}")
|
|||
|
|
break
|
|||
|
|
|
|||
|
|
if self._socket:
|
|||
|
|
self._socket.close()
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ===== 设计模式：设备工厂 + 管理器 =====
|
|||
|
|
class DeviceManager:
|
|||
|
|
"""设备工厂 + 单例管理（避免重复创建同IP:Port设备）"""
|
|||
|
|
_instances: Dict[str, MetricDevice] = {}
|
|||
|
|
_lock = threading.Lock()
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
def get_device(cls, ip: str, port: int, name: str = "") -> MetricDevice:
|
|||
|
|
"""✅ 工厂方法：按 (ip, port) 单例返回设备"""
|
|||
|
|
key = f"{ip}:{port}"
|
|||
|
|
with cls._lock:
|
|||
|
|
if key not in cls._instances:
|
|||
|
|
# 🔧 策略模式：未来可扩展 ModbusDevice、HTTPDevice 等
|
|||
|
|
if port in (502, 8234): # 您的设备端口
|
|||
|
|
device = TCPScaleDevice(ip, port, name)
|
|||
|
|
device.start()
|
|||
|
|
cls._instances[key] = device
|
|||
|
|
else:
|
|||
|
|
raise ValueError(f"Unsupported port {port} for auto-detect")
|
|||
|
|
return cls._instances[key]
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
def get_all_devices(cls) -> List[MetricDevice]:
|
|||
|
|
return list(cls._instances.values())
|