bot_dev1
f8f6d977c5
- 连接参数 ip/port/rack/slot/timeout 全部来自模板配置（去硬编码）
- 点位→DB 地址映射经点位字典 opcNode 列携带（DB{db}.{byte}[.{bit}]）
- 按 dataType 解码：float(REAL/4B) int(WORD/2B) bool(位)，S7 大端
- read_area 用 snap7 常量 0x84(DB区)，单点失败容错记 None 计入丢失率
- 严格只读，无任何写/控制指令（PRD 9 章安全约束）
- 新增 test_s7_driver.py 18 个用例（mock snap7），全量通过
113 lines
4.4 KiB
Python
113 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""西门子 S7-1200 驱动 —— 参数化适配（issue #24 落点）。
|
||
|
||
连接参数（ip / port / rack / slot）全部来自模板配置；点位到 DB 地址的
|
||
映射通过点位字典 CSV 的 opcNode 列携带，约定格式：
|
||
|
||
DB{db}.{byte} → 按点位 dataType 读取整字（WORD/DWORD/REAL）
|
||
DB{db}.{byte}.{bit} → 读取位（BOOL）
|
||
|
||
数据类型解析（对齐点位字典 schema.dataType ∈ float/int/bool）：
|
||
bool → 1 字节，取指定位
|
||
int → 2 字节，有符号 WORD（big-endian，S7 默认大端）
|
||
float → 4 字节，IEEE754 单精度 REAL（big-endian）
|
||
|
||
安全约束（PRD 9 章）：仅暴露只读 `read_points()`，无任何写/控制指令。
|
||
依赖 `python-snap7`，未安装时给出明确提示。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import struct
|
||
from typing import Dict, List, Optional
|
||
|
||
from point_dict.loader import Point
|
||
from .base import Driver, SampleValue
|
||
|
||
_S7_NODE_RE = re.compile(r"^DB(\d+)\.(\d+)(?:\.(\d+))?$")
|
||
|
||
# S7 area 标识（snap7 常量）：0x84 = DB 区
|
||
_S7_AREA_DB = 0x84
|
||
|
||
# dataType → (字节数, python struct 格式字符)
|
||
# S7 PLC 默认大端（big-endian），故用 ">" 前缀。
|
||
_TYPE_SPEC = {
|
||
"bool": (1, None), # 位读取，单独处理
|
||
"int": (2, ">h"), # 有符号 16 位整数（WORD/INT）
|
||
"float": (4, ">f"), # IEEE754 32 位单精度（REAL）
|
||
}
|
||
|
||
|
||
class S7Driver(Driver):
|
||
"""西门子 S7-1200 只读采集驱动（参数化适配）。
|
||
|
||
配置项（drivers.s7.*）：
|
||
ip str 必填 PLC IP 地址
|
||
port int 可选 ISO TCP 端口，默认 102
|
||
rack int 可选 机架号，默认 0
|
||
slot int 可选 插槽号，S7-1200 默认 1
|
||
timeout int 可选 连接/读取超时(秒)，默认 10
|
||
"""
|
||
|
||
protocol = "s7"
|
||
|
||
def __init__(self, config: Optional[dict] = None):
|
||
super().__init__(config)
|
||
self.ip: str = self.config.get("ip", "")
|
||
self.port: int = int(self.config.get("port", 102))
|
||
self.rack: int = int(self.config.get("rack", 0))
|
||
self.slot: int = int(self.config.get("slot", 1))
|
||
self.timeout: int = int(self.config.get("timeout", 10))
|
||
self._client = None
|
||
|
||
def connect(self) -> None:
|
||
try:
|
||
import snap7 # type: ignore
|
||
except ImportError as exc: # pragma: no cover - 依赖缺失路径
|
||
raise ConnectionError(
|
||
"S7 驱动依赖库未安装：请 `pip install python-snap7`"
|
||
) from exc
|
||
if not self.ip:
|
||
raise ConnectionError("S7 驱动缺少配置: drivers.s7.ip")
|
||
self._client = snap7.client.Client()
|
||
# python-snap7 支持 (host, rack, slot) 或 (host, rack, slot, tcp_port)
|
||
try:
|
||
self._client.connect(self.ip, self.rack, self.slot, self.port)
|
||
except TypeError:
|
||
# 旧版本签名不支持 port 参数，回退三参数
|
||
self._client.connect(self.ip, self.rack, self.slot)
|
||
|
||
def read_points(self, points: List[Point]) -> Dict[str, SampleValue]:
|
||
if self._client is None:
|
||
raise ConnectionError("S7 客户端未连接，请先 connect()")
|
||
result: Dict[str, SampleValue] = {}
|
||
for p in points:
|
||
match = _S7_NODE_RE.match(p.opc_node or "")
|
||
if not match:
|
||
continue # 非 S7 点位由其它驱动采集
|
||
db = int(match.group(1))
|
||
byte_offset = int(match.group(2))
|
||
bit = match.group(3)
|
||
try:
|
||
if bit is not None:
|
||
# BOOL：读 1 字节，取指定位
|
||
raw = self._client.read_area(_S7_AREA_DB, db, byte_offset, 1)
|
||
result[p.point_id] = bool(raw[0] & (1 << int(bit)))
|
||
else:
|
||
dtype = (p.data_type or "float").lower()
|
||
spec = _TYPE_SPEC.get(dtype, _TYPE_SPEC["float"])
|
||
size, fmt = spec
|
||
raw = self._client.read_area(_S7_AREA_DB, db, byte_offset, size)
|
||
result[p.point_id] = struct.unpack(fmt, bytes(raw[:size]))[0]
|
||
except Exception:
|
||
# 单点失败不中断整批；由引擎按“未读到”计入丢失率
|
||
result[p.point_id] = None
|
||
return result
|
||
|
||
def close(self) -> None:
|
||
if self._client is not None:
|
||
try:
|
||
self._client.disconnect()
|
||
finally:
|
||
self._client = None
|