Skip to content

Instantly share code, notes, and snippets.

@ptman

ptman/ipm.py

Created September 13, 2013 13:53
Show Gist options
  • Save ptman/6551021 to your computer and use it in GitHub Desktop.
Save ptman/6551021 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
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 <[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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment