Changeset 156

Timestamp:

Jul 16, 2010, 10:24:29 AM (14 years ago)

Author:

Rick van der Zwet

Message:

Work in progress, lets make it look pretty

File:

• powerbar/serial-npm4000.py (modified) (10 diffs)

powerbar/serial-npm4000.py

@@ -3 +3 @@
 # application. 
 #
-# XXX: Not all features are ported yet (like amp monitoring)
-# XXX: Some dirty poking around with hex representations and numeric representations of ports
-#  c2 04 ff ff a1 67XXX: Make proper classes for use 
-# XXX: Documentation
+# TODO: Not all features are ported yet (like amp monitoring)
+# TODO: Error handling not implemented
 #
 # Licence: BSD
@@ -16 +14 @@
 import time
 import random
-
+import getopt
+
+# Set to true to enable verbose communication aka debugging
 DEBUG = False
 
+# Default for options
+opt_serial_port = '/dev/ttyUSB0'
+opt_password = 0x1245678
+opt_address_code = 0xFFFF
+opt_baudrate = 19200
+
+# Serial connection port status is cached globally to avoid overhead
+serial = None
+port_status_synced = False
+ports_status = None
+
 def make_checksum(command):
+    """ Generate CRC checksum using XOR on all bytes """
     crc = 0
     for item in command:
@@ -25 +37 @@
     return crc
 
+
+
 def debug(msg):
-  if DEBUG:
-    print msg
+    """ Print debug statement if DEBUG is set """
+    if DEBUG:
+      print msg
+
+
 
 def hex_to_str(command):
+    """ Human readable representation of command """
     return " ".join(["%02x" % item for item in command])
 
+
+
 def str_to_hex(s):
-    """Hexadecimal string representation of 's'"""
+    """ Hexadecimal string representation of 's' """
     return [ord(x) for x in s]
 
+
+
 def port_to_hex(port_number):
+  """ Convert integer port number to hexadecimal presentation as internal
+      location 
+  """
+  if port_number < 1:
+    assert False, "Invalid port port_number (%i)" % port_number
   if port_number <= 8:
     port = 0xa0 + port_number
@@ -49 +76 @@
 
 
+
 def hex_to_port(port):
+  """ Convert hexadecimal port to human port number """
   base = port & 0xf0
   index = port & 0x0f
@@ -64 +93 @@
 
 
-def send_raw_command(ser, raw_command, response_size=1024):
-    """ Send command to serial device and wait for response """
+
+def send_raw_command(raw_command, response_size=1024):
+    """ Send raw command to serial device and wait for response """
 
     print "Going to send: " + hex_to_str(raw_command)
     send_line = "".join([chr(item) for item in raw_command])
-    f = open('input.bin','w')
-    f.write(send_line)
-    f.close()
-    
-    ser.write(send_line)
-    recv_line = ser.read(response_size)
+    serial.write(send_line)
+    recv_line = serial.read(response_size)
     recv_command = str_to_hex(recv_line)
     print "Received: %s (%i)" % (hex_to_str(recv_command), len(recv_line))
     return(recv_command)
 
