python实现Ethernet/IP协议的客户端（三）

news2024/11/26 21:25:23

Ethernet/IP是一种工业自动化领域中常用的网络通信协议，它是基于标准以太网技术的应用层协议。作为工业领域的通信协议之一，Ethernet/IP 提供了一种在工业自动化设备之间实现通信和数据交换的标准化方法。python要实现Ethernet/IP的客户端，可以采用pycomm3模块，但不一定是pycomm3，其中原委咋们慢慢聊，本文为第三篇。

一、关于CIPDriver模块

1、官方例子

Generic Messaging - pycomm3 1.2.14 documentation

在官方例子中，出现率最高的方法是 generic_message ，我们再去看看他的API。

2、API

CIPDriver API - pycomm3 1.2.14 documentation

在这个文档中，我们惊讶地发现他的主要方法也只是generic_message，这让人感觉不太可能实现第一篇文章的交互。

3、还是先用吧

明知这个类大概率实现不了，为什么还要继续呢？

我们在实现一个协议时，最起码先了解这个协议吧？在第一篇文章中，我们了解的情况是阅读了一些官方文档，也了解了EDS文件，但没有一个具体的实验。因此，继续是有必要的，至少能了解协议本身。

4、generic_message方法

对于官方例子，我们先结合源码描述简单分析：

（1）源码描述

    def generic_message(
        self,
        service: Union[int, bytes],
        class_code: Union[int, bytes],
        instance: Union[int, bytes],
        attribute: Union[int, bytes] = b"",
        request_data: Any = b"",
        data_type: Optional[Union[Type[DataType], DataType]] = None,
        name: str = "generic",
        connected: bool = True,
        unconnected_send: bool = False,
        route_path: Union[bool, Sequence[CIPSegment], bytes, str] = True,
        **kwargs,
    ) -> Tag:
        """
        Perform a generic CIP message.  Similar to how MSG instructions work in Logix.

        :param service: service code for the request (single byte)
        :param class_code: request object class ID
        :param instance: ID for an instance of the class
                         If set with 0, request class attributes.
        :param attribute: (optional) attribute ID for the service/class/instance
        :param request_data: (optional) any additional data required for the request.
        :param data_type: a ``DataType`` class that will be used to decode the response, None to return just bytes
        :param name:  return ``Tag.tag`` value, arbitrary but can be used for tracking returned Tags
        :param connected: ``True`` if service required a CIP connection (forward open), ``False`` to use UCMM
        :param unconnected_send: (Unconnected Only) wrap service in an UnconnectedSend service
        :param route_path: (Unconnected Only) ``True`` to use current connection route to destination, ``False`` to ignore,
                           Or provide a path string, list of segments to be encoded as a PADDED_EPATH, or
                           an already encoded path.
        :return: a Tag with the result of the request. (Tag.value for writes will be the request_data)
        """

（2）简单分析

官方文档直接给了一个函数，我做了一些引入的补充，即“from pycomm3 import CIPDriver, Services, INT” ：

from pycomm3 import CIPDriver, Services, INT

def read_pf525_parameter():
    drive_path = '10.10.10.100/bp/1/enet/192.168.1.55'

    with CIPDriver(drive_path) as drive:
        param = drive.generic_message(
            service=Services.get_attribute_single,
            class_code=b'\x93',
            instance=41,  # Parameter 41 = Accel Time
            attribute=b'\x09',
            data_type=INT,
            connected=False,
            unconnected_send=True,
            route_path=True,
            name='pf525_param'
        )
        print(param)

对于 drive_path = '10.10.10.100/bp/1/enet/192.168.1.55' 这个语句，再结合上次的实验，我猜测EIP设备是10.10.10.100，并且该EIP设备是有一些子模块的，192.168.1.55便是内部的子模块地址。中间的/bp/1/enet则是按照EIP协议规定的路径。

在使用 with 语句创建了一个 CIPDriver 的实例（即自动调用了open()方法），确保在代码块结束时正确关闭连接后，接着调用了generic_message方法发送了一个通用的消息。具体参数如下：

service=Services.get_attribute_single: 使用 EtherNet/IP 协议中的 get_attribute_single 服务。
class_code=b'\x93': 类别代码，可能是特定设备或对象类型的标识。
instance=41: 代表要访问的具体实例，这里是 Parameter 41。
attribute=b'\x09': 属性标识，可能代表具体要获取的信息的类型。
data_type=INT: 数据类型，这里是整数。
connected=False: 表示使用不连接的方式发送消息。
unconnected_send=True: 允许未连接发送，可能用于广播或多播。
route_path=True: 表示要使用路由路径（如果需要）。
name='pf525_param': 为此消息指定了一个名称，这个名称在返回的结果中可以用于标识。

好了，通过上述解释其实还有很多不理解的地方，我们继续实验来进一步了解。

二、继续实验

1、请求看看

修改代码让其请求成功，提示如下：

pf525_param, None, INT, IOI syntax error. A syntax error was detected decoding the Request Path (see extended status) - Extended status out of memory (04, 00)

表明没有异常，只是服务器没有这些扩展状态，说明我们请求的资源不存在。

from pycomm3 import CIPDriver, Services, INT

def read_pf525_parameter():
    drive_path = '192.168.1.189'

    with CIPDriver(drive_path) as drive:
        param = drive.generic_message(
            service=Services.get_attribute_single,
            class_code=b'\x93',
            instance=41,
            attribute=b'\x09',
            data_type=INT,
            connected=False,
            unconnected_send=True,
            route_path=True,
            name='pf525_param'
        )
        print(param)

read_pf525_parameter()

查看了报文依旧在报路径错误。