""" S7 PLC通信模块 """ import logging import snap7 from snap7.type import * from snap7.util import * from typing import Optional, Any from datetime import datetime # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(funcName)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class VWEDS7Module: """ S7 PLC通信模块类 """ def __init__(self, script_id: str): self.script_id = script_id logger.debug(f"初始化 S7 模块，脚本ID: {script_id}") def _get_rack_slot(self, plc_type: str) -> tuple: """ 根据PLC类型获取默认的rack和slot值 Args: plc_type (str): PLC类型 Returns: tuple: (rack, slot) 默认值 """ # 根据常见默认值设置 # S7-200 SMART: rack=0, slot=1 # S7-300: rack=0, slot=2 # S7-400: rack=0, slot=3 # S7-1200: rack=0, slot=1 # S7-1500: rack=0, slot=1 # S7-200: rack=0, slot=1 defaults = { 'S200Smart': (0, 1), 'S300': (0, 2), 'S400': (0, 3), 'S1200': (0, 1), 'S1500': (0, 1), 'S200': (0, 1) } return defaults.get(plc_type, (0, 1)) def _parse_address(self, block_and_offset: str) -> tuple: """ 解析地址字符串，返回区域、DB号、字节地址和位地址 Args: block_and_offset (str): 地址字符串，如M100, DB1.100, DB1.100.7等 Returns: tuple: (area, db_number, byte_addr, bit_addr) """ area = None db_number = 0 byte_addr = 0 bit_addr = None # 移除空格 block_and_offset = block_and_offset.replace(' ', '') # 处理DB块地址 if block_and_offset.startswith('DB'): parts = block_and_offset[2:].split('.') if len(parts) >= 2: db_number = int(parts[0]) byte_addr = int(parts[1]) if len(parts) == 3: bit_addr = int(parts[2]) area = Areas.DB # 处理其他区域地址 else: if block_and_offset[0] == 'M': area = Areas.MK addr_part = block_and_offset[1:] elif block_and_offset[0] == 'I': area = Areas.PE addr_part = block_and_offset[1:] elif block_and_offset[0] == 'Q': area = Areas.PA addr_part = block_and_offset[1:] elif block_and_offset[0] == 'V': # V区通常映射到M区 area = Areas.MK addr_part = block_and_offset[1:] elif block_and_offset[0] == 'T': area = Areas.TM addr_part = block_and_offset[1:] elif block_and_offset[0] == 'C': area = Areas.CT addr_part = block_and_offset[1:] elif block_and_offset[0] == 'AI': area = Areas.AI addr_part = block_and_offset[2:] elif block_and_offset[0] == 'AQ': area = Areas.AQ addr_part = block_and_offset[2:] else: raise ValueError(f"不支持的地址区域: {block_and_offset[0]}") # 处理地址中的位信息 if '.' in addr_part: parts = addr_part.split('.') byte_addr = int(parts[0]) bit_addr = int(parts[1]) else: byte_addr = int(addr_part) return area, db_number, byte_addr, bit_addr def read_s7_int(self, plc_type: str, ip: str, block_and_offset: str) -> Optional[int]: """ S7 读取 Int。 Args: plc_type (str): PLC 类型，可选值：S1200/S300/S400/S1500/S200Smart/S200 ip (str): PLC IP地址 block_and_offset (str): 读取的地址，如M100, DB1.100等 Returns: Optional[int]: 读取到的整数值，失败则返回None """ try: # 获取rack和slot rack, slot = self._get_rack_slot(plc_type) # 解析地址 area, db_number, byte_addr, bit_addr = self._parse_address(block_and_offset) # 创建客户端并连接 client = snap7.client.Client() client.connect(ip, rack, slot, 1102) # 读取数据 if area == Areas.DB: data = client.db_read(db_number, byte_addr, 2) # Int是2个字节 else: data = client.read_area(area, 0, byte_addr, 2) # 断开连接 client.disconnect() # 解析数据 value = snap7.util.get_int(data, 0) logger.info(f"成功读取 {plc_type} {ip} 的 {block_and_offset} 地址的Int值: {value}") return value except Exception as e: logger.error(f"{datetime.now()} - VWEDS7Module - read_s7_int - 读取Int值时发生错误: {e}") return None def read_s7_dint(self, plc_type: str, ip: str, block_and_offset: str) -> Optional[int]: """ S7 读取 DInt。 Args: plc_type (str): PLC 类型，可选值：S1200/S300/S400/S1500/S200Smart/S200 ip (str): PLC IP地址 block_and_offset (str): 读取的地址，如M100, DB1.100等 Returns: Optional[int]: 读取到的双整数值，失败则返回None """ try: # 获取rack和slot rack, slot = self._get_rack_slot(plc_type) # 解析地址 area, db_number, byte_addr, bit_addr = self._parse_address(block_and_offset) # 创建客户端并连接 client = snap7.client.Client() client.connect(ip, rack, slot, 1102) # 读取数据 if area == Areas.DB: data = client.db_read(db_number, byte_addr, 4) # DInt是4个字节 else: data = client.read_area(area, 0, byte_addr, 4) # 断开连接 client.disconnect() # 解析数据 value = snap7.util.get_dint(data, 0) logger.info(f"成功读取 {plc_type} {ip} 的 {block_and_offset} 地址的DInt值: {value}") return value except Exception as e: logger.error(f"{datetime.now()} - VWEDS7Module - read_s7_dint - 读取DInt值时发生错误: {e}") return None def read_s7_word(self, plc_type: str, ip: str, block_and_offset: str) -> Optional[int]: """ S7 读取 Word。 Args: plc_type (str): PLC 类型，可选值：S1200/S300/S400/S1500/S200Smart/S200 ip (str): PLC IP地址 block_and_offset (str): 读取的地址，如M100, DB1.100等 Returns: Optional[int]: 读取到的字值，失败则返回None """ try: # 获取rack和slot rack, slot = self._get_rack_slot(plc_type) # 解析地址 area, db_number, byte_addr, bit_addr = self._parse_address(block_and_offset) # 创建客户端并连接 client = snap7.client.Client() client.connect(ip, rack, slot, 1102) # 读取数据 if area == Areas.DB: data = client.db_read(db_number, byte_addr, 2) # Word是2个字节 else: data = client.read_area(area, 0, byte_addr, 2) # 断开连接 client.disconnect() # 解析数据 value = snap7.util.get_word(data, 0) logger.info(f"成功读取 {plc_type} {ip} 的 {block_and_offset} 地址的Word值: {value}") return value except Exception as e: logger.error(f"{datetime.now()} - VWEDS7Module - read_s7_word - 读取Word值时发生错误: {e}") return None def read_s7_dword(self, plc_type: str, ip: str, block_and_offset: str) -> Optional[int]: """ S7 读取 DWord。 Args: plc_type (str): PLC 类型，可选值：S1200/S300/S400/S1500/S200Smart/S200 ip (str): PLC IP地址 block_and_offset (str): 读取的地址，如M100, DB1.100等 Returns: Optional[int]: 读取到的双字值，失败则返回None """ try: # 获取rack和slot rack, slot = self._get_rack_slot(plc_type) # 解析地址 area, db_number, byte_addr, bit_addr = self._parse_address(block_and_offset) # 创建客户端并连接 client = snap7.client.Client() client.connect(ip, rack, slot, 1102) # 读取数据 if area == Areas.DB: data = client.db_read(db_number, byte_addr, 4) # DWord是4个字节 else: data = client.read_area(area, 0, byte_addr, 4) # 断开连接 client.disconnect() # 解析数据 value = snap7.util.get_dword(data, 0) logger.info(f"成功读取 {plc_type} {ip} 的 {block_and_offset} 地址的DWord值: {value}") return value except Exception as e: logger.error(f"{datetime.now()} - VWEDS7Module - read_s7_dword - 读取DWord值时发生错误: {e}") return None def read_s7_string(self, plc_type: str, ip: str, block_and_offset: str) -> Optional[str]: """ S7 读取 String。 Args: plc_type (str): PLC 类型，可选值：S1200/S300/S400/S1500/S200Smart/S200 ip (str): PLC IP地址 block_and_offset (str): 读取的地址，如M100, DB1.100等 Returns: Optional[str]: 读取到的字符串值，失败则返回None """ try: # 获取rack和slot rack, slot = self._get_rack_slot(plc_type) # 解析地址 area, db_number, byte_addr, bit_addr = self._parse_address(block_and_offset) # 创建客户端并连接 client = snap7.client.Client() client.connect(ip, rack, slot, 1102) # 先读取字符串的长度信息（前2个字节） # 第一个字节是最大长度，第二个字节是实际长度 if area == Areas.DB: length_data = client.db_read(db_number, byte_addr, 2) else: length_data = client.read_area(area, 0, byte_addr, 2) # 获取字符串的实际长度 max_length = length_data[0] # 最大长度 actual_length = length_data[1] # 实际长度 # 读取字符串数据（根据实际长度读取） if area == Areas.DB: data = client.db_read(db_number, byte_addr + 2, actual_length) else: data = client.read_area(area, 0, byte_addr + 2, actual_length) # 断开连接 client.disconnect() # 解析数据 value = data.decode('utf-8').rstrip('\x00').rstrip() # 移除可能的空字符和空格 logger.info(f"成功读取 {plc_type} {ip} 的 {block_and_offset} 地址的String值: {value}") return value except Exception as e: logger.error(f"{datetime.now()} - VWEDS7Module - read_s7_string - 读取String值时发生错误: {e}") return None def read_s7_boolean(self, plc_type: str, ip: str, block_and_offset: str) -> Optional[bool]: """ S7 读取 Bool。 Args: plc_type (str): PLC 类型，可选值：S1200/S300/S400/S1500/S200Smart/S200 ip (str): PLC IP地址 block_and_offset (str): 读取的地址，如M100.0, DB1.100.7等 Returns: Optional[bool]: 读取到的布尔值，失败则返回None """ try: # 获取rack和slot rack, slot = self._get_rack_slot(plc_type) # 解析地址 area, db_number, byte_addr, bit_addr = self._parse_address(block_and_offset) # 检查是否有位地址 if bit_addr is None: raise ValueError("读取布尔值需要指定位地址，如M100.0") # 创建客户端并连接 client = snap7.client.Client() client.connect(ip, rack, slot, 1102) # 读取数据 if area == Areas.DB: data = client.db_read(db_number, byte_addr, 1) # 读取1个字节 else: data = client.read_area(area, 0, byte_addr, 1) # 断开连接 client.disconnect() # 解析数据 value = snap7.util.get_bool(data, 0, bit_addr) logger.info(f"成功读取 {plc_type} {ip} 的 {block_and_offset} 地址的Boolean值: {value}") return value except Exception as e: logger.error(f"{datetime.now()} - VWEDS7Module - read_s7_boolean - 读取Boolean值时发生错误: {e}") return None def write_s7_int(self, plc_type: str, ip: str, block_and_offset: str, value: int) -> bool: """ S7 写入 Int。 Args: plc_type (str): PLC 类型，可选值：S1200/S300/S400/S1500/S200Smart/S200 ip (str): PLC IP地址 block_and_offset (str): 写入的地址，如M100, DB1.100等 value (int): 要写入的整数值 Returns: bool: 写入成功返回True，失败返回False """ try: # 获取rack和slot rack, slot = self._get_rack_slot(plc_type) # 解析地址 area, db_number, byte_addr, bit_addr = self._parse_address(block_and_offset) # 创建客户端并连接 client = snap7.client.Client() client.connect(ip, rack, slot, 1102) # 准备数据 data = bytearray(2) snap7.util.set_word(data, 0, value) # 写入数据 logger.info(f"准备写入数据到DB{db_number}, 偏移量{byte_addr}, 数据: {data}") if area == Areas.DB: client.db_write(db_number, byte_addr, data) else: client.write_area(area, 0, byte_addr, data) logger.info(f"数据写入完成") # 断开连接 client.disconnect() logger.info(f"成功向 {plc_type} {ip} 的 {block_and_offset} 地址写入Int值 {value}") return True except Exception as e: logger.error(f"{datetime.now()} - VWEDS7Module - write_s7_int - 写入Int值时发生错误: {e}") return False def write_s7_dint(self, plc_type: str, ip: str, block_and_offset: str, value: int) -> bool: """ S7 写入 DInt。 Args: plc_type (str): PLC 类型，可选值：S1200/S300/S400/S1500/S200Smart/S200 ip (str): PLC IP地址 block_and_offset (str): 写入的地址，如M100, DB1.100等 value (int): 要写入的双整数值 Returns: bool: 写入成功返回True，失败返回False """ try: # 获取rack和slot rack, slot = self._get_rack_slot(plc_type) # 解析地址 area, db_number, byte_addr, bit_addr = self._parse_address(block_and_offset) # 创建客户端并连接 client = snap7.client.Client() client.connect(ip, rack, slot, 1102) # 准备数据 data = bytearray(4) snap7.util.set_dint(data, 0, value) # 写入数据 if area == Areas.DB: client.db_write(db_number, byte_addr, data) else: client.write_area(area, 0, byte_addr, data) # 断开连接 client.disconnect() logger.info(f"成功向 {plc_type} {ip} 的 {block_and_offset} 地址写入DInt值 {value}") return True except Exception as e: logger.error(f"{datetime.now()} - VWEDS7Module - write_s7_dint - 写入DInt值时发生错误: {e}") return False def write_s7_word(self, plc_type: str, ip: str, block_and_offset: str, value: int) -> bool: """ S7 写入 Word。 Args: plc_type (str): PLC 类型，可选值：S1200/S300/S400/S1500/S200Smart/S200 ip (str): PLC IP地址 block_and_offset (str): 写入的地址，如M100, DB1.100等 value (int): 要写入的字值 Returns: bool: 写入成功返回True，失败返回False """ try: # 获取rack和slot rack, slot = self._get_rack_slot(plc_type) # 解析地址 area, db_number, byte_addr, bit_addr = self._parse_address(block_and_offset) # 创建客户端并连接 client = snap7.client.Client() client.connect(ip, rack, slot, 1102) # 准备数据 data = bytearray(2) snap7.util.set_word(data, 0, value) # 写入数据 if area == Areas.DB: client.db_write(db_number, byte_addr, data) else: client.write_area(area, 0, byte_addr, data) # 断开连接 client.disconnect() logger.info(f"成功向 {plc_type} {ip} 的 {block_and_offset} 地址写入Word值 {value}") return True except Exception as e: logger.error(f"{datetime.now()} - VWEDS7Module - write_s7_word - 写入Word值时发生错误: {e}") return False def write_s7_dword(self, plc_type: str, ip: str, block_and_offset: str, value: int) -> bool: """ S7 写入 DWord。 Args: plc_type (str): PLC 类型，可选值：S1200/S300/S400/S1500/S200Smart/S200 ip (str): PLC IP地址 block_and_offset (str): 写入的地址，如M100, DB1.100等 value (int): 要写入的双字值 Returns: bool: 写入成功返回True，失败返回False """ try: # 获取rack和slot rack, slot = self._get_rack_slot(plc_type) # 解析地址 area, db_number, byte_addr, bit_addr = self._parse_address(block_and_offset) # 创建客户端并连接 client = snap7.client.Client() client.connect(ip, rack, slot, 1102) # 准备数据 data = bytearray(4) snap7.util.set_dword(data, 0, value) # 写入数据 if area == Areas.DB: client.db_write(db_number, byte_addr, data) else: client.write_area(area, 0, byte_addr, data) # 断开连接 client.disconnect() logger.info(f"成功向 {plc_type} {ip} 的 {block_and_offset} 地址写入DWord值 {value}") return True except Exception as e: logger.error(f"{datetime.now()} - VWEDS7Module - write_s7_dword - 写入DWord值时发生错误: {e}") return False def write_s7_string(self, plc_type: str, ip: str, block_and_offset: str, value: str) -> bool: """ S7 写入 String。 Args: plc_type (str): PLC 类型，可选值：S1200/S300/S400/S1500/S200Smart/S200 ip (str): PLC IP地址 block_and_offset (str): 写入的地址，如M100, DB1.100等 value (str): 要写入的字符串值 Returns: bool: 写入成功返回True，失败返回False """ try: # 获取rack和slot rack, slot = self._get_rack_slot(plc_type) # 解析地址 area, db_number, byte_addr, bit_addr = self._parse_address(block_and_offset) # 创建客户端并连接 client = snap7.client.Client() client.connect(ip, rack, slot, 1102) # 准备数据 # 字符串格式：前2个字节为长度，后面是字符串内容 data = bytearray(256) # 最大256字节 encoded_value = value.encode('utf-8') length = min(len(encoded_value), 254) # 最大长度254 snap7.util.set_word(data, 0, length) # 设置长度 data[2:2+length] = encoded_value[:length] # 设置字符串内容 # 写入数据 if area == Areas.DB: client.db_write(db_number, byte_addr, data) else: client.write_area(area, 0, byte_addr, data) # 断开连接 client.disconnect() logger.info(f"成功向 {plc_type} {ip} 的 {block_and_offset} 地址写入String值 {value}") return True except Exception as e: logger.error(f"{datetime.now()} - VWEDS7Module - write_s7_string - 写入String值时发生错误: {e}") return False def write_s7_boolean(self, plc_type: str, ip: str, block_and_offset: str, value: bool) -> bool: """ S7 写入 Bool。 Args: plc_type (str): PLC 类型，可选值：S1200/S300/S400/S1500/S200Smart/S200 ip (str): PLC IP地址 block_and_offset (str): 写入的地址，如M100.0, DB1.100.7等 value (bool): 要写入的布尔值 Returns: bool: 写入成功返回True，失败返回False """ try: # 获取rack和slot rack, slot = self._get_rack_slot(plc_type) # 解析地址 area, db_number, byte_addr, bit_addr = self._parse_address(block_and_offset) # 检查是否有位地址 if bit_addr is None: raise ValueError("写入布尔值需要指定位地址，如M100.0") # 创建客户端并连接 client = snap7.client.Client() client.connect(ip, rack, slot, 1102) # 读取当前字节数据 if area == Areas.DB: data = client.db_read(db_number, byte_addr, 1) else: data = client.read_area(area, 0, byte_addr, 1) # 设置布尔值 snap7.util.set_bool(data, 0, bit_addr, value) # 写入数据 if area == Areas.DB: client.db_write(db_number, byte_addr, data) else: client.write_area(area, 0, byte_addr, data) # 断开连接 client.disconnect() logger.info(f"成功向 {plc_type} {ip} 的 {block_and_offset} 地址写入Boolean值 {value}") return True except Exception as e: logger.error(f"{datetime.now()} - VWEDS7Module - write_s7_boolean - 写入Boolean值时发生错误: {e}") return False def write_s7(self, plc_type: str, ip: str, block_and_offset: str, value: bytes, rack: int = 0, slot: int = 1) -> bool: """ S7 通用写入。 Args: plc_type (str): PLC 类型，可选值：S1200/S300/S400/S1500/S200Smart/S200 ip (str): PLC IP地址 block_and_offset (str): 写入的地址，如M100, DB1.100等 value (bytes): 要写入的字节数据 rack (int): 中央机架，默认为0 slot (int): CPU模块插槽号，默认为1 Returns: bool: 写入成功返回True，失败返回False """ try: # 获取rack和slot的默认值 default_rack, default_slot = self._get_rack_slot(plc_type) if rack == 0 and slot == 1: rack, slot = default_rack, default_slot # 解析地址 area, db_number, byte_addr, bit_addr = self._parse_address(block_and_offset) # 创建客户端并连接 client = snap7.client.Client() client.connect(ip, rack, slot, 1102) # 写入数据 if area == Areas.DB: client.db_write(db_number, byte_addr, value) else: client.write_area(area, 0, byte_addr, value) # 断开连接 client.disconnect() logger.info(f"成功向 {plc_type} {ip} 的 {block_and_offset} 地址写入数据 {value}") return True except Exception as e: logger.error(f"{datetime.now()} - VWEDS7Module - write_s7 - 通用写入时发生错误: {e}") return False def read_s7(self, plc_type: str, ip: str, block_and_offset: str, size: int, rack: int = 0, slot: int = 1) -> Optional[bytes]: """ S7 通用读取。 Args: plc_type (str): PLC 类型，可选值：S1200/S300/S400/S1500/S200Smart/S200 ip (str): PLC IP地址 block_and_offset (str): 读取的地址，如M100, DB1.100等 size (int): 读取的数据大小（字节数） rack (int): 中央机架，默认为0 slot (int): CPU模块插槽号，默认为1 Returns: Optional[bytes]: 读取到的字节数据，失败则返回None """ try: # 获取rack和slot的默认值 default_rack, default_slot = self._get_rack_slot(plc_type) if rack == 0 and slot == 1: rack, slot = default_rack, default_slot # 解析地址 area, db_number, byte_addr, bit_addr = self._parse_address(block_and_offset) # 创建客户端并连接 client = snap7.client.Client() client.connect(ip, rack, slot, 1102) # 读取数据 if area == Areas.DB: data = client.db_read(db_number, byte_addr, size) else: data = client.read_area(area, 0, byte_addr, size) # 断开连接 client.disconnect() logger.info(f"成功读取 {plc_type} {ip} 的 {block_and_offset} 地址的 {size} 字节数据") return data except Exception as e: logger.error(f"{datetime.now()} - VWEDS7Module - read_s7 - 通用读取时发生错误: {e}") return None