-def send_command(ser, command, response_size=1024):
-    raw_command = command + [make_checksum(command)]
-    return send_raw_command(ser, raw_command, response_size)
-
-
-
-#print "%02X" % make_checksum([0xAA, 0x03, 0xFF, 0xFF])
-
-
-#[01:38:14] Send: 55 07 ff ff 12 34 56 78 5a ff ff d2 00
-#[01:38:14] Recv: ff ff a9 d1 28 ff ff 00 07 00 11 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 28 00 28 00 00 10 ff
-#[01:38:14] Comm: Refresh
-
-#55 07 ff ff 12 34 56 78  5a
-action_login = [0x55, 0x07]
-device_id = [0xff, 0xff]
-password = [0x12, 0x34, 0x56, 0x78]
-
-ser = serial.Serial('/dev/ttyUSB0', 19200, timeout=5)
-print ser
-
-# Login cycle
-#command = action_login + device_id + password
-#send_command(ser, command)
-
-# Refresh cycle
-#action_refresh = [0xd2, 0x00]
-#raw_command = command + [0x5a] + device_id + action_refresh
-#send_raw_command(ser, raw_command)
-# 55 07 ff ff 12 34 56 78  5a c2 04 ff ff a1 67
-#send_raw_command(ser, [0x55,0x07,0xff,0xff,0x12,0x34,0x56,0x78,0x5a])
-# 55 07 ff ff 12 34 56 78  5a d1 03 ff ff d2 00
-#send_raw_command(ser, [0x55, 0x07, 0xff, 0xff, 0x12, 0x34, 0x56, 0x78, 0x5a, 0xd1, 0x03, 0xff, 0xff, 0xd2, 0x00])
-
-# 55 07 FF FF 12 34 56 78 5A D1 03 FF FF D2 00  
-login_line = [0x55, 0x07, 0xff, 0xff, 0x12, 0x34, 0x56, 0x78]
-refresh_line = [0xd1, 0x03, 0xff, 0xff]
-allon_line = [0xb1, 0x03, 0xff, 0xff]
-alloff_line = [0xc1, 0x03, 0xff, 0xff]
-port_on_line = [0xb2, 0x04, 0xff, 0xff, 0xa1]
-port_off_line = [0xc2, 0x04, 0xff,0xff, 0xa1]
-power_on_interval_125_line = [0xd6, 0x04, 0xff, 0xff, 0xfa, 0x28]
-power_on_interval_05_line = [0xd6, 0x04, 0xff, 0xff, 0x01, 0xd3]
-change_address_code_line = [0x05, 0x00, 0xff, 0xff, 0x00, 0x04]
-modify_password_line = [0xd3, 0x07, 0xff, 0xff, 0x11, 0x11, 0x11, 0x11]
-
-def port_on_cmd(port_number):
-  line = port_on_line
-  line[4] = port_to_hex(port_number)
-  return line
-
-def port_off_cmd(port_number):
-  line = port_off_line
-  line[4] = port_to_hex(port_number)
-  return line
+
 
 def address_to_num(address):
+   """ Convert internal address representation to integer """
    return (address[0] << 8) ^ address[1]
 
+
+
 def num_to_address(npm_number):
+   """ Convert address number to internal representation """
    return [npm_number >> 8, npm_number & 0x00ff]
+
+
 
 def bin_reverse(number,width=8):
   """Little hacking using string logic to binary reverse number"""
   return int(bin(number)[2:].zfill(width)[::-1],2)
-
-
-def get_port_status(retval):
-   # Example port 1 is off
-   # d1 28 ff ff fe ff ff
-   #             ^^ ^^ ^^
-   port_array = [False] * 25
-   status_array = bin_reverse((bin_reverse(retval[4]) << 16) ^ (bin_reverse(retval[5]) << 8) ^ bin_reverse(retval[6]),24)
-   for port in range(0,24):
-      if (status_array & (1 << port)) > 0:
-        port_array[port+1] = True
-   return port_array
-      
 
 #send_command(ser, login_line, 5)
@@ -169 +134 @@
 #retval = send_command(ser, refresh_line, 42)
 #print get_port_status(retval)
