easymodbus/modbusClient.py
def write_multiple_coils(self, starting_address, values):
        """
        Write multiple coils to Master device (Function code 15)
        starting_address :  First coil to be written
        values:  Coil Values [0..quantity-1] to be written
        """
        logging.info("Request to write multiple coil (FC15), starting address: {0}, values: {1}"
                     .format(str(starting_address), str(values)))
        self.__transactionIdentifier += 1
        if self.__ser is not None:
            if self.__ser.closed:
                raise Exception.SerialPortNotOpenedException("serial port not opened")
        function_code = 15
        if len(values)%8 == 0:
            length = 7 + math.floor(len(values) / 8)
        else:
            length = 7 + math.floor(len(values)/8) + 1
        transaction_identifier_lsb = self.__transactionIdentifier & 0xFF
        transaction_identifier_msb = ((self.__transactionIdentifier & 0xFF00) >> 8)
        starting_address_lsb = starting_address & 0xFF
        starting_address_msb = (starting_address & 0xFF00) >> 8
        quantityLSB = len(values) & 0xFF
        quantityMSB = (len(values) & 0xFF00) >> 8
        valueToWrite = list()
        singleCoilValue = 0
        for i in range(0, len(values)):
            if ((i % 8) == 0):
                if i > 0:
                    valueToWrite.append(singleCoilValue)
                singleCoilValue = 0

            if values[i] == True:
                coilValue = 1
            else:
                coilValue = 0
            singleCoilValue = ((coilValue) << (i % 8) | (singleCoilValue))

        valueToWrite.append(singleCoilValue)
        if self.__ser is not None:
            data = bytearray(
                [self.__unitIdentifier, function_code, starting_address_msb, starting_address_lsb, quantityMSB,
                 quantityLSB])
            data.append(len(valueToWrite))  # Bytecount
            for i in range(0, len(valueToWrite)):
                data.append(valueToWrite[i] & 0xFF)

            crc = self.__calculateCRC(data, len(data), 0)
            crcLSB = crc & 0xFF
            crcMSB = (crc & 0xFF00) >> 8
            data.append(crcLSB)
            data.append(crcMSB)
            logging.debug("---------Transmit: {0}"
                          .format(str(data.hex(' '))))
            self.__ser.write(data)
            bytes_to_read = 8
            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] == 0x8F) & (data[2] == 0x01):
                raise Exceptions.function_codeNotSupportedException("Function code not supported by master")
            if (data[1] == 0x8F) & (data[2] == 0x02):
                raise Exceptions.starting_addressInvalidException(
                    "Starting address invalid or starting address + quantity invalid")
            if (data[1] == 0x8F) & (data[2] == 0x03):
                raise Exceptions.QuantityInvalidException("quantity invalid")
            if (data[1] == 0x8F) & (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 = length & 0xFF
            length_msb = (length & 0xFF00) >> 8
            data = bytearray(
                [transaction_identifier_msb, transaction_identifier_lsb, protocolIdentifierMSB, protocolIdentifierLSB,
                 length_msb, length_lsb, self.__unitIdentifier, function_code, starting_address_msb,
                 starting_address_lsb, quantityMSB, quantityLSB])
            data.append(len(valueToWrite))  # Bytecount
            for i in range(0, len(valueToWrite)):
                data.append(valueToWrite[i] & 0xFF)
            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] == 0x8F) & (data[2] == 0x01):
                raise Exceptions.function_codeNotSupportedException("Function code not supported by master")
            if (data[1] == 0x8F) & (data[2] == 0x02):
                raise Exceptions.starting_addressInvalidException(
                    "Starting address invalid or starting address + quantity invalid")
            if (data[1] == 0x8F) & (data[2] == 0x03):
                raise Exceptions.QuantityInvalidException("quantity invalid")
            if (data[1] == 0x8F) & (data[2] == 0x04):
                raise Exceptions.ModbusException("error reading")

            return True

        Write multiple coils to Master device (Function code 15)
        starting_address :  First coil to be written
        values:  Coil Values [0..quantity-1] to be written
        
