Skip to content

Instantly share code, notes, and snippets.

@mcclory
Created August 3, 2017 14:56
Show Gist options
  • Save mcclory/1d123c829ca7561c660ab98a8f8b6a6b to your computer and use it in GitHub Desktop.
Save mcclory/1d123c829ca7561c660ab98a8f8b6a6b to your computer and use it in GitHub Desktop.
Ubuntu MAAS dli.py power driver status edits
# Copyright 2015-2016 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""DLI Power Driver."""
__all__ = []
import re
from time import sleep
from provisioningserver.drivers import (
make_ip_extractor,
make_setting_field,
SETTING_SCOPE,
)
from provisioningserver.drivers.power import (
PowerActionError,
PowerDriver,
PowerError,
)
from provisioningserver.utils import shell
from provisioningserver.utils.shell import (
call_and_check,
ExternalProcessError,
select_c_utf8_locale,
)
class DLIPowerDriver(PowerDriver):
name = 'dli'
description = "Digital Loggers, Inc. PDU"
settings = [
make_setting_field(
'outlet_id', "Outlet ID", scope=SETTING_SCOPE.NODE,
required=True),
make_setting_field('power_address', "Power address", required=True),
make_setting_field('power_user', "Power user"),
make_setting_field(
'power_pass', "Power password", field_type='password'),
]
ip_extractor = make_ip_extractor('power_address')
queryable = False
def detect_missing_packages(self):
if not shell.has_command_available('wget'):
return ['wget']
return []
def _set_outlet_state(
self, power_change, outlet_id=None, power_user=None,
power_pass=None, power_address=None, **extra):
"""Power DLI outlet ON/OFF."""
try:
url = 'http://%s:%s@%s/outlet?%s=%s' % (
power_user, power_pass, power_address, outlet_id, power_change)
# --auth-no-challenge: send Basic HTTP authentication
# information without first waiting for the server's challenge.
call_and_check(
['wget', '--auth-no-challenge', '-O', '/dev/null', url],
env=select_c_utf8_locale())
except ExternalProcessError as e:
raise PowerActionError(
"Failed to power %s outlet %s: %s" % (
power_change, outlet_id, e.output_as_unicode))
def _query_outlet_state(
self, outlet_id=None, power_user=None,
power_pass=None, power_address=None, **extra):
"""Query DLI outlet power state.
Sample snippet of query output from DLI:
...
<!--
function reg() {
window.open('http://www.digital-loggers.com/reg.html?SN=LPC751740');
}
//-->
</script>
</head>
<!-- state=02 lock=00 -->
<body alink="#0000FF" vlink="#0000FF">
<FONT FACE="Arial, Helvetica, Sans-Serif">
...
"""
try:
url = 'http://%s:%s@%s/index.htm' % (
power_user, power_pass, power_address)
# --auth-no-challenge: send Basic HTTP authentication
# information without first waiting for the server's challenge.
wget_output = call_and_check(
['wget', '--auth-no-challenge', '-qO-', url],
env=select_c_utf8_locale())
wget_output = wget_output.decode('utf-8')
match = re.search("<!-- state=([0-9a-fA-F]+)", wget_output)
if match is None:
raise PowerError(
"Unable to extract power state for outlet %s from "
"wget output: %s" % (outlet_id, wget_output))
else:
''' original implementation in maas
state = match.group(1)
# state is a bitmap of the DLI's oulet states, where bit 0
# corresponds to oulet 1's power state, bit 1 corresponds to
# outlet 2's power state, etc., encoded as hexadecimal.
if (int(state, 16) & (1 << int(outlet_id) - 1)) > 0:
return 'on'
else:
return 'off'
'''
status = list(eval("bin(0x%02d)[2:]" % (int(match.group(1)))))
while len(status) < 8: # pad array back to 8... but inverted so pad the front
status.insert(0, '0')
status.reverse() # per observation, this is just inverted, so reverse the array
if int(status[int(outlet_id) -1]) > 0:
return 'on'
else:
return: 'off'
except ExternalProcessError as e:
raise PowerActionError(
"Failed to power query outlet %s: %s" % (
outlet_id, e.output_as_unicode))
def power_on(self, system_id, context):
"""Power on DLI outlet."""
# Power off the outlet if it is currently on
if self._query_outlet_state(**context) == 'on':
self._set_outlet_state('OFF', **context)
sleep(1)
if self._query_outlet_state(**context) != 'off':
raise PowerError(
"Unable to power off outlet %s that is already on."
% context['outlet_id'])
self._set_outlet_state('ON', **context)
def power_off(self, system_id, context):
"""Power off DLI outlet."""
self._set_outlet_state('OFF', **context)
def power_query(self, system_id, context):
"""Power query DLI outlet."""
return self._query_outlet_state(**context)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment