mcclory/dli-power-snippet.py

## dli-power-snippet.py
# 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)
	# 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)