Here is something in python for that specific relay module.
[code]#!/usr/bin/python
import os
import sys
import time
import math
import numpy # sudo apt-get python-numpy
import usb.core # sudo apt-get python-pyusb
import usb.util # sudo apt-get python-pyusb
os.environ[“PYUSB_DEBUG”] = “debug”
os.environ[“PYUSB_LOG_FILENAME”] = “/home/pi/starkwolf/pyusb.log”
class HIDRelay:
“”“HID USB SainSmart 16 Relay Module using NUC122ZC1 - AN413AD22 - 31B033 - ARM”""
print_cfg = False
vid = 0x0416
pid = 0x5020
hid = None
ep_in = None # From Device to Host Computer.
ep_in_address = 0x84 # bEndpointAddress -> 0x84 EP 4 IN
ep_out = None # From Host Computer to Device.
ep_out_address = 0x05 # bEndpointAddress -> 0x05 EP 5 OUT
relayBitmap = None
relayBitmap16 = [128, 256, 64, 512, 32, 1024, 16, 2048, 8, 4096, 4, 8192, 2, 16384, 1, 32768]
relayBitmap8 = [128, 256, 64, 512, 32, 1024, 16, 2048, 8]
max_relay_id = None
max_relay_id16 = 15
max_relay_id8 = 7
packet_len = 64
"""Initiates de device. Ends the process if no device was found."""
def __init__(self, version=16):
if (version is 8) or (version is "8"):
self.relayBitmap = self.relayBitmap8
self.max_relay_id = self.max_relay_id8
else:
self.relayBitmap = self.relayBitmap16
self.max_relay_id = self.max_relay_id16
init_ok = self.get_device()
return None
"""Small function to pack array of bytes"""
def pack_bytes(self, bytesArray):
packet = [0x0] * self.packet_len
i = 0
for byte in bytesArray:
packet[i] = byte
i += 1
bytesString = ''.join([chr(c) for c in packet])
return bytesString
"""Small function to unpack array of bytes"""
def unpack_bytes(self, bytesString):
bytesArray = bytearray()
bytesArray.extend(bytesString)
return bytesArray
"""Gets the devices and sets the two endpoints."""
def get_device(self):
# initialising debuging - don't have a clue if this will work
self.hid = usb.core.find(idVendor=self.vid, idProduct=self.pid)
# was it found?
if self.hid is None:
print 'Device not found'
return False
try:
self.hid.detach_kernel_driver(0)
except: # this usually mean that kernel driver has already been dettached
pass
if self.print_cfg is True:
for cfg in self.hid:
sys.stdout.write('cfg.bConfigurationValue: ' + str(cfg.bConfigurationValue) + '\
‘)
for intf in cfg:
sys.stdout.write(’\t’ + \
'intf.bInterfaceNumber: ’ + str(intf.bInterfaceNumber) + \
‘,’ + \
‘intf.bAlternateSetting: ’ + str(intf.bAlternateSetting) + \
’
‘)
for ep in intf:
sys.stdout.write(’\t\t’ + \
'ep.bEndpointAddress: ’ + str(ep.bEndpointAddress) + \
’
')
# access the second configuration
cfg = self.hid[0]
# access the first interface
intf = cfg[(0,0)]
# IN endpoint
self.ep_in = intf[0]
# OUT endpoint
self.ep_out = intf[1]
# Set the first configuration
self.hid.set_configuration()
return True
def read(self):
readCmd = [0xD2, 0x0E, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x48, 0x49, 0x44,
0x43, 0x80, 0x02, 0x00, 0x00, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC,
0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC,
0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC,
0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC]
readCmd_pack = self.pack_bytes(readCmd)
# assert len(self.hid.write(self.ep_out_address, readCmd_pack, 100)) == len(readCmd_pack)
self.hid.write(self.ep_out_address, readCmd_pack, 100)
data_pack = self.hid.read(self.ep_in_address, self.packet_len, 100)
np_arr8 = numpy.uint8([data_pack[2], data_pack[3]])
arr16 = np_arr8.view('uint16')
mask = arr16[0]
return mask
def reset(self):
resetCmd = [0x71, 0x0E, 0x71, 0x00, 0x00, 0x00, 0x11, 0x11, 0x00, 0x00, 0x48, 0x49, 0x44,
0x43, 0x2A, 0x02, 0x00, 0x00, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC,
0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC,
0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC,
0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC]
resetCmd_pack = self.pack_bytes(resetCmd)
self.hid.write(self.ep_out_address, resetCmd_pack, 100)
def write(self, mask):
if (type(mask) is not int) or (mask < 0) or (mask > 0xFFFF):
raise ValueError('Invalid write mask: ', mask)
writeCmd = [0xC3, 0x0E, 0x05, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x48, 0x49, 0x44,
0x43, 0xEE, 0x01, 0x00, 0x00, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC,
0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC,
0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC,
0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC, 0xCC];
np_arr16 = numpy.uint16([mask])
np_arr8 = np_arr16.view('uint8')
writeCmd[2] = np_arr8[0]
writeCmd[3] = np_arr8[1]
size = writeCmd[1]
writeCmd_sliced = writeCmd[0:size]
checksum = reduce((lambda a, b: a + b), writeCmd_sliced)
np_arr32 = numpy.uint32([checksum])
arr8 = np_arr32.view('uint8')
for x in range(0, 3):
writeCmd[size + x] = arr8[x]
writeCmd_pack = self.pack_bytes(writeCmd)
self.hid.write(self.ep_out_address, writeCmd_pack, 100)
return True
def set(self, relay_id, state):
if type(relay_id) is not int:
raise ValueError('Invalid relay ID type: ', relay_id)
if (relay_id < 0) or (relay_id > self.max_relay_id):
raise ValueError('Invalid relay ID: ', relay_id)
if type(state) is not bool:
raise ValueError('Invalid state: ', state)
readmask = self.read()
print "Readmask: ", readmask
bit = int(math.pow(2, relay_id))
mask = 0
isOn = False
for x in range(0, self.max_relay_id):
if (readmask & self.relayBitmap[x]) != 0:
currentRelayMask = int(math.pow(2, x))
mask = mask | currentRelayMask
if currentRelayMask == int(bit):
isOn = True
print "Led ", x, " is ON"
print "Current Mask: ", mask
if state is True:
mask = mask | bit
elif state is False:
if isOn is True:
mask = mask ^ bit
print "Writing Mask:" , mask
writemask = self.write(mask)
print "Writemask: ", writemask
new_readmask = self.read()
print "New Readmask: ", new_readmask
return writemask
def testRelay():
relay = HIDRelay(8)
relay.write(0)
relay.reset()
time.sleep(1)
relay.read()
time.sleep(0.25)
relay.write(0xFFFF)
time.sleep(0.25)
relay.write(0)
time.sleep(0.25)
relay.write(0xFFFF)
time.sleep(0.25)
relay.write(0)
p = relay.read()
time.sleep(0.5)
for x in range(0, relay.max_relay_id):
p = relay.set(x, True)
time.sleep(0.7)
for x in range(relay.max_relay_id, 0, -1):
p = relay.set(x, False)
time.sleep(0.7)
time.sleep(0.5)
relay.write(0xFFFF)
time.sleep(0.25)
relay.write(0)
def mapRelay():
relay = HIDRelay(8)
relayBitmap = []
# Turn off all relays
relay.write(0)
relay.read()
time.sleep(2)
for x in range(0, relay.max_relay_id):
relay.write(int(math.pow(2, x)))
readmask = relay.read()
print "Readmask", readmask
relayBitmap.insert(x, readmask)
time.sleep(2)
# Clear all relays
relay.write(0)
print "Relay BitMap:", str(relayBitmap)
print ‘Number of arguments:’, len(sys.argv), ‘arguments.’
print ‘Argument List:’, str(sys.argv)
if (len(sys.argv) > 1):
relay = HIDRelay(8)
cliArgs = sys.argv
print ‘Command is:’, cliArgs[1]
if (cliArgs[1] == 'reset'):
print 'Resetting...'
relay.read()
relay.write(0)
relay.reset()
if (cliArgs[1] == 'test'):
print 'Testing...'
testRelay()
if (cliArgs[1] == 'map'):
print 'Mapping...'
mapRelay()
if (cliArgs[1] == 'on') and len(cliArgs) == 3:
state = True
relay_id = int(cliArgs[2])
print "Turning Relay", relay_id, "On"
relay.set(relay_id, state)
if (cliArgs[1] == 'off') and len(cliArgs) == 3:
state = False
relay_id = int(cliArgs[2])
print "Turning Relay", relay_id, "Off"
relay.set(relay_id, state)
if (cliArgs[1] == 'write') and len(cliArgs) == 3:
mask = int(cliArgs[2])
relay.write(mask)
readmask = relay.read()
print "Readmask: ", readmask
if (cliArgs[1] == 'read'):
readmask = relay.read()
print "Readmask: ", readmask[/code]