-while True:
-  port_number = random.randint(1,24)
-  print port_number
-  send_command(ser, login_line, 5)
-  send_command(ser, port_on_cmd(port_number),6)
-  send_command(ser, login_line, 5)
-  send_command(ser, port_off_cmd(port_number),6)
+#while True:
+#  port_number = random.randint(1,24)
+#  print port_number
+#  send_command(ser, login_line, 5)
+#  send_command(ser, port_on_cmd(port_number),6)
+#  send_command(ser, login_line, 5)
+#  send_command(ser, port_off_cmd(port_number),6)
+
+# Login cycle
+#command = action_login + device_id + password
+#send_command(ser, command)
+# Reference implementation lines
+# A = action, B = address, C = password, P = port
+# Mostly of type [A, A, B, B, C, C, C, C] or [A, A, B, B] or [A, A, B, B, P]
+# (command, return_type, timeout)
+line = dict()
+line['login'] = ([0x55, 0x07, 0xff, 0xff, 0x12, 0x34, 0x56, 0x78], 5, 1)
+line['status'] = ([0xd1, 0x03, 0xff, 0xff], 42, 1)
+line['allon'] = ([0xb1, 0x03, 0xff, 0xff], 6, 13)
+line['alloff'] = ([0xc1, 0x03, 0xff, 0xff], 6, 13)
+line['port_on'] = ([0xb2, 0x04, 0xff, 0xff, 0xa1], 6, 1)
+line['port_off'] = ([0xc2, 0x04, 0xff,0xff, 0xa1], 6, 1)
+line['power_on_interval_125'] = ([0xd6, 0x04, 0xff, 0xff, 0xfa, 0x28], 5, 1)
+line['power_on_interval_05'] = ([0xd6, 0x04, 0xff, 0xff, 0x01, 0xd3], 5, 1)
+line['change_address_code'] = ([0x05, 0x00, 0xff, 0xff, 0x00, 0x04], 5, 1)
+line['modify_password'] = ([0xd3, 0x07, 0xff, 0xff, 0x11, 0x11, 0x11, 0x11], 5, 1)
+
+def num_to_hex(number):
+  """ Number to internal hexadecimal representation """
+  if number == None:
+    return []
+  length = len(hex(number)[2:]) + (len(hex(number)[2:]) % 2)
+  return [hex(number)[2:].zfill(length)[x:x+2] for x in range(0,length,2)]
+
+
+
+def send_command(action, argument=None):
+  """ Send CRC computed command to serial device and wait for response """
+  (command, response_size, timeout) = line[action]
+  command = command[0:1] + num_to_hex(opt_address_code) + num_to_hex(argument)
+  serial.timeout = timeout
+  raw_command = command + [make_checksum(command)]
+  return send_raw_command(raw_command, response_size)
+
+
+def action_login():
+  """ Login to device """
+  return send_command('login', opt_password)
+
+def action_status():
+  """ Get port status from device """
+  action_login()
+  return send_command('status')
+
+def action_port_on(port):
+  """ Enable port on device """
+  action_login()
+  port_status_synced = False
+  return send_command('port_on', port)
+
+def action_port_off(port):
+  """ Disable port on device """
+  action_login()
+  port_status_synced = False
+  return send_command('port_off', port)
+
+
+def get_ports_status():
+   # Example port 1 is off
+   # d1 28 ff ff fe ff ff
+   #             ^^ ^^ ^^
+   # TODO: Implement Ampere monitoring
+   #[01:38:14] Send: 55 07 ff ff 12 34 56 78 5a ff ff d2 00
+   #[01:38:14] Recv: ff ff a9 d1 28 ff ff 00 07 00 11 00 00 00 00 00 00 00 00
+   #                 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 
+   #                 28 00 28 00 00 10 ff
+   #[01:38:14] Comm: Refresh
+   port_array = [False] * 25
+
+   retval = send_command('status')
+   status_array = bin_reverse((bin_reverse(retval[4]) << 16) ^ (bin_reverse(retval[5]) << 8) ^ bin_reverse(retval[6]),24)
+   for port in range(0,24):
+      if (status_array & (1 << port)) > 0:
+        port_array[port+1] = True
+
+   # Update global state
+   ports_status = port_array
+   port_status_synced = True 
+   return port_array
+
+
+def get_port_status(port):
+   """ Get specific port status """
+   if port_status_synced:
+     return ports_status[port]
+   else:
+     return get_ports_status()[port]
+
+
   
-ser.close()
-exit(0)
-
-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 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)
-
-    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)
 
 
@@ -257 +254 @@
 
 
-def usage():
-    print """
+
+def usage(msg="",exitcode=None):
+    print """%s
 Usage %s arguments
 Version: $Id: npm4000.py 750 2009-09-27 14:34:55Z rick $
