ptman · September 13, 2013 13:53 · Sep 13, 2013
diff --git a/ipm.py b/ipm.py
@@ -0,0 +1,315 @@
+#!/usr/bin/env python
+# coding: utf-8
+# vim: set ts=4 sts=4 sw=4 si ai et ft=python:
+# author: Paul Tötterman <[email protected]>
+#
+# Copyright (c) 2013, ZenRobotics Ltd.
+#
+# Permission to use, copy, modify, and/or distribute this software for any
+# purpose with or without fee is hereby granted, provided that the above
+# copyright notice and this permission notice appear in all copies.
+#
+# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+
+"""Planet IP Power Manager Command Line Interface.
+
+Supports model IPM-1200x.
+
+Usage:
+    ipm [-dPqv] [-H HOST] [-u USER] [-p PASS] serial
+    ipm [-dPqv] [-H HOST] [-u USER] [-p PASS] status
+    ipm [-dPqv] [-H HOST] [-u USER] [-p PASS] on OUTLETS
+    ipm [-dPqv] [-H HOST] [-u USER] [-p PASS] off OUTLETS
+    ipm [-dPqv] [-H HOST] [-u USER] [-p PASS] cycle OUTLETS
+    ipm [-dPqv] [-H HOST] [-u USER] [-p PASS] name OUTLET NAME
+    ipm [-dPqv] [-H HOST] [-u USER] [-p PASS] location OUTLET LOC
+    ipm -h
+    ipm -V
+
+Options:
+    -h --help               Show this usage message.
+    -V --version            Show version.
+    -d --debug              Debug output.
+    -v --verbose            Verbose output.
+    -q --quiet              Less output.
+    -H --host HOST          Specify hostname [default: ipm].
+    -u --user USER          Specify username.
+    -p --pass PASS          Specify password.
+    -P --prompt-password    Prompt for password.
+
+Commands:
+    serial              Get serial number of IPM.
+    status              Get status of outlets from IPM.
+    on OUTLETS          Turn on outlets.
+    off OUTLETS         Turn off outlets.
+    cycle OUTLETS       Power cycle outlets.
+    name OUTLET NAME    Change the name of an outlet.
+    location OUTLET LOC Change the location of an outlet.
+
+Note: Outlets are specific by lower-case letters. Several can be specified by
+      separating with commas (without whitespace) or the keyword 'all'.
+"""
+
+import BeautifulSoup
+import collections
+import docopt
+import getpass
+import logging
+import pprint
+import re
+import requests
+import sys
+
+
+class IPM12(object):
+    """Remotely manage a Planet IPM-1200x device."""
+    ACTION_NONE = '1'
+    ACTION_ON = '2'
+    ACTION_OFF = '3'
+    ACTION_CYCLE = '4'
+    ALLON = {'XAAAAAAABADMM': '0'}
+    ALLOFF = {'XAAAAAAABADMM': '4095'}
+    OUTLET_ACTIONS = {'a': 'XAAAAAAABOEBA',
+                      'b': 'XAAAAAAABOEBB',
+                      'c': 'XAAAAAAABOEBC',
+                      'd': 'XAAAAAAABOEBD',
+                      'e': 'XAAAAAAABOEBE',
+                      'f': 'XAAAAAAABOEBF',
+                      'g': 'XAAAAAAABOEBG',
+                      'h': 'XAAAAAAABOEBH',
+                      'i': 'XAAAAAAABOEBI',
+                      'j': 'XAAAAAAABOEBJ',
+                      'k': 'XAAAAAAABOEBK',
+                      'l': 'XAAAAAAABOEBL'}
+    OUTLET_NAMES = {'a': 'XAAAAAAABADJB',
+                    'b': 'XAAAAAAABADJC',
+                    'c': 'XAAAAAAABADJD',
+                    'd': 'XAAAAAAABADJE',
+                    'e': 'XAAAAAAABADJF',
+                    'f': 'XAAAAAAABADJG',
+                    'g': 'XAAAAAAABADJH',
+                    'h': 'XAAAAAAABADJI',
+                    'i': 'XAAAAAAABADJJ',
+                    'j': 'XAAAAAAABADJK',
+                    'k': 'XAAAAAAABADJL',
+                    'l': 'XAAAAAAABADJM'}
+    OUTLET_LOCS = {'a': 'XAAAAAAABADKB',
+                   'b': 'XAAAAAAABADKC',
+                   'c': 'XAAAAAAABADKD',
+                   'd': 'XAAAAAAABADKE',
+                   'e': 'XAAAAAAABADKF',
+                   'f': 'XAAAAAAABADKG',
+                   'g': 'XAAAAAAABADKH',
+                   'h': 'XAAAAAAABADKI',
+                   'i': 'XAAAAAAABADKJ',
+                   'j': 'XAAAAAAABADKK',
+                   'k': 'XAAAAAAABADKL',
+                   'l': 'XAAAAAAABADKM'}
+    STATUS_RE = re.compile(r'var\s+Power_status\s+= new Array\((.*?)\);')
+    URL_CONTROL = 'PageControl.htm'
+    URL_LOGO = 'logo.js'
+    URL_OUTLET = 'Outlet.js'
+    VERSION_RE = re.compile(r'IPM System v1.00 \(SN (\w+)\)')
+
+    def __init__(self, hostname='ipm', username='zrripm', password='zrripm'):
+        self.hostname = hostname
+        self.username = username
+        self.password = password
+        self.baseurl = 'http://%s' % hostname
+        self.auth = (username, password)
+        self.supported = None
+        self.serial = None
+
+    def _get(self, url):
+        """GET request with a few assumtions."""
+        # pylint: disable-msg=E1103
+        url = '%s/%s' % (self.baseurl, url)
+        try:
+            resp = requests.get(url, auth=self.auth)
+        except requests.exceptions.ConnectionError:
+            logging.error('Could not connect to IPM')
+            return False
+
+        if resp.status_code == 200:
+            logging.debug('Successful GET')
+            return resp.content
+        elif resp.status_code == 401:
+            logging.error('Invalid credentials')
+        return False
+
+    def _post(self, url, data=None):
+        """POST request with a few assumtions."""
+        # pylint: disable-msg=E1103
+        url = '%s/%s' % (self.baseurl, url)
+        try:
+            resp = requests.post(url, auth=self.auth, data=data)
+        except requests.exceptions.ConnectionError:
+            logging.error('Could not connect to IPM')
+            return False
+
+        if resp.status_code == 200:
+            return True
+        elif resp.status_code == 401:
+            logging.error('Invalid credentials')
+        return False
+
+    def check_supported(self):
+        """Check if the IPM is of the expected version."""
+        content = self._get(IPM12.URL_LOGO)
+
+        if content is False:
+            self.supported = False
+            return self.supported
+
+        match = IPM12.VERSION_RE.search(content)
+        if not match:
+            self.supported = False
+            return self.supported
+
+        self.serial = match.group(1)
+        self.supported = True
+
+        return self.supported
+
+    def power_on_all(self):
+        """Power on all outlets."""
+        return self._post(IPM12.URL_CONTROL, data=IPM12.ALLON)
+
+    def power_off_all(self):
+        """Power off all outlets."""
+        return self._post(IPM12.URL_CONTROL, data=IPM12.ALLOFF)
+
+    def power_on(self, outlets):
+        """Power on a set of outlets or 'all'."""
+        if outlets == 'all':
+            return self.power_on_all()
+
+        data = {}
+        for outlet in outlets:
+            data[IPM12.OUTLET_ACTIONS[outlet]] = IPM12.ACTION_ON
+
+        return self._post(IPM12.URL_CONTROL, data=data)
+
+    def power_off(self, outlets):
+        """Power off a set of outlets or 'all'."""
+        if outlets == 'all':
+            return self.power_off_all()
+
+        data = {}
+        for outlet in outlets:
+            data[IPM12.OUTLET_ACTIONS[outlet]] = IPM12.ACTION_OFF
+
+        return self._post(IPM12.URL_CONTROL, data=data)
+
+    def power_cycle(self, outlets):
+        """Power cycle a set of outlets or 'all'."""
+        if outlets == 'all':
+            outlets = normalize_outlets('a,b,c,d,e,f,g,h,i,j,k,l')
+
+        data = {}
+        for outlet in outlets:
+            data[IPM12.OUTLET_ACTIONS[outlet]] = IPM12.ACTION_CYCLE
+
+        return self._post(IPM12.URL_CONTROL, data=data)
+
+    def set_name(self, outlet, name):
+        """Set the name of an outlet."""
+        data = {IPM12.OUTLET_NAMES[outlet]: name}
+        return self._post(IPM12.URL_CONTROL, data=data)
+
+    def set_location(self, outlet, location):
+        """Set the location -field of an outlet."""
+        data = {IPM12.OUTLET_LOCS[outlet]: location}
+        return self._post(IPM12.URL_CONTROL, data=data)
+
+    def status(self):
+        """Return status of IPM as dict."""
+        soup = BeautifulSoup.BeautifulSoup(self._get(IPM12.URL_CONTROL))
+        outlets = collections.defaultdict(dict)
+
+        for outlet, oid in IPM12.OUTLET_NAMES.iteritems():
+            tag = soup.find('input', attrs={'name': oid})
+            if tag is not None:
+                outlets[outlet]['name'] = tag.get('value')
+        for outlet, oid in IPM12.OUTLET_LOCS.iteritems():
+            tag = soup.find('input', attrs={'name': oid})
+            if tag is not None:
+                outlets[outlet]['location'] = tag.get('value')
+
+        power = IPM12.STATUS_RE.search(self._get(IPM12.URL_OUTLET)).group(1)
+        power = [x.strip('"') for x in power.split(',')]
+        for i, k in enumerate(sorted(IPM12.OUTLET_NAMES.keys())):
+            outlets[k]['status'] = power[i]
+
+        return dict(outlets)
+
+
+def normalize_outlets(outlets):
+    """Return list of outlets, lower-cased, or 'all'."""
+    if outlets.lower() == 'all':
+        return 'all'
+    if ',' in outlets:
+        return [x.lower() for x in outlets.split(',')]
+    return [outlets.lower()]
+
+
+def main():
+    """Main function."""
+    opts = docopt.docopt(__doc__, version='0.1.0')
+
+    if opts['--debug']:
+        logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
+    elif opts['--verbose']:
+        logging.basicConfig(stream=sys.stderr, level=logging.INFO)
+    elif opts['--quiet']:
+        logging.basicConfig(stream=sys.stderr, level=logging.ERROR)
+    else:
+        logging.basicConfig(stream=sys.stderr, level=logging.WARNING)
+
+    if opts['--prompt-password']:
+        opts['--pass'] = getpass.getpass()
+
+    ipm = IPM12(hostname=opts['--host'], username=opts['--user'],
+                password=opts['--pass'])
+
+    if not ipm.check_supported():
+        logging.error('Unsupported device')
+        sys.exit(2)
+
+    if opts['serial']:
+        print ipm.serial
+    elif opts['status']:
+        pprint.pprint(ipm.status())
+    elif opts['on']:
+        if not ipm.power_on(normalize_outlets(opts['OUTLETS'])):
+            logging.error('Problem powering on')
+            sys.exit(4)
+    elif opts['off']:
+        if not ipm.power_off(normalize_outlets(opts['OUTLETS'])):
+            logging.error('Problem powering off')
+            sys.exit(4)
+    elif opts['cycle']:
+        if not ipm.power_cycle(normalize_outlets(opts['OUTLETS'])):
+            logging.error('Problem cycling power')
+            sys.exit(4)
+    elif opts['name']:
+        if not ipm.set_name(opts['OUTLET'].lower(), opts['NAME']):
+            logging.error('Problem naming outlet')
+            sys.exit(4)
+    elif opts['location']:
+        if not ipm.set_location(opts['OUTLET'].lower(), opts['LOC']):
+            logging.error('Problem changing location for outlet')
+            sys.exit(4)
+    else:
+        logging.critical('Error, should never be reachable')
+        sys.exit(5)
+
+
+if __name__ == '__main__':
+    main()