easymodbus/modbusClient.py
def write_single_register(self, starting_address, value):
        """
        Write single Register to Master device (Function code 6)
        starting_address:  Register to be written
        value: Register Value to be written
        """
        logging.info("Request to write single register (FC06), starting address: {0}, value: {1}"
                     .format(str(starting_address), str(value)))
        self.__transactionIdentifier += 1
        if self.__ser is not None:
            if self.__ser.closed:
                raise Exception.SerialPortNotOpenedException("serial port not opened")
        function_code = 6
        length = 6
        transaction_identifier_lsb = self.__transactionIdentifier & 0xFF
        transaction_identifier_msb = ((self.__transactionIdentifier & 0xFF00) >> 8)
        length_lsb = length & 0xFF
        length_msb = (length & 0xFF00) >> 8
        starting_address_lsb = starting_address & 0xFF
        starting_address_msb = (starting_address & 0xFF00) >> 8
        valueLSB = value & 0xFF
        valueMSB = (value & 0xFF00) >> 8
        if self.__ser is not None:
            data = bytearray(
                [self.__unitIdentifier, function_code, starting_address_msb, starting_address_lsb, valueMSB, valueLSB,
                 0, 0])
            crc = self.__calculateCRC(data, len(data) - 2, 0)
            crcLSB = crc & 0xFF
            crcMSB = (crc & 0xFF00) >> 8
            data[6] = crcLSB
            data[7] = crcMSB
            logging.debug("---------Transmit: {0}"
                          .format(str(data.hex(' '))))
            self.__ser.write(data)
            bytes_to_read = 8
            data = None
            data = self.__ser.read(bytes_to_read)
            b = bytearray(data)
            data = b
            logging.debug("---------Receive: {0}"
                          .format(str(data.hex(' '))))
            # Check for Exception
            if len(data) < bytes_to_read:
                raise Exceptions.TimeoutError('Read timeout Exception')
            if (data[1] == 0x86) & (data[2] == 0x01):
                raise Exceptions.function_codeNotSupportedException("Function code not supported by master")
            if (data[1] == 0x86) & (data[2] == 0x02):
                raise Exceptions.starting_addressInvalidException("Register address invalid")
            if (data[1] == 0x86) & (data[2] == 0x03):
                raise Exceptions.QuantityInvalidException("Invalid Register Value")
            if (data[1] == 0x86) & (data[2] == 0x04):
                raise Exceptions.ModbusException("error reading")
            crc = self.__calculateCRC(data, len(data) - 2, 0)
            crcLSB = crc & 0xFF
            crcMSB = (crc & 0xFF00) >> 8
            if (crcLSB != data[len(data) - 2]) & (crcMSB != data[len(data) - 1]):
                raise Exceptions.CRCCheckFailedException("CRC check failed")

            if data[1] == self.__unitIdentifier:
                return True
            else:
                return False
        else:
            protocolIdentifierLSB = 0x00
            protocolIdentifierMSB = 0x00
            length_lsb = 0x06
            length_msb = 0x00
            data = bytearray(
                [transaction_identifier_msb, transaction_identifier_lsb, protocolIdentifierMSB, protocolIdentifierLSB,
                 length_msb, length_lsb, self.__unitIdentifier, function_code, starting_address_msb,
                 starting_address_lsb, valueMSB, valueLSB])
            logging.debug("---------Transmit: {0}"
                          .format(str(data.hex(' '))))
            self.__tcpClientSocket.send(data)
            bytes_to_read = 12
            self.__receivedata = bytearray()
            try:
                while (len(self.__receivedata) == 0):
                    pass
            except Exception:
                raise Exception('Read Timeout')
            data = bytearray(self.__receivedata)
            logging.debug("---------Receive: {0}"
                          .format(str(data.hex(' '))))
            if (data[1 + 6] == 0x86) & (data[2 + 6] == 0x01):
                raise Exceptions.function_codeNotSupportedException("Function code not supported by master")
            if (data[1 + 6] == 0x86) & (data[2 + 6] == 0x02):
                raise Exceptions.starting_addressInvalidException("Register address invalid")
            if (data[1 + 6] == 0x86) & (data[2 + 6] == 0x03):
                raise Exceptions.QuantityInvalidException("Invalid Register Value")
            if (data[1 + 6] == 0x86) & (data[2 + 6] == 0x04):
                raise Exceptions.ModbusException("error reading")

                return True

        Write single Register to Master device (Function code 6)
        starting_address:  Register to be written
        value: Register Value to be written
        
