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()) |