simAGV/plc/modbus.py

import traceback
import click
from modbus_tk import modbus_tcp, hooks


class ModbusConnection:
    '''
    Modbus连接管理器
    '''
    
    def __init__(self, host: str, port: int) -> None:
        self.host = host
        self.port = port
        self.master = None

    def __enter__(self) -> modbus_tcp.TcpMaster:

        def on_after_recv(args):
            response = args[1]
            # print("on_after_recv {0} bytes received".format(len(response)))
            # print("res: ", response)

        hooks.install_hook("modbus_tcp.TcpMaster.after_recv", on_after_recv)

        self.master = modbus_tcp.TcpMaster(self.host, self.port)
        self.master.set_timeout(5.0)
        return self.master

    def __exit__(self, exc_type, exc_value, _traceback) -> None:
        try:
            if self.master:
                self.master.close()

            if exc_type is not None:
                click.echo("发生异常:")
                click.echo(f"类型: {exc_type}")
                click.echo(f"值: {exc_value}")
                click.echo("调用栈:")
                traceback.print_tb(_traceback)
        except Exception as e :
            print('PLC通信异常:::',str(e))