iaq_sensor_wired/tests/query_device.py

#!/usr/bin/env python3
from sensor import Sensor
from sys import argv,exit

def print_help():
    print( \
f'''{argv[0]} ADDR [BAUDRATE] [{{read [REGISTER ...] | write VALUE REGISTER}}]
where:
    - ADDR is Modbus address (use 0 for broadcast; note that only write can be broadcast)
    - BAUDRATE is (optional) baud rate; default value is 19200
    - action can be "read", "write", "reset" or it can be omitted completely (defaults to "read all")
    - REGISTER can be either human-friendly register name (e.g. "CO2") or register number (30001)
    - read may have list of REGISTER arguments (combining both register numbers and names)
    - VALUE is 16-bit integer in decimal format
    - to get list of human-friendly names just read all registers from sensor (run script with address only)

EXAMPLES

Read all registers from sensor 247, default baud rate:
    {argv[0]} 247
Read only CO2 register from sensor 123, baudrate 4800:
    {argv[0]} 123 4800 read CO2
Read CO2, T_SHT4x and RH_SCD4x from sensor 222, default baudrate:
    {argv[0]} 222 read CO2 T_SHT4x RH_SCD4x
Turn LED off for sensor 247, baud 115200:
    {argv[0]} 247 115200 write 0 LED_on_register
Set brightness of all connected sensors to 50%:
    {argv[0]} 0 write 50 LED_brightness_register
''')

# default values
DEFAULT_BAUDRATE = 19200
DEFAULT_ACTION = 'all'

# first positional argument (ADDR) is necessary
arg_index = 1
try:
    addr = int(argv[arg_index])
except IndexError:
    print('Argument needed: ADDR')
    print_help()
    exit(-1)
except ValueError:
    print(f'{argv[arg_index]} is invalid form for argument ADDR')
    print_help()
    exit(-2)
# second argument might be either BAUDRATE, read or write (or nothing)
arg_index += 1
action = None
register_number = []
register_name = []
baudrate = None
value = None
while arg_index < len(argv):
    if not baudrate:
        try:
            baudrate = int(argv[arg_index])
        except ValueError:
            baudrate = DEFAULT_BAUDRATE
            continue
    elif not action:
        action = argv[arg_index]
    else:
        # if action is write, last argument should be value
        if action == 'write' and arg_index == len(argv) - 1:
            try:
                value = int(argv[arg_index])
            except ValueError:
                print(f'Wrong format for argument "value"')
                exit(-3)
        else:
            # rest of the arguments should be register numbers/names
            try:
                register_number.append(int(argv[arg_index]))
            except ValueError:
                register_name.append(argv[arg_index])
    arg_index += 1


# sanity check
if not action:
    action = 'all'
if action != 'read' and action != 'write' and action != 'all' and action != 'reset':
    print(f'Unknown action {action}')
    exit(-4)
if action == 'write' and value == None:
    print(f'Action "write" needs value"')
    exit(-5)
if action == 'write' and len(register_number) + len(register_name) != 1:
    print(f'Action "write" needs exactly one register to write to')
    exit(-6)
if not baudrate:
    baudrate = DEFAULT_BAUDRATE
if action != 'write' and len(register_name) + len(register_number) == 0:
    input_registers = [ x for x in Sensor.input_registers.keys() ]
    holding_registers = [ x for x in Sensor.holding_registers.keys() ]
    register_name = input_registers + holding_registers
if action != 'write' and addr == 0:
    print(f'Cannot broadcast action "{action}"')
    exit(-12)
    
# Query device
print('---- Query device ----')
print(f'Address: {addr}\nBaudrate: {baudrate}\nAction: {action}')
if action == 'write':
    print(f'Value to be written: {value}')
# open device
s = Sensor(address=addr, baudrate=baudrate)
if action == 'read' or action == 'all':
    print('---- Register readout ----')
    for register in register_name + register_number:
        if isinstance(register, str):
            reg_name = register
            all_registers = Sensor.input_registers.copy()
            all_registers.update(Sensor.holding_registers)
            if reg_name in all_registers:
                reg_number = all_registers[reg_name]
            else:
                print(f'Register name {register} not known')
                exit(-7)
        else:
            reg_number = register
            reg_name = ''
        try:
            result = s.read_register(reg_number)
        except:
            print(f'{reg_number : <10} {"N/A" : <10} Failed to read register')
            continue
        print(f'{reg_number : <10} {int(result) : <10} {reg_name}')
elif action == 'write':
    if len(register_name) > 0:
        if register_name[0] not in Sensor.holding_registers:
            print(f'Register {register_name[0]} does not exist or is not holding register')
            exit(-9)
        reg_number = Sensor.holding_registers[register_name[0]]
    elif len(register_number) > 0:
        reg_number = register_number[0]
    print('---- Register write ----')
    try:
        s.write_register(reg_number, value)
        print(f'{reg_number : <10} {int(value) : <10}')
    except ValueError:
        print(f'Register number {reg_number} cannot be written')
        exit(-9)
elif action == 'reset':
    if not s.reset():
        # reset should throw NoResponse; if it doesn't, reset failed
        print('Device reset failed!')
    else:
        print('---- Device reset ----')

exit(0)