Write single Register to Master device (Function code 6)
starting_address:  Register to be written
value: Register Value to be written
Write single Register to Master device ( Function code 6 ) starting_address : Register to be written value : Register Value to be written
def write_single_register(self, starting_address, value):
        logging.info("Request to write single register (FC06), starting address: {0}, value: {1}"
                     .format(str(starting_address), str(value)))
        self.__transactionIdentifier += 1
        if self.__ser is not None:
            if self.__ser.closed:
                raise Exception.SerialPortNotOpenedException("serial port not opened")
        function_code = 6
        length = 6
        transaction_identifier_lsb = self.__transactionIdentifier & 0xFF
        transaction_identifier_msb = ((self.__transactionIdentifier & 0xFF00) >> 8)
        length_lsb = length & 0xFF
        length_msb = (length & 0xFF00) >> 8
        starting_address_lsb = starting_address & 0xFF
        starting_address_msb = (starting_address & 0xFF00) >> 8
        valueLSB = value & 0xFF
        valueMSB = (value & 0xFF00) >> 8
        if self.__ser is not None:
            data = bytearray(
                [self.__unitIdentifier, function_code, starting_address_msb, starting_address_lsb, valueMSB, valueLSB,
                 0, 0])
            crc = self.__calculateCRC(data, len(data) - 2, 0)
            crcLSB = crc & 0xFF
            crcMSB = (crc & 0xFF00) >> 8
            data[6] = crcLSB
            data[7] = crcMSB
            logging.debug("---------Transmit: {0}"
                          .format(str(data.hex(' '))))
            self.__ser.write(data)
            bytes_to_read = 8
            data = None
            data = self.__ser.read(bytes_to_read)
            b = bytearray(data)
            data = b
            logging.debug("---------Receive: {0}"
                          .format(str(data.hex(' '))))
            if len(data) < bytes_to_read:
                raise Exceptions.TimeoutError('Read timeout Exception')
            if (data[1] == 0x86) & (data[2] == 0x01):
                raise Exceptions.function_codeNotSupportedException("Function code not supported by master")
            if (data[1] == 0x86) & (data[2] == 0x02):
                raise Exceptions.starting_addressInvalidException("Register address invalid")
            if (data[1] == 0x86) & (data[2] == 0x03):
                raise Exceptions.QuantityInvalidException("Invalid Register Value")
            if (data[1] == 0x86) & (data[2] == 0x04):
                raise Exceptions.ModbusException("error reading")
            crc = self.__calculateCRC(data, len(data) - 2, 0)
            crcLSB = crc & 0xFF
            crcMSB = (crc & 0xFF00) >> 8
            if (crcLSB != data[len(data) - 2]) & (crcMSB != data[len(data) - 1]):
                raise Exceptions.CRCCheckFailedException("CRC check failed")
            if data[1] == self.__unitIdentifier:
                return True
            else:
                return False
        else:
            protocolIdentifierLSB = 0x00
            protocolIdentifierMSB = 0x00
            length_lsb = 0x06
            length_msb = 0x00
            data = bytearray(
                [transaction_identifier_msb, transaction_identifier_lsb, protocolIdentifierMSB, protocolIdentifierLSB,
                 length_msb, length_lsb, self.__unitIdentifier, function_code, starting_address_msb,
                 starting_address_lsb, valueMSB, valueLSB])
            logging.debug("---------Transmit: {0}"
                          .format(str(data.hex(' '))))
            self.__tcpClientSocket.send(data)
            bytes_to_read = 12
            self.__receivedata = bytearray()
            try:
                while (len(self.__receivedata) == 0):
                    pass
            except Exception:
                raise Exception('Read Timeout')
            data = bytearray(self.__receivedata)
            logging.debug("---------Receive: {0}"
                          .format(str(data.hex(' '))))
            if (data[1 + 6] == 0x86) & (data[2 + 6] == 0x01):
                raise Exceptions.function_codeNotSupportedException("Function code not supported by master")
            if (data[1 + 6] == 0x86) & (data[2 + 6] == 0x02):
                raise Exceptions.starting_addressInvalidException("Register address invalid")
            if (data[1 + 6] == 0x86) & (data[2 + 6] == 0x03):
                raise Exceptions.QuantityInvalidException("Invalid Register Value")
            if (data[1 + 6] == 0x86) & (data[2 + 6] == 0x04):
                raise Exceptions.ModbusException("error reading")
                return True
