source: powerbar/npm4000.py@ 127

#!/usr/bin/env python
# NPM 4000 powerbar managment script, to be used instead of the windows
# application. 
#
# XXX: Not all features are ported yet (like amp monitoring)
# XXX: Some dirty poking around with hex representations and numeric representations of ports
# XXX: Make proper classes for use 
# XXX: Documentation
#
# Licence: BSD
# Version: $Id: npm4000.py 750 2009-09-27 14:34:55Z rick $
# Rick van der Zwet <info@rickvanderzwet.nl>

import socket, time
import getopt, sys

MAX_BUFFER = 100
MAX_PORTS = 24

# XOR all the 8bit values to get a checksum
def checksum(s):
    crc = 0
    for p in range(0, len(s),2):
        crc ^= int(s[p:p+2], 16)
    return "%02X" % crc

inet_addr = '192.168.0.178'
inet_port = 4001

passwd = 0x12345678
addr_code = 0xFFFF 

verbose = False

cmd2code = { 'login'       : (0x5507, 1),
         'portOn'      : (0xB204, 2),
         'portOff'     : (0xC204, 2),
         'allPortsOff' : (0xC103, 6),
         'allPortsOn'  : (0xB104, 15),
         'status'      : (0xD103, 2),
         'knownError'  : (0xFFFF, 1),
       }

# Socket used troughout the code
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

def getCode(type):
    return(cmd2code[type][0])

def getTimeout(type):
    return(cmd2code[type][1])

def dprint(msg):
    if verbose:
        print time.strftime('[%Y-%m-%d %H:%M:%S]'), msg

def getPortNumber(name):
    port_char = name[0]
    port_number = int(name[1])
    number = -1
    if port_char == 'A':
        number = port_number
    elif port_char == 'B':
        number = port_number + 8
    elif port_char == 'C':
        number = port_number + 16
    else:
        raise ValueError, "Port %s not defined" % name

    return(number)

def getPortName(number):
    name = "U3"
    if number <= 8:
        name = "A%i" % number
    elif number <= 16:
        name = "B%i" % (number - 8)
    elif number <= 24:
        name = "C%i" % (number - 16)
    else:
        raise ValueError, "Port %i not defined" % number
    return(name)

def getPortHex(number):
    return(int(getPortName(number),16))

def generateCommand(type,arg=None):
    if arg == None:
        command = "%04X%04X" % (getCode(type), addr_code)
    else:
        command = "%04X%04X%02X" % (getCode(type), addr_code, arg)
    command += checksum(command)
    return (command)


def sendCommand(type, arg=None):
    cmd = generateCommand(type,arg)
    timeout = getTimeout(type)
    dprint('Send: ' + cmd + ' (%s)' % (type))
    s.send(cmd)
    s.settimeout(timeout)
    try:
        retval = s.recv(MAX_BUFFER)
        dprint('Recv: ' + retval)
    except socket.timeout:
        retval = None
        print 'Recv: ERR timeout (wrong command, bad connection)'
        #XXX: XOR Checking for data correctness
    return(retval)

def doCommand(type, arg=None):
    sendCommand('login', passwd)
    retval = sendCommand(type, arg)
    #XXX: Check if command returned succesfull D128FF
    return(retval)


def getPortState(port, raw_status):
    """Port configuration is put into 24 bits, every port has one bit
    representing the statewith a awkward order: 
    A8,A7,A6,A5,A4,A3,A2,A1,B8,..,B1,C8,..,C1"""
    # Portion of retval we are interested in
    ports_state = int(raw_status[8:14],16)
    port_bit_location = -1
    if port <= 8:
        port_bit_location =  (1 << 15 << port)
    elif port <= 16:
        port_bit_location = (1 << 7 << (port - 8))
    elif port <= 24:
        port_bit_location = (1 << (port - 16 - 1))

    if port_bit_location & ports_state:
        return True
    else:
        return False

def getPortStatus(i):
    raw_status = doCommand('status')
    print "Port %02i [%s]:" % (i, getPortName(i)),
    if getPortState(i,raw_status):
        print "1"
    else:
        print "0"
    return(retval)

def getStatusAll():
    raw_status= doCommand('status')
    for i in range(1,25):
        print "Port %02i [%s]:" % (i, getPortName(i)),
        if getPortState(i,raw_status):
            print "1"
        else:
            print "0"