Write multiple coils to Master device (Function code 15)
starting_address :  First coil to be written
values:  Coil Values [0..quantity-1] to be written
Write multiple coils to Master device ( Function code 15 ) starting_address : First coil to be written values : Coil Values [ 0 .. quantity - 1 ] to be written
def write_multiple_coils(self, starting_address, values):
        logging.info("Request to write multiple coil (FC15), starting address: {0}, values: {1}"
                     .format(str(starting_address), str(values)))
        self.__transactionIdentifier += 1
        if self.__ser is not None:
            if self.__ser.closed:
                raise Exception.SerialPortNotOpenedException("serial port not opened")
        function_code = 15
        if len(values)%8 == 0:
            length = 7 + math.floor(len(values) / 8)
        else:
            length = 7 + math.floor(len(values)/8) + 1
        transaction_identifier_lsb = self.__transactionIdentifier & 0xFF
        transaction_identifier_msb = ((self.__transactionIdentifier & 0xFF00) >> 8)
        starting_address_lsb = starting_address & 0xFF
        starting_address_msb = (starting_address & 0xFF00) >> 8
        quantityLSB = len(values) & 0xFF
        quantityMSB = (len(values) & 0xFF00) >> 8
        valueToWrite = list()
        singleCoilValue = 0
        for i in range(0, len(values)):
            if ((i % 8) == 0):
                if i > 0:
                    valueToWrite.append(singleCoilValue)
                singleCoilValue = 0
            if values[i] == True:
                coilValue = 1
            else:
                coilValue = 0
            singleCoilValue = ((coilValue) << (i % 8) | (singleCoilValue))
        valueToWrite.append(singleCoilValue)
        if self.__ser is not None:
            data = bytearray(
                [self.__unitIdentifier, function_code, starting_address_msb, starting_address_lsb, quantityMSB,
                 quantityLSB])
            data.append(len(valueToWrite))  
            for i in range(0, len(valueToWrite)):
                data.append(valueToWrite[i] & 0xFF)
            crc = self.__calculateCRC(data, len(data), 0)
            crcLSB = crc & 0xFF
            crcMSB = (crc & 0xFF00) >> 8
            data.append(crcLSB)
            data.append(crcMSB)
            logging.debug("---------Transmit: {0}"
                          .format(str(data.hex(' '))))
            self.__ser.write(data)
            bytes_to_read = 8
            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] == 0x8F) & (data[2] == 0x01):
                raise Exceptions.function_codeNotSupportedException("Function code not supported by master")
            if (data[1] == 0x8F) & (data[2] == 0x02):
                raise Exceptions.starting_addressInvalidException(
                    "Starting address invalid or starting address + quantity invalid")
            if (data[1] == 0x8F) & (data[2] == 0x03):
                raise Exceptions.QuantityInvalidException("quantity invalid")
            if (data[1] == 0x8F) & (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 = length & 0xFF
            length_msb = (length & 0xFF00) >> 8
            data = bytearray(
                [transaction_identifier_msb, transaction_identifier_lsb, protocolIdentifierMSB, protocolIdentifierLSB,
                 length_msb, length_lsb, self.__unitIdentifier, function_code, starting_address_msb,
                 starting_address_lsb, quantityMSB, quantityLSB])
            data.append(len(valueToWrite))  
            for i in range(0, len(valueToWrite)):
                data.append(valueToWrite[i] & 0xFF)
            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] == 0x8F) & (data[2] == 0x01):
                raise Exceptions.function_codeNotSupportedException("Function code not supported by master")
            if (data[1] == 0x8F) & (data[2] == 0x02):
                raise Exceptions.starting_addressInvalidException(
                    "Starting address invalid or starting address + quantity invalid")
            if (data[1] == 0x8F) & (data[2] == 0x03):
                raise Exceptions.QuantityInvalidException("quantity invalid")
            if (data[1] == 0x8F) & (data[2] == 0x04):
                raise Exceptions.ModbusException("error reading")
            return True
