ipm.py

#!/usr/bin/env python
# coding: utf-8
# vim: set ts=4 sts=4 sw=4 si ai et ft=python:
# author: Paul Tötterman <paul.totterman@zenrobotics.com>
#
# 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()