def togglePort(port):
    if getPortState(port):
        doCommand('portOff',port)
    else:
        doCommand('portOn', port)


def runTest():
    print "TEST: Start will all port Offline"
    doCommand('allPortsOff')

    print "TEST: All On and Off again, in mass execution"
    doCommand('allPortsOn',0x01)
    doCommand('allPortsOff')

    print "TEST: Enable and disable ports one by one"
    for i in range(1,25):
        print getPortName(i)
        doCommand('portOn', getPortHex(i))
        doCommand('portOff', getPortHex(i))

    print "TEST: Send known error"
    doCommand('knownError')


def usage():
    print """
Usage %s arguments
Version: $Id: npm4000.py 750 2009-09-27 14:34:55Z rick $

Arguments:
  [-h|--help]                   Reading right know
  [-v|--verbose]                Print extra communication output
  --host=                       IP adress of FQDN hostname [%s]
  --port=                       Port to connect to [%s]
  --password=                   Password to use in hex notation [0x1234568]
  --addresscode=                Internal device number in hex notation [0xFFFF]
  [-s|--status]                 Current port configuration
  [-t <port>|--toggle=<port>]   Toggle port(s)
  [-o <port>|--on=<port>]       Turn on port(s)
  [-f <port>|--off=<port>]      Turn off port(s)

Note: <port> has different notations:
  Numeric value of port         1,2,3,4,5,..
  Actual value of port          A1,..,A8,B1,..,B8,C1,..,C8 
  All ports                     all
    """ % (sys.argv[0], inet_addr, inet_port)

def main():
    global verbose, addr_code, inet_addr, inet_port, passwd, s
    try:
        opts, args = getopt.getopt(sys.argv[1:], 
            "hf:s:t:o:v", ["help","verbose","host=", "port=", "password=", "addresscode=","toggle=","off=", "on=", "status="])
    except getopt.GetoptError, err:
        # print help information and exit:
        print str(err) # will print something like "option -a not recognized"
        usage()
        sys.exit(2)

    opt_port = None
    opt_action = None
    for o, a in opts:
        if o in ("-v", "--verbose"):
            verbose = True
        elif o in ("-h", "--help"):
            usage()
            sys.exit()
        elif o in ("--addresscode"):
            addr_code = int(a,16)
        elif o in ("--host"):
            inet_addr = a
        elif o in ("--password"):
            passwd = int(a,16)
        elif o in ("--port"):
            inet_port = a
        elif o in ("-s", "--status"):
            opt_action = "status"
            opt_port = a
        elif o in ("-t","--toggle"):
            opt_action = "toggle"
            opt_port = a
        elif o in ("-f","--off"):
            opt_action = "off"
            opt_port = a
        elif o in ("-o","--on"):
            opt_action = "on"
            opt_port = a
        else:
            assert False, "unhandled option"

    if (opt_port == None or opt_action == None):
        usage()
        sys.exit(2)

    dprint ('action: ' + opt_action + ' port: ' + opt_port) 

    s.connect((inet_addr, inet_port))

    # Status needs real integers, hack
    if opt_action == "status":
        if opt_port == "all":
            getStatusAll()
        else:
            print "XXX: Implement"
        sys.exit(0)

    # Resolve port to proper number
    if opt_port == "all":
        None # Blank resolution, as it is done elsewhere
    elif opt_port[0] in ("A","B","C"):
        opt_port = int(opt_port,16)
        dprint('Hexcode of port: %i' % opt_port)
    else:
        # Dirty hack to have conversion and checking at the same time
        opt_port = getPortHex(int(opt_port))
        dprint('Hexcode of port: %i' % opt_port)

    if opt_action == "toggle":
        if opt_port == "all":
            for i in range(1,25):
                togglePort(getPortHex(i))
        else:
                togglePort(opt_port)
    elif opt_action == "on":
        if opt_port == "all":
            doCommand("allPortsOn",0x01)
        else:
            doCommand("portOn", opt_port)
    elif opt_action == "off":
        if opt_port == "all":
            doCommand("allPortsOff");
        else:
            doCommand("portOn", opt_port)


if __name__ == "__main__":
    main()