docker-thermostat/relay.py

#!/usr/bin/env python3
# Pilotage GCE USB 8 Relay Board
import time
import serial
import serial.tools.list_ports
import argparse
import sys
import subprocess

parser = argparse.ArgumentParser(description='Control relays.')
parser.add_argument('relay_list', type=str,
                    help='list of relays in ["1", ..., "8"] separated by comma, no spaces. "all" means all relays.')
parser.add_argument('action', type=str, choices=['on', 'off', 'status'],
                    help='Action on the list of relays.')

args = parser.parse_args()

if args.relay_list == 'all':
  relays = [ 'RLY'+str(i) for i in range(1,9)]
else:
  relays = []
  for relay in args.relay_list.split(','):
    if relay in [ str(i) for i in range(1,9)]:
      relays.append('RLY'+relay)
    else:
      print("ERROR: Relay '"+relay+"' does not exist.")
      sys.exit(1)

if args.action == 'on':
  action = '1'
elif args.action == 'off':
  action = '0'
else:
  action = 'status'

#print(relays)

port = None
    
def list_serial_ports():
    ports = serial.tools.list_ports.comports()
    for port, desc, hwid in sorted(ports):
        print("{}: {} [{}]".format(port, desc, hwid))

def find_serial_ports():
    ports = serial.tools.list_ports.comports()
    for port, desc, hwid in sorted(ports):
        if "FT232R USB UART" in desc:
            return port

port = find_serial_ports()
if port is None:
    print("Unable to find serial port for relay board please use one of the following")
    list_serial_ports()
    sys.exit(2)

def init_serial(pPort):         
  global ser          
  ser = serial.Serial()
  ser.baudrate = 9600 
  ser.port = pPort
  ser.bytesize = 8
  ser.parity = 'N'
  ser.timeout = None
  ser.xonxoff = False
  ser.rtscts=False
  ser.dsrdtr=False
  ser.timeout = 1
  ser.open()          

  if ser.isOpen():
    pass
    #print('Connected to ' + ser.portstr) 
  else:
    print('Could not connect to ' + ser.portstr) 
    sys.exit(1)

init_serial(port)
if action != 'status':
  for relay in relays:
    command = relay+action+'\r\n'
    ser.write(command.encode('ascii'))
    bytes = ser.readline() 
    #print ('Renvoie :\r\n' + bytes) 

ser.write('?RLY\r\n'.encode('ascii'))
answer = ser.readline().decode('ascii')
status = ''
for char in answer:
  if char in ['0', '1']:
    status+= char
if len(status) != 8:
  print(len(status))
  print("ERROR: status cannot be parsed.")
  print('status:\r\n' + status) 
  sys.exit(1)
elif action != 'status':
  print ('status:\r\n' + status) 

if action != 'status':
  for i in range(1,9):
    relay = status[i-1]
    if relay not in ['0', '1']:
      print("ERROR: unrecognized value '"+relay+"' for relay "+str(i))
    elif relay in relays and relay != action:
      print("ERROR: wrong status '"+relay+"' for relay "+str(i))
else:
  for i in [int(relay[3]) for relay in relays]:
    relay = status[i-1]
    if relay not in ['0', '1']:
      print("ERROR: unrecognized value '"+relay+"' for relay "+str(i))
    else:
      print(relay, end='')

ser.close()