46 lines
1.7 KiB
Python
46 lines
1.7 KiB
Python
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""称重终端驱动 —— 化工/树脂行业称重仪参数化适配。
|
|||
|
|
|
|||
|
|
多数称重仪表走串口（连续输出 / 命令应答）或 Modbus TCP；
|
|||
|
|
本驱动作为模板封装：优先按配置走 Modbus（复用 ModbusDriver），
|
|||
|
|
若配置指定 `mode: serial` 则走串口协议（需现场协议文档，由子类定制）。
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from typing import Dict, List, Optional
|
|||
|
|
|
|||
|
|
from point_dict.loader import Point
|
|||
|
|
from .base import Driver, SampleValue
|
|||
|
|
from .modbus_driver import ModbusDriver
|
|||
|
|
|
|||
|
|
|
|||
|
|
class WeighingDriver(Driver):
|
|||
|
|
"""称重终端只读采集驱动（参数化适配）。"""
|
|||
|
|
|
|||
|
|
protocol = "weighing"
|
|||
|
|
|
|||
|
|
def __init__(self, config: Optional[dict] = None):
|
|||
|
|
super().__init__(config)
|
|||
|
|
self.mode: str = self.config.get("mode", "modbus") # modbus | serial
|
|||
|
|
self._delegate: Optional[Driver] = None
|
|||
|
|
|
|||
|
|
def connect(self) -> None:
|
|||
|
|
if self.mode == "modbus":
|
|||
|
|
# 称重仪表普遍支持 Modbus RTU/TCP，复用 ModbusDriver 参数化实现
|
|||
|
|
self._delegate = ModbusDriver(self.config)
|
|||
|
|
else:
|
|||
|
|
raise ConnectionError(
|
|||
|
|
"称重驱动 serial 模式需要现场协议文档支持，请实现子类或改用 modbus 模式"
|
|||
|
|
)
|
|||
|
|
self._delegate.connect()
|
|||
|
|
|
|||
|
|
def read_points(self, points: List[Point]) -> Dict[str, SampleValue]:
|
|||
|
|
if self._delegate is None:
|
|||
|
|
raise ConnectionError("称重驱动未连接，请先 connect()")
|
|||
|
|
return self._delegate.read_points(points)
|
|||
|
|
|
|||
|
|
def close(self) -> None:
|
|||
|
|
if self._delegate is not None:
|
|||
|
|
self._delegate.close()
|
|||
|
|
self._delegate = None
|