def write_single_register ( self , starting_address , value ) : logging . info ( "Request to write single register (FC06), starting address: {0}, value: {1}" . format ( str ( starting_address ) , str ( value ) ) ) self . __transactionIdentifier += 1 if self . __ser is not None : if self . __ser . closed : raise Exception . SerialPortNotOpenedException ( "serial port not opened" ) function_code = 6 length = 6 transaction_identifier_lsb = self . __transactionIdentifier & 0xFF transaction_identifier_msb = ( ( self . __transactionIdentifier & 0xFF00 ) >> 8 ) length_lsb = length & 0xFF length_msb = ( length & 0xFF00 ) >> 8 starting_address_lsb = starting_address & 0xFF starting_address_msb = ( starting_address & 0xFF00 ) >> 8 valueLSB = value & 0xFF valueMSB = ( value & 0xFF00 ) >> 8 if self . __ser is not None : data = bytearray ( [ self . __unitIdentifier , function_code , starting_address_msb , starting_address_lsb , valueMSB , valueLSB , 0 , 0 ] ) crc = self . __calculateCRC ( data , len ( data ) - 2 , 0 ) crcLSB = crc & 0xFF crcMSB = ( crc & 0xFF00 ) >> 8 data [ 6 ] = crcLSB data [ 7 ] = crcMSB logging . debug ( "---------Transmit: {0}" . format ( str ( data . hex ( ' ' ) ) ) ) self . __ser . write ( data ) bytes_to_read = 8 data = None data = self . __ser . read ( bytes_to_read ) b = bytearray ( data ) data = b logging . debug ( "---------Receive: {0}" . format ( str ( data . hex ( ' ' ) ) ) ) if len ( data ) < bytes_to_read : raise Exceptions . TimeoutError ( 'Read timeout Exception' ) if ( data [ 1 ] == 0x86 ) & ( data [ 2 ] == 0x01 ) : raise Exceptions . function_codeNotSupportedException ( "Function code not supported by master" ) if ( data [ 1 ] == 0x86 ) & ( data [ 2 ] == 0x02 ) : raise Exceptions . starting_addressInvalidException ( "Register address invalid" ) if ( data [ 1 ] == 0x86 ) & ( data [ 2 ] == 0x03 ) : raise Exceptions . QuantityInvalidException ( "Invalid Register Value" ) if ( data [ 1 ] == 0x86 ) & ( data [ 2 ] == 0x04 ) : raise Exceptions . ModbusException ( "error reading" ) crc = self . __calculateCRC ( data , len ( data ) - 2 , 0 ) crcLSB = crc & 0xFF crcMSB = ( crc & 0xFF00 ) >> 8 if ( crcLSB != data [ len ( data ) - 2 ] ) & ( crcMSB != data [ len ( data ) - 1 ] ) : raise Exceptions . CRCCheckFailedException ( "CRC check failed" ) if data [ 1 ] == self . __unitIdentifier : return True else : return False else : protocolIdentifierLSB = 0x00 protocolIdentifierMSB = 0x00 length_lsb = 0x06 length_msb = 0x00 data = bytearray ( [ transaction_identifier_msb , transaction_identifier_lsb , protocolIdentifierMSB , protocolIdentifierLSB , length_msb , length_lsb , self . __unitIdentifier , function_code , starting_address_msb , starting_address_lsb , valueMSB , valueLSB ] ) logging . debug ( "---------Transmit: {0}" . format ( str ( data . hex ( ' ' ) ) ) ) self . __tcpClientSocket . send ( data ) bytes_to_read = 12 self . __receivedata = bytearray ( ) try : while ( len ( self . __receivedata ) == 0 ) : pass except Exception : raise Exception ( 'Read Timeout' ) data = bytearray ( self . __receivedata ) logging . debug ( "---------Receive: {0}" . format ( str ( data . hex ( ' ' ) ) ) ) if ( data [ 1 + 6 ] == 0x86 ) & ( data [ 2 + 6 ] == 0x01 ) : raise Exceptions . function_codeNotSupportedException ( "Function code not supported by master" ) if ( data [ 1 + 6 ] == 0x86 ) & ( data [ 2 + 6 ] == 0x02 ) : raise Exceptions . starting_addressInvalidException ( "Register address invalid" ) if ( data [ 1 + 6 ] == 0x86 ) & ( data [ 2 + 6 ] == 0x03 ) : raise Exceptions . QuantityInvalidException ( "Invalid Register Value" ) if ( data [ 1 + 6 ] == 0x86 ) & ( data [ 2 + 6 ] == 0x04 ) : raise Exceptions . ModbusException ( "error reading" ) return True
Write single Register to Master device (Function code 6)
starting_address:  Register to be written
value: Register Value to be written
Write single Register to Master device ( Function code 6 ) starting_address : Register to be written value : Register Value to be written
"""
        Write single Register to Master device (Function code 6)
        starting_address:  Register to be written
        value: Register Value to be written
        """ # Check for Exception
self
starting_address
value