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 | reset}}]
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
Read register 30010 from sensor 247, default baudrate:
    {argv[0]} 247 read 30010
Read registers 30011, 30012, 30020 from sensor 247, baudrate 19200:
    {argv[0]} 247 19200 read 30011 30012 30020
Turn LED off for sensor 247, baud 115200:
    {argv[0]} 247 115200 write LED_on 0
Set brightness of all connected sensors to 50%, default baudrate (broadcast message):
    {argv[0]} 0 write LED_brightness 50
Reset device 247, baud 19200:
    {argv[0]} 247 19200 reset
Reset all devices, default baudrate:
    {argv[0]} 0 reset
''')

# 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 addr == 0 and action != 'write' and action != 'reset':
    print(f'Cannot broadcast action "{action}"')
    exit(-12)
    
# Query device
print('---- Query device ----')
print(f'Address: {addr if addr != 0 else "0 (broadcast)"}\nBaudrate: {baudrate}\nAction: {action}')
if action == 'write':
    print(f'Value to be written: {value}')
# open device
s = Sensor(address=addr, baudrate=baudrate)
all_registers = Sensor.input_registers.copy()
all_registers.update(Sensor.holding_registers)
all_register_numbers = {}
for name, number in all_registers.items():
    all_register_numbers[number] = name
if action == 'read' or action == 'all':
    print('---- Register readout ----')
    for register in register_name + register_number:
        signed = False
        if isinstance(register, str):
            reg_name = register
            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
            # try reverse lookup
            reg_name = all_register_numbers[reg_number] if reg_number in all_register_numbers else ''
        try:
            signed = '_signed' in reg_name
            result = int(s.read_register(reg_number, signed=signed))
        except:
            result = 'N/A'
        print(f'{reg_number : <10} {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() and addr != 0:
        # reset should throw NoResponse; if it doesn't, reset failed (broadcast shouldn't return anything)
        print('Device reset failed!')
    else:
        print('---- Device reset ----')

exit(0)