pyHeatpump/pyheatpump/modbus.py

#!/usr/bin/env python3
from serial import Serial
from serial.serialutil import SerialException
from umodbus.client.serial import rtu
import umodbus
from pprint import pprint

from .config import config
from .models import *

serial_conn = None

def connect():
    global serial_conn

    if serial_conn is None:
        print('Connecting to serial port *{}*'.format(
            config.get('heatpump', 'serial_port')))
        serial_conn = Serial(
            port=config.get('heatpump', 'serial_port'),
            baudrate=config.get('heatpump', 'baudrate'),
            bytesize=8,
            parity='N',
            stopbits=1,
            timeout=10)

    if serial_conn.open is False:
        print('Opening serial port')
        serial_conn.open()

    return serial_conn

def read_coils(start, end):
    global serial_con
    connect()
    res = []

    address = -1
    qty = -1

    try:
        for address in range(start, end + 1, 125):
            qty = 125 if (end - address) >= 125 else (end - address)
            if not qty:
                break

            req_adu = rtu.read_coils(
                slave_id=1,
                starting_address=address,
                quantity=qty)

            resp = rtu.send_message(req_adu, serial_conn)
            res.extend(resp)
    except umodbus.exceptions.IllegalDataAddressError as e:
        print(e)
        print(f'{address} {qty}')

    return res



def read_holding_registers(start, end):
    global serial_conn
    connect()
    res = []

    address = -1
    qty = -1

    try:
        for address in range(start, end + 1, 125):
            qty = 125 if (end - address) >= 125 else (end - address)
            if not qty:
                break

            req_adu = rtu.read_holding_registers(
                slave_id=1,
                starting_address=address,
                quantity=qty)

            resp = rtu.send_message(req_adu, serial_conn)
            res.extend(resp)
    except umodbus.exceptions.IllegalDataAddressError as e:
        print(e)
        print(f'{address} {qty}')

    return res

def write_coil(var_value: VariableValue):
    global serial_conn
    connect()

    try:
        req_adu = rtu.write_single_coil(
            slave_id=1,
            address=var_value.address,
            value=var_value.value)
        return resp.send_message(req_adu, serial_conn)
    except Exception as e:
        raise e


def write_holding_register(var_value: VariableValue):
    global serial_conn
    connect()

    try:
        req_adu = rtu.write_single_register(
            slave_id=1,
            address=var_value.address,
            value=var_value.value)
        return resp.send_message(req_adu, serial_conn)
    except Exception as e:
        raise e

if __name__ == '__main__':
    resp = read_holding_registers(1, 5000)
    #pprint(resp)
    print(len(resp))
    resp = read_coils(1, 2000)
    #pprint(resp)
    print(len(resp))