def write_multiple_coils ( self , starting_address , values ) : logging . info ( "Request to write multiple coil (FC15), starting address: {0}, values: {1}" . format ( str ( starting_address ) , str ( values ) ) ) self . __transactionIdentifier += 1 if self . __ser is not None : if self . __ser . closed : raise Exception . SerialPortNotOpenedException ( "serial port not opened" ) function_code = 15 if len ( values ) % 8 == 0 : length = 7 + math . floor ( len ( values ) / 8 ) else : length = 7 + math . floor ( len ( values ) / 8 ) + 1 transaction_identifier_lsb = self . __transactionIdentifier & 0xFF transaction_identifier_msb = ( ( self . __transactionIdentifier & 0xFF00 ) >> 8 ) starting_address_lsb = starting_address & 0xFF starting_address_msb = ( starting_address & 0xFF00 ) >> 8 quantityLSB = len ( values ) & 0xFF quantityMSB = ( len ( values ) & 0xFF00 ) >> 8 valueToWrite = list ( ) singleCoilValue = 0 for i in range ( 0 , len ( values ) ) : if ( ( i % 8 ) == 0 ) : if i > 0 : valueToWrite . append ( singleCoilValue ) singleCoilValue = 0 if values [ i ] == True : coilValue = 1 else : coilValue = 0 singleCoilValue = ( ( coilValue ) << ( i % 8 ) | ( singleCoilValue ) ) valueToWrite . append ( singleCoilValue ) if self . __ser is not None : data = bytearray ( [ self . __unitIdentifier , function_code , starting_address_msb , starting_address_lsb , quantityMSB , quantityLSB ] ) data . append ( len ( valueToWrite ) ) for i in range ( 0 , len ( valueToWrite ) ) : data . append ( valueToWrite [ i ] & 0xFF ) crc = self . __calculateCRC ( data , len ( data ) , 0 ) crcLSB = crc & 0xFF crcMSB = ( crc & 0xFF00 ) >> 8 data . append ( crcLSB ) data . append ( crcMSB ) logging . debug ( "---------Transmit: {0}" . format ( str ( data . hex ( ' ' ) ) ) ) self . __ser . write ( data ) bytes_to_read = 8 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 ] == 0x8F ) & ( data [ 2 ] == 0x01 ) : raise Exceptions . function_codeNotSupportedException ( "Function code not supported by master" ) if ( data [ 1 ] == 0x8F ) & ( data [ 2 ] == 0x02 ) : raise Exceptions . starting_addressInvalidException ( "Starting address invalid or starting address + quantity invalid" ) if ( data [ 1 ] == 0x8F ) & ( data [ 2 ] == 0x03 ) : raise Exceptions . QuantityInvalidException ( "quantity invalid" ) if ( data [ 1 ] == 0x8F ) & ( 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 = length & 0xFF length_msb = ( length & 0xFF00 ) >> 8 data = bytearray ( [ transaction_identifier_msb , transaction_identifier_lsb , protocolIdentifierMSB , protocolIdentifierLSB , length_msb , length_lsb , self . __unitIdentifier , function_code , starting_address_msb , starting_address_lsb , quantityMSB , quantityLSB ] ) data . append ( len ( valueToWrite ) ) for i in range ( 0 , len ( valueToWrite ) ) : data . append ( valueToWrite [ i ] & 0xFF ) 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 ] == 0x8F ) & ( data [ 2 ] == 0x01 ) : raise Exceptions . function_codeNotSupportedException ( "Function code not supported by master" ) if ( data [ 1 ] == 0x8F ) & ( data [ 2 ] == 0x02 ) : raise Exceptions . starting_addressInvalidException ( "Starting address invalid or starting address + quantity invalid" ) if ( data [ 1 ] == 0x8F ) & ( data [ 2 ] == 0x03 ) : raise Exceptions . QuantityInvalidException ( "quantity invalid" ) if ( data [ 1 ] == 0x8F ) & ( data [ 2 ] == 0x04 ) : raise Exceptions . ModbusException ( "error reading" ) return True
Write multiple coils to Master device (Function code 15)
starting_address :  First coil to be written
values:  Coil Values [0..quantity-1] to be written
Write multiple coils to Master device ( Function code 15 ) starting_address : First coil to be written values : Coil Values [ 0 .. quantity - 1 ] to be written
"""
        Write multiple coils to Master device (Function code 15)
        starting_address :  First coil to be written
        values:  Coil Values [0..quantity-1] to be written
        """ # Bytecount # Bytecount
self
starting_address
values