@@ -264 +262 @@
 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
+  [-d|--debug]                  Print extra communication output
+  --serialport=                 Serial Port to connect to [%s]
+  --password=                   Password to use in hex notation [%s]
+  --addresscode=                Internal device number in hex notation [%s]
+  [-s <port>|--status=<port>]   Current port(s) configuration
   [-t <port>|--toggle=<port>]   Toggle port(s)
   [-o <port>|--on=<port>]       Turn on port(s)
   [-f <port>|--off=<port>]      Turn off port(s)
+  --allon                       Turn all ports on using internal safety [TODO: Implement]
+  --alloff                      Turn all ports off using internal safety [TODO: Implement]
+  --changepassword=             Change password [TODO: Implement]
+  --changeaddresscode=          Change addresscode [TODO: Implement]
+  --changetimerdelay=           Change internal timer delay [TODO: Implement]
+  --delay=                      Delay used between port operations [TODO: Implement]
+  --baudrate=                   Bautrate used for communication (19200,9600) [%s]
 
 Note: <port> has different notations:
@@ -278 +282 @@
   Actual value of port          A1,..,A8,B1,..,B8,C1,..,C8 
   All ports                     all
-    """ % (sys.argv[0], inet_addr, inet_port)
+    """ % (msg, sys.argv[0], opt_serial_port, hex(opt_password), hex(opt_address_code), opt_baudrate)
+    if exitcode:
+        sys.exit(exitcode)   
 
 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="])
+            "df:s:t:o:v", ["debug","verbose", "serialport=", "password=", "addresscode=","toggle=","off=", "on=", "status=", "buadrate="])
     except getopt.GetoptError, err:
-        # print help information and exit:
-        print str(err) # will print something like "option -a not recognized"
-        usage()
-        sys.exit(2)
+        usage(str(err),2)
 
     opt_port = None
     opt_action = None
     for o, a in opts:
-        if o in ("-v", "--verbose"):
-            verbose = True
+        if o in ("-d", "--debug"):
+            DEBUG = True
         elif o in ("-h", "--help"):
-            usage()
-            sys.exit()
+            usage("",0)
         elif o in ("--addresscode"):
-            addr_code = int(a,16)
-        elif o in ("--host"):
-            inet_addr = a
+            opt_address_code = int(a,16)
         elif o in ("--password"):
-            passwd = int(a,16)
-        elif o in ("--port"):
-            inet_port = a
+            opt_passwd = int(a,16)
+        elif o in ("--buadrate"):
+            opt_baudrate = a
+        elif o in ("--serialport"):
+            opt_serial_port = a
         elif o in ("-s", "--status"):
             opt_action = "status"
@@ -323 +324 @@
 
     if (opt_port == None or opt_action == None):
-        usage()
-        sys.exit(2)
-
-    dprint ('action: ' + opt_action + ' port: ' + opt_port) 
+        usage("[ERROR] No port or action defined",2)
+
+
+    # Resolve port to proper numbers array
+    ports = []
+    for port in opt_port.split(','):
+      if port == "all":
+          ports.extend(range(1,25))
+      elif port[0] in ("A","B","C"):
+          ports.extend(int(port,16))
+      else:
+          ports.extend(int(port))
+    debug("Operating on ports " + ",".join(port))
+
+    serial = serial.Serial(opt_serial_port, opt_baudrate, timeout=5)
+    debug(serial)
+      
 
     # Status needs real integers, hack
-    if opt_action == "status":
-        if opt_port == "all":
-            getStatusAll()
+    for port in ports:
+      if opt_action == "status":
+        action_status(port)
+      elif opt_action == "toggle":
+        if get_port_status(port):
+          action_port_off(port)
         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)
-
+          action_port_on(port)
+      elif opt_action == "on":
+          action_port_on(port)
+      elif opt_action == "off":
+          action_port_off(port)
+      else:
+          assert False, "Option '%s' invalid" % opt_action
+
+    serial.close()
 
 if __name__ == "__main__":