modpypes/modpypes/client.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
Client
======

This executable module is a console application for generating
read and write MODBUS PDUs.
"""

import math

from bacpypes.debugging import class_debugging, ModuleLogger
from bacpypes.consolecmd import ConsoleCmd
from bacpypes.consolelogging import ArgumentParser

from bacpypes.comm import Client, bind
from bacpypes.core import run

from pdu import ExceptionResponse, \
    ReadCoilsRequest, ReadCoilsResponse, \
    ReadDiscreteInputsRequest, ReadDiscreteInputsResponse, \
    ReadMultipleRegistersRequest, ReadMultipleRegistersResponse, \
    ModbusStruct
from app import ModbusClient, ModbusException

# some debugging
_debug = 0
_log = ModuleLogger(globals())


#
#   ConsoleClient
#


@class_debugging
class ConsoleClient(ConsoleCmd, Client):

    """
    Console Client
    """

    def __init__(self):
        if _debug: ConsoleClient._debug("__init__")
        ConsoleCmd.__init__(self)

        # no current request
        self.req = None

    def do_read(self, args):
        """read <addr> <unitID> <register> [ <count> ]

        :param addr: IP address of the MODBUS/TCP device or gateway
        :param unitID: unit identifier
        :param register: register in 5-digit or 6-digit format
        :param count: number of registers to read, defaults to one

        This command generates a :class:`ReadCoilsRequest`,
        :class:`ReadDiscreteInputsRequest` or :class:`ReadMultipleRegistersRequest`
        depending on the address prefix; 1, 5, or 6.
        """
        args = args.split()
        if _debug: ConsoleClient._debug("do_read %r", args)

        if (len(args) < 3):
            print "address, unit and register required"
            return

        # get the address and unit
        addr, unitID, register = args[:3]
        unitID = int(unitID)
        if _debug: ConsoleClient._debug("    - addr, unitID: %r, %r", addr, unitID)

        # get the register and count
        register = int(register)
        if len(args) == 4:
            rcount = int(args[3])
        else:
            rcount = 1
        if _debug: ConsoleClient._debug("    - register, rcount: %r, %r", register, rcount)

        # decode the register into a type
        digits = int(math.log10(register)) + 1
        if digits < 4:
            # must be a coil
            registerType = 0
        elif digits == 5:
            registerType = register // 10000
            register = register % 10000
        elif digits == 6:
            registerType = register // 100000
            register = register % 100000
        else:
            print "5 or 6 digit addresses please"
            return
        if _debug: ConsoleClient._debug("    - registerType, register: %r, %r", registerType, register)

        # build a request
        if registerType == 0:
            # coil
            req = ReadCoilsRequest(register - 1, rcount)
        elif registerType == 1:
            # input discrete
            req = ReadDiscreteInputsRequest(register - 1, 1)
        elif registerType == 4:
            # holding register
            req = ReadMultipleRegistersRequest(register - 1, rcount)
        else:
            print "unsupported register type"
            return

        # set the destination
        req.pduDestination = (addr, 502)
        req.mpduUnitID = unitID
        if _debug: ConsoleClient._debug("    - req: %r", req)

        # save the request
        self.req = req

        # send it along
        self.request(req)

    def confirmation(self, pdu):
        """Prints out the contents of the response from the
        device.
        """

        if _debug: ConsoleClient._debug("confirmation %r", pdu)

        if isinstance(pdu, ExceptionResponse):
            print ModbusException(pdu.exceptionCode)

        elif isinstance(pdu, ReadCoilsResponse):
            print "  ::=", pdu.bits

        elif isinstance(pdu, ReadDiscreteInputsResponse):
            print "  ::=", pdu.bits

        elif isinstance(pdu, ReadMultipleRegistersResponse):
            print "  ::=", pdu.registers

            for dtype, codec in ModbusStruct.items():
                try:
                    value = codec.unpack(pdu.registers)
                    print "   ", dtype, "::=", value
                except Exception as err:
                    if _debug: ConsoleClient._debug("unpack exception %r: %r", codec, err)

        else:
            raise TypeError("unsupported response")


#
#   main
#


def main():
    try:
        # parse the command line arguments
        parser = ArgumentParser(description=__doc__)

        # now parse the arguments
        args = parser.parse_args()

        if _debug: _log.debug("initialization")
        if _debug: _log.debug("    - args: %r", args)

        # local IO functions
        bind(ConsoleClient(), ModbusClient())

        _log.debug("running")

        run()

    except Exception, e:
        _log.exception("an error has occurred: %s", e)
    finally:
        _log.debug("finally")

if __name__ == "__main__":
    main()