libraries

Package: phue
File: phue-1.1/PKG-INFO

Metadata-Version: 1.0
Name: phue
Version: 1.1
Summary: A Philips Hue Python library
Home-page: https://github.com/studioimaginaire/phue
Author: Nathanaël Lécaudé
Author-email: UNKNOWN
License: MIT
Description: UNKNOWN
Platform: UNKNOWN



Package: phue
File: phue-1.1/setup.py

# -*- coding: utf-8 -*-

from phue import __version__
from distutils.core import setup

setup(
    name = 'phue',
    version = __version__,
    description = 'A Philips Hue Python library',
    author = 'Nathanaël Lécaudé',
    license='MIT',
    url = 'https://github.com/studioimaginaire/phue',
    py_modules=['phue'],
)



Package: phue
File: phue-1.1/phue.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

'''
phue by Nathanaël Lécaudé - A Philips Hue Python library
Contributions by Marshall Perrin, Justin Lintz
https://github.com/studioimaginaire/phue
Original protocol hacking by rsmck : http://rsmck.co.uk/hue

Published under the MIT license - See LICENSE file for more details.

"Hue Personal Wireless Lighting" is a trademark owned by Koninklijke Philips Electronics N.V., see www.meethue.com for more information.
I am in no way affiliated with the Philips organization.

'''

import json
import logging
import os
import platform
import sys
import socket
if sys.version_info[0] > 2:
    PY3K = True
else:
    PY3K = False

if PY3K:
    import http.client as httplib
else:
    import httplib

logger = logging.getLogger('phue')


if platform.system() == 'Windows':
    USER_HOME = 'USERPROFILE'
else:
    USER_HOME = 'HOME'

__version__ = '1.1'


def is_string(data):
    """Utility method to see if data is a string."""
    if PY3K:
        return isinstance(data, str)
    else:
        return isinstance(data, str) or isinstance(data, unicode)  # noqa

def encodeString(string):
    """Utility method to encode strings as utf-8."""
    if PY3K:
        return string
    else:
        return string.encode('utf-8')

def decodeString(string):
    """Utility method to decode strings as utf-8."""
    if PY3K:
        return string
    else:
        return string.decode('utf-8')

class PhueException(Exception):

    def __init__(self, id, message):
        self.id = id
        self.message = message


class PhueRegistrationException(PhueException):
    pass


class PhueRequestTimeout(PhueException):
    pass


class Light(object):

    """ Hue Light object

    Light settings can be accessed or set via the properties of this object.

    """
    def __init__(self, bridge, light_id):
        self.bridge = bridge
        self.light_id = light_id

        self._name = None
        self._on = None
        self._brightness = None
        self._colormode = None
        self._hue = None
        self._saturation = None
        self._xy = None
        self._colortemp = None
        self._effect = None
        self._alert = None
        self.transitiontime = None  # default
        self._reset_bri_after_on = None
        self._reachable = None
        self._type = None

    def __repr__(self):
        # like default python repr function, but add light name
        return '<{0}.{1} object "{2}" at {3}>'.format(
            self.__class__.__module__,
            self.__class__.__name__,
            self.name,
            hex(id(self)))

    # Wrapper functions for get/set through the bridge, adding support for
    # remembering the transitiontime parameter if the user has set it
    def _get(self, *args, **kwargs):
        return self.bridge.get_light(self.light_id, *args, **kwargs)

    def _set(self, *args, **kwargs):

        if self.transitiontime is not None:
            kwargs['transitiontime'] = self.transitiontime
            logger.debug("Setting with transitiontime = {0} ds = {1} s".format(
                self.transitiontime, float(self.transitiontime) / 10))

            if (args[0] == 'on' and args[1] is False) or (
                    kwargs.get('on', True) is False):
                self._reset_bri_after_on = True
        return self.bridge.set_light(self.light_id, *args, **kwargs)

    @property
    def name(self):
        '''Get or set the name of the light [string]'''
        return encodeString(self._get('name'))

    @name.setter
    def name(self, value):
        old_name = self.name
        self._name = value
        self._set('name', self._name)

        logger.debug("Renaming light from '{0}' to '{1}'".format(
            old_name, value))

        self.bridge.lights_by_name[self.name] = self
        del self.bridge.lights_by_name[old_name]

    @property
    def on(self):
        '''Get or set the state of the light [True|False]'''
        self._on = self._get('on')
        return self._on

    @on.setter
    def on(self, value):

        # Some added code here to work around known bug where
        # turning off with transitiontime set makes it restart on brightness = 1
        # see
        # http://www.everyhue.com/vanilla/discussion/204/bug-with-brightness-when-requesting-ontrue-transitiontime5

        # if we're turning off, save whether this bug in the hardware has been
        # invoked
        if self._on and value is False:
            self._reset_bri_after_on = self.transitiontime is not None
            if self._reset_bri_after_on:
                logger.warning(
                    'Turned off light with transitiontime specified, brightness will be reset on power on')

        self._set('on', value)

        # work around bug by resetting brightness after a power on
        if self._on is False and value is True:
            if self._reset_bri_after_on:
                logger.warning(
                    'Light was turned off with transitiontime specified, brightness needs to be reset now.')
                self.brightness = self._brightness
                self._reset_bri_after_on = False

        self._on = value

    @property
    def colormode(self):
        '''Get the color mode of the light [hs|xy|ct]'''
        self._colormode = self._get('colormode')
        return self._colormode

    @property
    def brightness(self):
        '''Get or set the brightness of the light [0-254].

        0 is not off'''

        self._brightness = self._get('bri')
        return self._brightness

    @brightness.setter
    def brightness(self, value):
        self._brightness = value
        self._set('bri', self._brightness)

    @property
    def hue(self):
        '''Get or set the hue of the light [0-65535]'''
        self._hue = self._get('hue')
        return self._hue

    @hue.setter
    def hue(self, value):
        self._hue = int(value)
        self._set('hue', self._hue)

    @property
    def saturation(self):
        '''Get or set the saturation of the light [0-254]

        0 = white
        254 = most saturated
        '''
        self._saturation = self._get('sat')
        return self._saturation

    @saturation.setter
    def saturation(self, value):
        self._saturation = value
        self._set('sat', self._saturation)

    @property
    def xy(self):
        '''Get or set the color coordinates of the light [ [0.0-1.0, 0.0-1.0] ]

        This is in a color space similar to CIE 1931 (but not quite identical)
        '''
        self._xy = self._get('xy')
        return self._xy

    @xy.setter
    def xy(self, value):
        self._xy = value
        self._set('xy', self._xy)

    @property
    def colortemp(self):
        '''Get or set the color temperature of the light, in units of mireds [154-500]'''
        self._colortemp = self._get('ct')
        return self._colortemp

    @colortemp.setter
    def colortemp(self, value):
        if value < 154:
            logger.warn('154 mireds is coolest allowed color temp')
        elif value > 500:
            logger.warn('500 mireds is warmest allowed color temp')
        self._colortemp = value
        self._set('ct', self._colortemp)

    @property
    def colortemp_k(self):
        '''Get or set the color temperature of the light, in units of Kelvin [2000-6500]'''
        self._colortemp = self._get('ct')
        return int(round(1e6 / self._colortemp))

    @colortemp_k.setter
    def colortemp_k(self, value):
        if value > 6500:
            logger.warn('6500 K is max allowed color temp')
            value = 6500
        elif value < 2000:
            logger.warn('2000 K is min allowed color temp')
            value = 2000

        colortemp_mireds = int(round(1e6 / value))
        logger.debug("{0:d} K is {1} mireds".format(value, colortemp_mireds))
        self.colortemp = colortemp_mireds

    @property
    def effect(self):
        '''Check the effect setting of the light. [none|colorloop]'''
        self._effect = self._get('effect')
        return self._effect

    @effect.setter
    def effect(self, value):
        self._effect = value
        self._set('effect', self._effect)

    @property
    def alert(self):
        '''Get or set the alert state of the light [select|lselect|none]'''
        self._alert = self._get('alert')
        return self._alert

    @alert.setter
    def alert(self, value):
        if value is None:
            value = 'none'
        self._alert = value
        self._set('alert', self._alert)

    @property
    def reachable(self):
        '''Get the reachable state of the light [boolean]'''
        self._reachable = self._get('reachable')
        return self._reachable

    @property
    def type(self):
        '''Get the type of the light [string]'''
        self._type = self._get('type')
        return self._type


class SensorState(dict):
    def __init__(self, bridge, sensor_id):
        self._bridge = bridge
        self._sensor_id = sensor_id

    def __setitem__(self, key, value):
        dict.__setitem__(self, key, value)
        self._bridge.set_sensor_state(self._sensor_id, self)


class SensorConfig(dict):
    def __init__(self, bridge, sensor_id):
        self._bridge = bridge
        self._sensor_id = sensor_id

    def __setitem__(self, key, value):
        dict.__setitem__(self, key, value)
        self._bridge.set_sensor_config(self._sensor_id, self)


class Sensor(object):

    """ Hue Sensor object

    Sensor config and state can be read and updated via the properties of this object

    """
    def __init__(self, bridge, sensor_id):
        self.bridge = bridge
        self.sensor_id = sensor_id

        self._name = None
        self._model = None
        self._swversion = None
        self._type = None
        self._uniqueid = None
        self._manufacturername = None
        self._state = SensorState(bridge, sensor_id)
        self._config = {}
        self._recycle = None

    def __repr__(self):
        # like default python repr function, but add sensor name
        return '<{0}.{1} object "{2}" at {3}>'.format(
            self.__class__.__module__,
            self.__class__.__name__,
            self.name,
            hex(id(self)))

    # Wrapper functions for get/set through the bridge
    def _get(self, *args, **kwargs):
        return self.bridge.get_sensor(self.sensor_id, *args, **kwargs)

    def _set(self, *args, **kwargs):
        return self.bridge.set_sensor(self.sensor_id, *args, **kwargs)

    @property
    def name(self):
        '''Get or set the name of the sensor [string]'''
        return encodeString(self._get('name'))

    @name.setter
    def name(self, value):
        old_name = self.name
        self._name = value
        self._set('name', self._name)

        logger.debug("Renaming sensor from '{0}' to '{1}'".format(
            old_name, value))

        self.bridge.sensors_by_name[self.name] = self
        del self.bridge.sensors_by_name[old_name]

    @property
    def modelid(self):
        '''Get a unique identifier of the hardware model of this sensor [string]'''
        self._modelid = self._get('modelid')
        return self._modelid

    @property
    def swversion(self):
        '''Get the software version identifier of the sensor's firmware [string]'''
        self._swversion = self._get('swversion')
        return self._swversion

    @property
    def type(self):
        '''Get the sensor type of this device [string]'''
        self._type = self._get('type')
        return self._type

    @property
    def uniqueid(self):
        '''Get the unique device ID of this sensor [string]'''
        self._uniqueid = self._get('uniqueid')
        return self._uniqueid

    @property
    def manufacturername(self):
        '''Get the name of the manufacturer [string]'''
        self._manufacturername = self._get('manufacturername')
        return self._manufacturername

    @property
    def state(self):
        ''' A dictionary of sensor state. Some values can be updated, some are read-only. [dict]'''
        data = self._get('state')
        self._state.clear()
        self._state.update(data)
        return self._state

    @state.setter
    def state(self, data):
        self._state.clear()
        self._state.update(data)

    @property
    def config(self):
        ''' A dictionary of sensor config. Some values can be updated, some are read-only. [dict]'''
        data = self._get('config')
        self._config.clear()
        self._config.update(data)
        return self._config

    @config.setter
    def config(self, data):
        self._config.clear()
        self._config.update(data)

    @property
    def recycle(self):
        ''' True if this resource should be automatically removed when the last reference to it disappears [bool]'''
        self._recycle = self._get('manufacturername')
        return self._manufacturername


class Group(Light):

    """ A group of Hue lights, tracked as a group on the bridge

    Example:

        >>> b = Bridge()
        >>> g1 = Group(b, 1)
        >>> g1.hue = 50000 # all lights in that group turn blue
        >>> g1.on = False # all will turn off

        >>> g2 = Group(b, 'Kitchen')  # you can also look up groups by name
        >>> # will raise a LookupError if the name doesn't match

    """

    def __init__(self, bridge, group_id):
        Light.__init__(self, bridge, None)
        del self.light_id  # not relevant for a group

        try:
            self.group_id = int(group_id)
        except:
            name = group_id
            groups = bridge.get_group()
            for idnumber, info in groups.items():
                if info['name'] == decodeString(name):
                    self.group_id = int(idnumber)
                    break
            else:
                raise LookupError("Could not find a group by that name.")

    # Wrapper functions for get/set through the bridge, adding support for
    # remembering the transitiontime parameter if the user has set it
    def _get(self, *args, **kwargs):
        return self.bridge.get_group(self.group_id, *args, **kwargs)

    def _set(self, *args, **kwargs):
        # let's get basic group functionality working first before adding
        # transition time...
        if self.transitiontime is not None:
            kwargs['transitiontime'] = self.transitiontime
            logger.debug("Setting with transitiontime = {0} ds = {1} s".format(
                self.transitiontime, float(self.transitiontime) / 10))

            if (args[0] == 'on' and args[1] is False) or (
                    kwargs.get('on', True) is False):
                self._reset_bri_after_on = True
        return self.bridge.set_group(self.group_id, *args, **kwargs)

    @property
    def name(self):
        '''Get or set the name of the light group [string]'''
        return encodeString(self._get('name'))

    @name.setter
    def name(self, value):
        old_name = self.name
        self._name = value
        logger.debug("Renaming light group from '{0}' to '{1}'".format(
            old_name, value))
        self._set('name', self._name)

    @property
    def lights(self):
        """ Return a list of all lights in this group"""
        # response = self.bridge.request('GET', '/api/{0}/groups/{1}'.format(self.bridge.username, self.group_id))
        # return [Light(self.bridge, int(l)) for l in response['lights']]
        return [Light(self.bridge, int(l)) for l in self._get('lights')]

    @lights.setter
    def lights(self, value):
        """ Change the lights that are in this group"""
        logger.debug("Setting lights in group {0} to {1}".format(
            self.group_id, str(value)))
        self._set('lights', value)


class AllLights(Group):

    """ All the Hue lights connected to your bridge

    This makes use of the semi-documented feature that
    "Group 0" of lights appears to be a group automatically
    consisting of all lights.  This is not returned by
    listing the groups, but is accessible if you explicitly
    ask for group 0.
    """
    def __init__(self, bridge=None):
        if bridge is None:
            bridge = Bridge()
        Group.__init__(self, bridge, 0)


class Scene(object):
    """ Container for Scene """

    def __init__(self, sid, appdata=None, lastupdated=None,
                 lights=None, locked=False, name="", owner="",
                 picture="", recycle=False, version=0, type="", group="",
                 *args, **kwargs):
        self.scene_id = sid
        self.appdata = appdata or {}
        self.lastupdated = lastupdated
        if lights is not None:
            self.lights = sorted([int(x) for x in lights])
        else:
            self.lights = []
        self.locked = locked
        self.name = encodeString(name)
        self.owner = owner
        self.picture = picture
        self.recycle = recycle
        self.version = version
        self.type = type
        self.group = group

    def __repr__(self):
        # like default python repr function, but add scene name
        return '<{0}.{1} id="{2}" name="{3}" lights={4}>'.format(
            self.__class__.__module__,
            self.__class__.__name__,
            self.scene_id,
            self.name,
            self.lights)


class Bridge(object):

    """ Interface to the Hue ZigBee bridge

    You can obtain Light objects by calling the get_light_objects method:

        >>> b = Bridge(ip='192.168.1.100')
        >>> b.get_light_objects()
        [<phue.Light at 0x10473d750>,
         <phue.Light at 0x1046ce110>]

    Or more succinctly just by accessing this Bridge object as a list or dict:

        >>> b[1]
        <phue.Light at 0x10473d750>
        >>> b['Kitchen']
        <phue.Light at 0x10473d750>



    """
    def __init__(self, ip=None, username=None, config_file_path=None):
        """ Initialization function.

        Parameters:
        ------------
        ip : string
            IP address as dotted quad
        username : string, optional

        """

        if config_file_path is not None:
            self.config_file_path = config_file_path
        elif os.getenv(USER_HOME) is not None and os.access(os.getenv(USER_HOME), os.W_OK):
            self.config_file_path = os.path.join(os.getenv(USER_HOME), '.python_hue')
        elif 'iPad' in platform.machine() or 'iPhone' in platform.machine() or 'iPad' in platform.machine():
            self.config_file_path = os.path.join(os.getenv(USER_HOME), 'Documents', '.python_hue')
        else:
            self.config_file_path = os.path.join(os.getcwd(), '.python_hue')

        self.ip = ip
        self.username = username
        self.lights_by_id = {}
        self.lights_by_name = {}
        self.sensors_by_id = {}
        self.sensors_by_name = {}
        self._name = None

        # self.minutes = 600 # these do not seem to be used anywhere?
        # self.seconds = 10

        self.connect()

    @property
    def name(self):
        '''Get or set the name of the bridge [string]'''
        self._name = self.request(
            'GET', '/api/' + self.username + '/config')['name']
        return self._name

    @name.setter
    def name(self, value):
        self._name = value
        data = {'name': self._name}
        self.request(
            'PUT', '/api/' + self.username + '/config', data)

    def request(self, mode='GET', address=None, data=None):
        """ Utility function for HTTP GET/PUT requests for the API"""
        connection = httplib.HTTPConnection(self.ip, timeout=10)

        try:
            if mode == 'GET' or mode == 'DELETE':
                connection.request(mode, address)
            if mode == 'PUT' or mode == 'POST':
                connection.request(mode, address, json.dumps(data))

            logger.debug("{0} {1} {2}".format(mode, address, str(data)))

        except socket.timeout:
            error = "{} Request to {}{} timed out.".format(mode, self.ip, address)

            logger.exception(error)
            raise PhueRequestTimeout(None, error)

        result = connection.getresponse()
        response = result.read()
        connection.close()
        if PY3K:
            response = response.decode('utf-8')

        logger.debug(response)
        return json.loads(response)

    def get_ip_address(self, set_result=False):

        """ Get the bridge ip address from the meethue.com nupnp api """

        connection = httplib.HTTPSConnection('www.meethue.com')
        connection.request('GET', '/api/nupnp')

        logger.info('Connecting to meethue.com/api/nupnp')

        result = connection.getresponse()

        if PY3K:
            data = json.loads(str(result.read(), encoding='utf-8'))
        else:
            result_str = result.read()
            data = json.loads(result_str)

        """ close connection after read() is done, to prevent issues with read() """

        connection.close()

        ip = str(data[0]['internalipaddress'])

        if ip is not '':
            if set_result:
                self.ip = ip

            return ip
        else:
            return False

    def register_app(self):
        """ Register this computer with the Hue bridge hardware and save the resulting access token """
        registration_request = {"devicetype": "python_hue"}
        response = self.request('POST', '/api', registration_request)
        for line in response:
            for key in line:
                if 'success' in key:
                    with open(self.config_file_path, 'w') as f:
                        logger.info(
                            'Writing configuration file to ' + self.config_file_path)
                        f.write(json.dumps({self.ip: line['success']}))
                        logger.info('Reconnecting to the bridge')
                    self.connect()
                if 'error' in key:
                    error_type = line['error']['type']
                    if error_type == 101:
                        raise PhueRegistrationException(error_type,
                                                        'The link button has not been pressed in the last 30 seconds.')
                    if error_type == 7:
                        raise PhueException(error_type,
                                            'Unknown username')

    def connect(self):
        """ Connect to the Hue bridge """
        logger.info('Attempting to connect to the bridge...')
        # If the ip and username were provided at class init
        if self.ip is not None and self.username is not None:
            logger.info('Using ip: ' + self.ip)
            logger.info('Using username: ' + self.username)
            return

        if self.ip is None or self.username is None:
            try:
                with open(self.config_file_path) as f:
                    config = json.loads(f.read())
                    if self.ip is None:
                        self.ip = list(config.keys())[0]
                        logger.info('Using ip from config: ' + self.ip)
                    else:
                        logger.info('Using ip: ' + self.ip)
                    if self.username is None:
                        self.username = config[self.ip]['username']
                        logger.info(
                            'Using username from config: ' + self.username)
                    else:
                        logger.info('Using username: ' + self.username)
            except Exception as e:
                logger.info(
                    'Error opening config file, will attempt bridge registration')
                self.register_app()

    def get_light_id_by_name(self, name):
        """ Lookup a light id based on string name. Case-sensitive. """
        lights = self.get_light()
        for light_id in lights:
            if decodeString(name) == lights[light_id]['name']:
                return light_id
        return False

    def get_light_objects(self, mode='list'):
        """Returns a collection containing the lights, either by name or id (use 'id' or 'name' as the mode)
        The returned collection can be either a list (default), or a dict.
        Set mode='id' for a dict by light ID, or mode='name' for a dict by light name.   """
        if self.lights_by_id == {}:
            lights = self.request('GET', '/api/' + self.username + '/lights/')
            for light in lights:
                self.lights_by_id[int(light)] = Light(self, int(light))
                self.lights_by_name[lights[light][
                    'name']] = self.lights_by_id[int(light)]
        if mode == 'id':
            return self.lights_by_id
        if mode == 'name':
            return self.lights_by_name
        if mode == 'list':
            # return ligts in sorted id order, dicts have no natural order
            return [self.lights_by_id[id] for id in sorted(self.lights_by_id)]

    def get_sensor_id_by_name(self, name):
        """ Lookup a sensor id based on string name. Case-sensitive. """
        sensors = self.get_sensor()
        for sensor_id in sensors:
            if decodeString(name) == sensors[sensor_id]['name']:
                return sensor_id
        return False

    def get_sensor_objects(self, mode='list'):
        """Returns a collection containing the sensors, either by name or id (use 'id' or 'name' as the mode)
        The returned collection can be either a list (default), or a dict.
        Set mode='id' for a dict by sensor ID, or mode='name' for a dict by sensor name.   """
        if self.sensors_by_id == {}:
            sensors = self.request('GET', '/api/' + self.username + '/sensors/')
            for sensor in sensors:
                self.sensors_by_id[int(sensor)] = Sensor(self, int(sensor))
                self.sensors_by_name[sensors[sensor][
                    'name']] = self.sensors_by_id[int(sensor)]
        if mode == 'id':
            return self.sensors_by_id
        if mode == 'name':
            return self.sensors_by_name
        if mode == 'list':
            return self.sensors_by_id.values()

    def __getitem__(self, key):
        """ Lights are accessibly by indexing the bridge either with
        an integer index or string name. """
        if self.lights_by_id == {}:
            self.get_light_objects()

        try:
            return self.lights_by_id[key]
        except:
            try:
                return self.lights_by_name[decodeString(key)]
            except:
                raise KeyError(
                    'Not a valid key (integer index starting with 1, or light name): ' + str(key))

    @property
    def lights(self):
        """ Access lights as a list """
        return self.get_light_objects()

    def get_api(self):
        """ Returns the full api dictionary """
        return self.request('GET', '/api/' + self.username)

    def get_light(self, light_id=None, parameter=None):
        """ Gets state by light_id and parameter"""

        if is_string(light_id):
            light_id = self.get_light_id_by_name(light_id)
        if light_id is None:
            return self.request('GET', '/api/' + self.username + '/lights/')
        state = self.request(
            'GET', '/api/' + self.username + '/lights/' + str(light_id))
        if parameter is None:
            return state
        if parameter in ['name', 'type', 'uniqueid', 'swversion']:
            return state[parameter]
        else:
            try:
                return state['state'][parameter]
            except KeyError as e:
                raise KeyError(
                    'Not a valid key, parameter %s is not associated with light %s)'
                    % (parameter, light_id))

    def set_light(self, light_id, parameter, value=None, transitiontime=None):
        """ Adjust properties of one or more lights.

        light_id can be a single lamp or an array of lamps
        parameters: 'on' : True|False , 'bri' : 0-254, 'sat' : 0-254, 'ct': 154-500

        transitiontime : in **deciseconds**, time for this transition to take place
                         Note that transitiontime only applies to *this* light
                         command, it is not saved as a setting for use in the future!
                         Use the Light class' transitiontime attribute if you want
                         persistent time settings.

        """
        if isinstance(parameter, dict):
            data = parameter
        else:
            data = {parameter: value}

        if transitiontime is not None:
            data['transitiontime'] = int(round(
                transitiontime))  # must be int for request format

        light_id_array = light_id
        if isinstance(light_id, int) or is_string(light_id):
            light_id_array = [light_id]
        result = []
        for light in light_id_array:
            logger.debug(str(data))
            if parameter == 'name':
                result.append(self.request('PUT', '/api/' + self.username + '/lights/' + str(
                    light_id), data))
            else:
                if is_string(light):
                    converted_light = self.get_light_id_by_name(light)
                else:
                    converted_light = light
                result.append(self.request('PUT', '/api/' + self.username + '/lights/' + str(
                    converted_light) + '/state', data))
            if 'error' in list(result[-1][0].keys()):
                logger.warn("ERROR: {0} for light {1}".format(
                    result[-1][0]['error']['description'], light))

        logger.debug(result)
        return result

    # Sensors #####

    @property
    def sensors(self):
        """ Access sensors as a list """
        return self.get_sensor_objects()

    def create_sensor(self, name, modelid, swversion, sensor_type, uniqueid, manufacturername, state={}, config={}, recycle=False):
        """ Create a new sensor in the bridge. Returns (ID,None) of the new sensor or (None,message) if creation failed. """
        data = {
            "name": name,
            "modelid": modelid,
            "swversion": swversion,
            "type": sensor_type,
            "uniqueid": uniqueid,
            "manufacturername": manufacturername,
            "recycle": recycle
        }
        if (isinstance(state, dict) and state != {}):
            data["state"] = state

        if (isinstance(config, dict) and config != {}):
            data["config"] = config

        result = self.request('POST', '/api/' + self.username + '/sensors/', data)

        if ("success" in result[0].keys()):
            new_id = result[0]["success"]["id"]
            logger.debug("Created sensor with ID " + new_id)
            new_sensor = Sensor(self, int(new_id))
            self.sensors_by_id[new_id] = new_sensor
            self.sensors_by_name[name] = new_sensor
            return new_id, None
        else:
            logger.debug("Failed to create sensor:" + repr(result[0]))
            return None, result[0]

    def get_sensor(self, sensor_id=None, parameter=None):
        """ Gets state by sensor_id and parameter"""

        if is_string(sensor_id):
            sensor_id = self.get_sensor_id_by_name(sensor_id)
        if sensor_id is None:
            return self.request('GET', '/api/' + self.username + '/sensors/')
        data = self.request(
            'GET', '/api/' + self.username + '/sensors/' + str(sensor_id))

        if isinstance(data, list):
            logger.debug("Unable to read sensor with ID {0}: {1}".format(sensor_id, repr(data)))
            return None

        if parameter is None:
            return data
        return data[parameter]

    def set_sensor(self, sensor_id, parameter, value=None):
        """ Adjust properties of a sensor

        sensor_id must be a single sensor.
        parameters: 'name' : string

        """
        if isinstance(parameter, dict):
            data = parameter
        else:
            data = {parameter: value}

        result = None
        logger.debug(str(data))
        result = self.request('PUT', '/api/' + self.username + '/sensors/' + str(
            sensor_id), data)
        if 'error' in list(result[0].keys()):
            logger.warn("ERROR: {0} for sensor {1}".format(
                result[0]['error']['description'], sensor_id))

        logger.debug(result)
        return result

    def set_sensor_state(self, sensor_id, parameter, value=None):
        """ Adjust the "state" object of a sensor

        sensor_id must be a single sensor.
        parameters: any parameter(s) present in the sensor's "state" dictionary.

        """
        self.set_sensor_content(sensor_id, parameter, value, "state")

    def set_sensor_config(self, sensor_id, parameter, value=None):
        """ Adjust the "config" object of a sensor

        sensor_id must be a single sensor.
        parameters: any parameter(s) present in the sensor's "config" dictionary.

        """
        self.set_sensor_content(sensor_id, parameter, value, "config")

    def set_sensor_content(self, sensor_id, parameter, value=None, structure="state"):
        """ Adjust the "state" or "config" structures of a sensor
        """
        if (structure != "state" and structure != "config"):
            logger.debug("set_sensor_current expects structure 'state' or 'config'.")
            return False

        if isinstance(parameter, dict):
            data = parameter.copy()
        else:
            data = {parameter: value}

        # Attempting to set this causes an error.
        if "lastupdated" in data:
            del data["lastupdated"]

        result = None
        logger.debug(str(data))
        result = self.request('PUT', '/api/' + self.username + '/sensors/' + str(
            sensor_id) + "/" + structure, data)
        if 'error' in list(result[0].keys()):
            logger.warn("ERROR: {0} for sensor {1}".format(
                result[0]['error']['description'], sensor_id))

        logger.debug(result)
        return result

    def delete_scene(self, scene_id):
        try:
            return self.request('DELETE', '/api/' + self.username + '/scenes/' + str(scene_id))
        except:
            logger.debug("Unable to delete scene with ID {0}".format(scene_id))

    def delete_sensor(self, sensor_id):
        try:
            name = self.sensors_by_id[sensor_id].name
            del self.sensors_by_name[name]
            del self.sensors_by_id[sensor_id]
            return self.request('DELETE', '/api/' + self.username + '/sensors/' + str(sensor_id))
        except:
            logger.debug("Unable to delete nonexistent sensor with ID {0}".format(sensor_id))

    # Groups of lights #####
    @property
    def groups(self):
        """ Access groups as a list """
        return [Group(self, int(groupid)) for groupid in self.get_group().keys()]

    def get_group_id_by_name(self, name):
        """ Lookup a group id based on string name. Case-sensitive. """
        groups = self.get_group()
        for group_id in groups:
            if decodeString(name) == groups[group_id]['name']:
                return int(group_id)
        return False

    def get_group(self, group_id=None, parameter=None):
        if is_string(group_id):
            group_id = self.get_group_id_by_name(group_id)
        if group_id is False:
            logger.error('Group name does not exit')
            return
        if group_id is None:
            return self.request('GET', '/api/' + self.username + '/groups/')
        if parameter is None:
            return self.request('GET', '/api/' + self.username + '/groups/' + str(group_id))
        elif parameter == 'name' or parameter == 'lights':
            return self.request('GET', '/api/' + self.username + '/groups/' + str(group_id))[parameter]
        else:
            return self.request('GET', '/api/' + self.username + '/groups/' + str(group_id))['action'][parameter]

    def set_group(self, group_id, parameter, value=None, transitiontime=None):
        """ Change light settings for a group

        group_id : int, id number for group
        parameter : 'name' or 'lights'
        value: string, or list of light IDs if you're setting the lights

        """

        if isinstance(parameter, dict):
            data = parameter
        elif parameter == 'lights' and (isinstance(value, list) or isinstance(value, int)):
            if isinstance(value, int):
                value = [value]
            data = {parameter: [str(x) for x in value]}
        else:
            data = {parameter: value}

        if transitiontime is not None:
            data['transitiontime'] = int(round(
                transitiontime))  # must be int for request format

        group_id_array = group_id
        if isinstance(group_id, int) or is_string(group_id):
            group_id_array = [group_id]
        result = []
        for group in group_id_array:
            logger.debug(str(data))
            if is_string(group):
                converted_group = self.get_group_id_by_name(group)
            else:
                converted_group = group
            if converted_group is False:
                logger.error('Group name does not exit')
                return
            if parameter == 'name' or parameter == 'lights':
                result.append(self.request('PUT', '/api/' + self.username + '/groups/' + str(converted_group), data))
            else:
                result.append(self.request('PUT', '/api/' + self.username + '/groups/' + str(converted_group) + '/action', data))

        if 'error' in list(result[-1][0].keys()):
            logger.warn("ERROR: {0} for group {1}".format(
                result[-1][0]['error']['description'], group))

        logger.debug(result)
        return result

    def create_group(self, name, lights=None):
        """ Create a group of lights

        Parameters
        ------------
        name : string
            Name for this group of lights
        lights : list
            List of lights to be in the group.

        """
        data = {'lights': [str(x) for x in lights], 'name': name}
        return self.request('POST', '/api/' + self.username + '/groups/', data)

    def delete_group(self, group_id):
        return self.request('DELETE', '/api/' + self.username + '/groups/' + str(group_id))

    # Scenes #####
    @property
    def scenes(self):
        return [Scene(k, **v) for k, v in self.get_scene().items()]

    def get_scene(self):
        return self.request('GET', '/api/' + self.username + '/scenes')

    def activate_scene(self, group_id, scene_id, transition_time=4):
        return self.request('PUT', '/api/' + self.username + '/groups/' +
                            str(group_id) + '/action',
                            {
                                "scene": scene_id,
                                "transitiontime": transition_time
                            })

    def run_scene(self, group_name, scene_name, transition_time=4):
        """Run a scene by group and scene name.

        As of 1.11 of the Hue API the scenes are accessable in the
        API. With the gen 2 of the official HUE app everything is
        organized by room groups.

        This provides a convenience way of activating scenes by group
        name and scene name. If we find exactly 1 group and 1 scene
        with the matching names, we run them.

        If we find more than one we run the first scene who has
        exactly the same lights defined as the group. This is far from
        perfect, but is convenient for setting lights symbolically (and
        can be improved later).

        :param transition_time: The duration of the transition from the
        light’s current state to the new state in a multiple of 100ms
        :returns True if a scene was run, False otherwise

        """
        groups = [x for x in self.groups if x.name == group_name]
        scenes = [x for x in self.scenes if x.name == scene_name]
        if len(groups) != 1:
            logger.warn("run_scene: More than 1 group found by name {}".format(group_name))
            return False
        group = groups[0]
        if len(scenes) == 0:
            logger.warn("run_scene: No scene found {}".format(scene_name))
            return False
        if len(scenes) == 1:
            self.activate_scene(group.group_id, scenes[0].scene_id, transition_time)
            return True
        # otherwise, lets figure out if one of the named scenes uses
        # all the lights of the group
        group_lights = sorted([x.light_id for x in group.lights])
        for scene in scenes:
            if group_lights == scene.lights:
                self.activate_scene(group.group_id, scene.scene_id, transition_time)
                return True
        logger.warn("run_scene: did not find a scene: {} "
                    "that shared lights with group {}".format(scene_name, group_name))
        return False

    # Schedules #####
    def get_schedule(self, schedule_id=None, parameter=None):
        if schedule_id is None:
            return self.request('GET', '/api/' + self.username + '/schedules')
        if parameter is None:
            return self.request('GET', '/api/' + self.username + '/schedules/' + str(schedule_id))

    def create_schedule(self, name, time, light_id, data, description=' '):
        schedule = {
            'name': name,
            'localtime': time,
            'description': description,
            'command':
            {
                'method': 'PUT',
                'address': ('/api/' + self.username +
                            '/lights/' + str(light_id) + '/state'),
                'body': data
            }
        }
        return self.request('POST', '/api/' + self.username + '/schedules', schedule)

    def set_schedule_attributes(self, schedule_id, attributes):
        """
        :param schedule_id: The ID of the schedule
        :param attributes: Dictionary with attributes and their new values
        """
        return self.request('PUT', '/api/' + self.username + '/schedules/' + str(schedule_id), data=attributes)

    def create_group_schedule(self, name, time, group_id, data, description=' '):
        schedule = {
            'name': name,
            'localtime': time,
            'description': description,
            'command':
            {
                'method': 'PUT',
                'address': ('/api/' + self.username +
                            '/groups/' + str(group_id) + '/action'),
                'body': data
            }
        }
        return self.request('POST', '/api/' + self.username + '/schedules', schedule)

    def delete_schedule(self, schedule_id):
        return self.request('DELETE', '/api/' + self.username + '/schedules/' + str(schedule_id))

if __name__ == '__main__':
    import argparse

    logging.basicConfig(level=logging.DEBUG)

    parser = argparse.ArgumentParser()
    parser.add_argument('--host', required=True)
    parser.add_argument('--config-file-path', required=False)
    args = parser.parse_args()

    while True:
        try:
            b = Bridge(args.host, config_file_path=args.config_file_path)
            break
        except PhueRegistrationException as e:
            if PY3K:
                input('Press button on Bridge then hit Enter to try again')
            else:
                raw_input('Press button on Bridge then hit Enter to try again')  # noqa



Package: broadlink
File: broadlink-0.19.0/broadlink.egg-info/top_level.txt

broadlink



Package: broadlink
File: broadlink-0.19.0/broadlink.egg-info/PKG-INFO

Metadata-Version: 2.1
Name: broadlink
Version: 0.19.0
Summary: Python API for controlling Broadlink devices
Home-page: http://github.com/mjg59/python-broadlink
Author: Matthew Garrett
Author-email: mjg59@srcf.ucam.org
License: UNKNOWN
Platform: UNKNOWN
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
License-File: LICENSE

UNKNOWN




Package: broadlink
File: broadlink-0.19.0/broadlink.egg-info/not-zip-safe





Package: broadlink
File: broadlink-0.19.0/broadlink.egg-info/SOURCES.txt

LICENSE
README.md
setup.py
broadlink/__init__.py
broadlink/alarm.py
broadlink/climate.py
broadlink/const.py
broadlink/cover.py
broadlink/device.py
broadlink/exceptions.py
broadlink/helpers.py
broadlink/hub.py
broadlink/light.py
broadlink/protocol.py
broadlink/remote.py
broadlink/sensor.py
broadlink/switch.py
broadlink.egg-info/PKG-INFO
broadlink.egg-info/SOURCES.txt
broadlink.egg-info/dependency_links.txt
broadlink.egg-info/not-zip-safe
broadlink.egg-info/requires.txt
broadlink.egg-info/top_level.txt


Package: broadlink
File: broadlink-0.19.0/broadlink.egg-info/dependency_links.txt





Package: broadlink
File: broadlink-0.19.0/broadlink.egg-info/requires.txt

cryptography>=3.2



Package: broadlink
File: broadlink-0.19.0/LICENSE

The MIT License (MIT)

Copyright (c) 2014 Mike Ryan
Copyright (c) 2016 Matthew Garrett

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.



Package: broadlink
File: broadlink-0.19.0/setup.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-


from setuptools import setup, find_packages


version = '0.19.0'

setup(
    name="broadlink",
    version=version,
    author="Matthew Garrett",
    author_email="mjg59@srcf.ucam.org",
    url="http://github.com/mjg59/python-broadlink",
    packages=find_packages(),
    scripts=[],
    install_requires=["cryptography>=3.2"],
    description="Python API for controlling Broadlink devices",
    classifiers=[
        "Development Status :: 4 - Beta",
        "Intended Audience :: Developers",
        "License :: OSI Approved :: MIT License",
        "Operating System :: OS Independent",
        "Programming Language :: Python",
    ],
    include_package_data=True,
    zip_safe=False,
)



Package: broadlink
File: broadlink-0.19.0/PKG-INFO

Metadata-Version: 2.1
Name: broadlink
Version: 0.19.0
Summary: Python API for controlling Broadlink devices
Home-page: http://github.com/mjg59/python-broadlink
Author: Matthew Garrett
Author-email: mjg59@srcf.ucam.org
License: UNKNOWN
Platform: UNKNOWN
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
License-File: LICENSE

UNKNOWN




Package: broadlink
File: broadlink-0.19.0/setup.cfg

[egg_info]
tag_build = 
tag_date = 0




Package: broadlink
File: broadlink-0.19.0/broadlink/device.py

"""Support for Broadlink devices."""
import socket
import threading
import random
import time
from typing import Generator, Optional, Tuple, Union

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

from . import exceptions as e
from .const import (
    DEFAULT_BCAST_ADDR,
    DEFAULT_PORT,
    DEFAULT_RETRY_INTVL,
    DEFAULT_TIMEOUT,
)
from .protocol import Datetime

HelloResponse = Tuple[int, Tuple[str, int], str, str, bool]


def scan(
    timeout: int = DEFAULT_TIMEOUT,
    local_ip_address: Optional[str] = None,
    discover_ip_address: str = DEFAULT_BCAST_ADDR,
    discover_ip_port: int = DEFAULT_PORT,
) -> Generator[HelloResponse, None, None]:
    """Broadcast a hello message and yield responses."""
    conn = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    conn.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    conn.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)

    if local_ip_address:
        conn.bind((local_ip_address, 0))
        port = conn.getsockname()[1]
    else:
        local_ip_address = "0.0.0.0"
        port = 0

    packet = bytearray(0x30)
    packet[0x08:0x14] = Datetime.pack(Datetime.now())
    packet[0x18:0x1C] = socket.inet_aton(local_ip_address)[::-1]
    packet[0x1C:0x1E] = port.to_bytes(2, "little")
    packet[0x26] = 6

    checksum = sum(packet, 0xBEAF) & 0xFFFF
    packet[0x20:0x22] = checksum.to_bytes(2, "little")

    start_time = time.time()
    discovered = []

    try:
        while (time.time() - start_time) < timeout:
            time_left = timeout - (time.time() - start_time)
            conn.settimeout(min(DEFAULT_RETRY_INTVL, time_left))
            conn.sendto(packet, (discover_ip_address, discover_ip_port))

            while True:
                try:
                    resp, host = conn.recvfrom(1024)
                except socket.timeout:
                    break

                devtype = resp[0x34] | resp[0x35] << 8
                mac = resp[0x3A:0x40][::-1]

                if (host, mac, devtype) in discovered:
                    continue
                discovered.append((host, mac, devtype))

                name = resp[0x40:].split(b"\x00")[0].decode()
                is_locked = bool(resp[0x7F])
                yield devtype, host, mac, name, is_locked
    finally:
        conn.close()


def ping(ip_address: str, port: int = DEFAULT_PORT) -> None:
    """Send a ping packet to an address.

    This packet feeds the watchdog timer of firmwares >= v53.
    Useful to prevent reboots when the cloud cannot be reached.
    It must be sent every 2 minutes in such cases.
    """
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as conn:
        conn.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
        packet = bytearray(0x30)
        packet[0x26] = 1
        conn.sendto(packet, (ip_address, port))


class Device:
    """Controls a Broadlink device."""

    TYPE = "Unknown"

    __INIT_KEY = "097628343fe99e23765c1513accf8b02"
    __INIT_VECT = "562e17996d093d28ddb3ba695a2e6f58"

    def __init__(
        self,
        host: Tuple[str, int],
        mac: Union[bytes, str],
        devtype: int,
        timeout: int = DEFAULT_TIMEOUT,
        name: str = "",
        model: str = "",
        manufacturer: str = "",
        is_locked: bool = False,
    ) -> None:
        """Initialize the controller."""
        self.host = host
        self.mac = bytes.fromhex(mac) if isinstance(mac, str) else mac
        self.devtype = devtype
        self.timeout = timeout
        self.name = name
        self.model = model
        self.manufacturer = manufacturer
        self.is_locked = is_locked
        self.count = random.randint(0x8000, 0xFFFF)
        self.iv = bytes.fromhex(self.__INIT_VECT)
        self.id = 0
        self.type = self.TYPE  # For backwards compatibility.
        self.lock = threading.Lock()

        self.aes = None
        self.update_aes(bytes.fromhex(self.__INIT_KEY))

    def __repr__(self) -> str:
        """Return a formal representation of the device."""
        return (
            "%s.%s(%s, mac=%r, devtype=%r, timeout=%r, name=%r, "
            "model=%r, manufacturer=%r, is_locked=%r)"
        ) % (
            self.__class__.__module__,
            self.__class__.__qualname__,
            self.host,
            self.mac,
            self.devtype,
            self.timeout,
            self.name,
            self.model,
            self.manufacturer,
            self.is_locked,
        )

    def __str__(self) -> str:
        """Return a readable representation of the device."""
        return "%s (%s / %s:%s / %s)" % (
            self.name or "Unknown",
            " ".join(filter(None, [self.manufacturer, self.model, hex(self.devtype)])),
            *self.host,
            ":".join(format(x, "02X") for x in self.mac),
        )

    def update_aes(self, key: bytes) -> None:
        """Update AES."""
        self.aes = Cipher(
            algorithms.AES(bytes(key)), modes.CBC(self.iv), backend=default_backend()
        )

    def encrypt(self, payload: bytes) -> bytes:
        """Encrypt the payload."""
        encryptor = self.aes.encryptor()
        return encryptor.update(bytes(payload)) + encryptor.finalize()

    def decrypt(self, payload: bytes) -> bytes:
        """Decrypt the payload."""
        decryptor = self.aes.decryptor()
        return decryptor.update(bytes(payload)) + decryptor.finalize()

    def auth(self) -> bool:
        """Authenticate to the device."""
        self.id = 0
        self.update_aes(bytes.fromhex(self.__INIT_KEY))

        packet = bytearray(0x50)
        packet[0x04:0x14] = [0x31] * 16
        packet[0x1E] = 0x01
        packet[0x2D] = 0x01
        packet[0x30:0x36] = "Test 1".encode()

        response = self.send_packet(0x65, packet)
        e.check_error(response[0x22:0x24])
        payload = self.decrypt(response[0x38:])

        self.id = int.from_bytes(payload[:0x4], "little")
        self.update_aes(payload[0x04:0x14])
        return True

    def hello(self, local_ip_address=None) -> bool:
        """Send a hello message to the device.

        Device information is checked before updating name and lock status.
        """
        responses = scan(
            timeout=self.timeout,
            local_ip_address=local_ip_address,
            discover_ip_address=self.host[0],
            discover_ip_port=self.host[1],
        )
        try:
            devtype, _, mac, name, is_locked = next(responses)

        except StopIteration as err:
            raise e.NetworkTimeoutError(
                -4000,
                "Network timeout",
                f"No response received within {self.timeout}s",
            ) from err

        if mac != self.mac:
            raise e.DataValidationError(
                -2040,
                "Device information is not intact",
                "The MAC address is different",
                f"Expected {self.mac} and received {mac}",
            )

        if devtype != self.devtype:
            raise e.DataValidationError(
                -2040,
                "Device information is not intact",
                "The product ID is different",
                f"Expected {self.devtype} and received {devtype}",
            )

        self.name = name
        self.is_locked = is_locked
        return True

    def ping(self) -> None:
        """Ping the device.

        This packet feeds the watchdog timer of firmwares >= v53.
        Useful to prevent reboots when the cloud cannot be reached.
        It must be sent every 2 minutes in such cases.
        """
        ping(self.host[0], port=self.host[1])

    def get_fwversion(self) -> int:
        """Get firmware version."""
        packet = bytearray([0x68])
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        payload = self.decrypt(response[0x38:])
        return payload[0x4] | payload[0x5] << 8

    def set_name(self, name: str) -> None:
        """Set device name."""
        packet = bytearray(4)
        packet += name.encode("utf-8")
        packet += bytearray(0x50 - len(packet))
        packet[0x43] = self.is_locked
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        self.name = name

    def set_lock(self, state: bool) -> None:
        """Lock/unlock the device."""
        packet = bytearray(4)
        packet += self.name.encode("utf-8")
        packet += bytearray(0x50 - len(packet))
        packet[0x43] = bool(state)
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        self.is_locked = bool(state)

    def get_type(self) -> str:
        """Return device type."""
        return self.type

    def send_packet(self, packet_type: int, payload: bytes) -> bytes:
        """Send a packet to the device."""
        self.count = ((self.count + 1) | 0x8000) & 0xFFFF
        packet = bytearray(0x38)
        packet[0x00:0x08] = bytes.fromhex("5aa5aa555aa5aa55")
        packet[0x24:0x26] = self.devtype.to_bytes(2, "little")
        packet[0x26:0x28] = packet_type.to_bytes(2, "little")
        packet[0x28:0x2A] = self.count.to_bytes(2, "little")
        packet[0x2A:0x30] = self.mac[::-1]
        packet[0x30:0x34] = self.id.to_bytes(4, "little")

        p_checksum = sum(payload, 0xBEAF) & 0xFFFF
        packet[0x34:0x36] = p_checksum.to_bytes(2, "little")

        padding = (16 - len(payload)) % 16
        payload = self.encrypt(payload + bytes(padding))
        packet.extend(payload)

        checksum = sum(packet, 0xBEAF) & 0xFFFF
        packet[0x20:0x22] = checksum.to_bytes(2, "little")

        with self.lock and socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as conn:
            timeout = self.timeout
            start_time = time.time()

            while True:
                time_left = timeout - (time.time() - start_time)
                conn.settimeout(min(DEFAULT_RETRY_INTVL, time_left))
                conn.sendto(packet, self.host)

                try:
                    resp = conn.recvfrom(2048)[0]
                    break
                except socket.timeout as err:
                    if (time.time() - start_time) > timeout:
                        raise e.NetworkTimeoutError(
                            -4000,
                            "Network timeout",
                            f"No response received within {timeout}s",
                        ) from err

        if len(resp) < 0x30:
            raise e.DataValidationError(
                -4007,
                "Received data packet length error",
                f"Expected at least 48 bytes and received {len(resp)}",
            )

        nom_checksum = int.from_bytes(resp[0x20:0x22], "little")
        real_checksum = sum(resp, 0xBEAF) - sum(resp[0x20:0x22]) & 0xFFFF

        if nom_checksum != real_checksum:
            raise e.DataValidationError(
                -4008,
                "Received data packet check error",
                f"Expected a checksum of {nom_checksum} and received {real_checksum}",
            )

        return resp



Package: broadlink
File: broadlink-0.19.0/broadlink/protocol.py

"""The networking part of the python-broadlink library."""
import datetime as dt
import time


class Datetime:
    """Helps to pack and unpack datetime objects for the Broadlink protocol."""

    @staticmethod
    def pack(datetime: dt.datetime) -> bytes:
        """Pack the timestamp to be sent over the Broadlink protocol."""
        data = bytearray(12)
        utcoffset = int(datetime.utcoffset().total_seconds() / 3600)
        data[:0x04] = utcoffset.to_bytes(4, "little", signed=True)
        data[0x04:0x06] = datetime.year.to_bytes(2, "little")
        data[0x06] = datetime.minute
        data[0x07] = datetime.hour
        data[0x08] = int(datetime.strftime("%y"))
        data[0x09] = datetime.isoweekday()
        data[0x0A] = datetime.day
        data[0x0B] = datetime.month
        return data

    @staticmethod
    def unpack(data: bytes) -> dt.datetime:
        """Unpack a timestamp received over the Broadlink protocol."""
        utcoffset = int.from_bytes(data[0x00:0x04], "little", signed=True)
        year = int.from_bytes(data[0x04:0x06], "little")
        minute = data[0x06]
        hour = data[0x07]
        subyear = data[0x08]
        isoweekday = data[0x09]
        day = data[0x0A]
        month = data[0x0B]

        tz_info = dt.timezone(dt.timedelta(hours=utcoffset))
        datetime = dt.datetime(year, month, day, hour, minute, 0, 0, tz_info)

        if datetime.isoweekday() != isoweekday:
            raise ValueError("isoweekday does not match")
        if int(datetime.strftime("%y")) != subyear:
            raise ValueError("subyear does not match")

        return datetime

    @staticmethod
    def now() -> dt.datetime:
        """Return the current date and time with timezone info."""
        tz_info = dt.timezone(dt.timedelta(seconds=-time.timezone))
        return dt.datetime.now(tz_info)



Package: broadlink
File: broadlink-0.19.0/broadlink/climate.py

"""Support for climate control."""
import enum
import struct
from typing import List, Sequence

from . import exceptions as e
from .device import Device
from .helpers import CRC16


class hysen(Device):
    """Controls a Hysen heating thermostat.

    This device is manufactured by Hysen and sold under different
    brands, including Floureon, Beca Energy, Beok and Decdeal.

    Supported models:
    - HY02B05H
    - HY03WE
    """

    TYPE = "HYS"

    def send_request(self, request: Sequence[int]) -> bytes:
        """Send a request to the device."""
        packet = bytearray()
        packet.extend((len(request) + 2).to_bytes(2, "little"))
        packet.extend(request)
        packet.extend(CRC16.calculate(request).to_bytes(2, "little"))

        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        payload = self.decrypt(response[0x38:])

        p_len = int.from_bytes(payload[:0x02], "little")
        nom_crc = int.from_bytes(payload[p_len:p_len+2], "little")
        real_crc = CRC16.calculate(payload[0x02:p_len])

        if nom_crc != real_crc:
            raise e.DataValidationError(
                -4008,
                "Received data packet check error",
                f"Expected a checksum of {nom_crc} and received {real_crc}",
            )

        return payload[0x02:p_len]

    def _decode_temp(self, payload, base_index):
        base_temp = payload[base_index] / 2.0
        add_offset = (payload[4] >> 3) & 1  # should offset be added?
        offset_raw_value = (payload[17] >> 4) & 3  # offset value
        offset = (offset_raw_value + 1) / 10 if add_offset else 0.0
        return base_temp + offset

    def get_temp(self) -> float:
        """Return the room temperature in degrees celsius."""
        payload = self.send_request([0x01, 0x03, 0x00, 0x00, 0x00, 0x08])
        return self._decode_temp(payload, 5)

    def get_external_temp(self) -> float:
        """Return the external temperature in degrees celsius."""
        payload = self.send_request([0x01, 0x03, 0x00, 0x00, 0x00, 0x08])
        return self._decode_temp(payload, 18)

    def get_full_status(self) -> dict:
        """Return the state of the device.

        Timer schedule included.
        """
        payload = self.send_request([0x01, 0x03, 0x00, 0x00, 0x00, 0x16])
        data = {}
        data["remote_lock"] = payload[3] & 1
        data["power"] = payload[4] & 1
        data["active"] = (payload[4] >> 4) & 1
        data["temp_manual"] = (payload[4] >> 6) & 1
        data["heating_cooling"] = (payload[4] >> 7) & 1
        data["room_temp"] = self._decode_temp(payload, 5)
        data["thermostat_temp"] = payload[6] / 2.0
        data["auto_mode"] = payload[7] & 0x0F
        data["loop_mode"] = payload[7] >> 4
        data["sensor"] = payload[8]
        data["osv"] = payload[9]
        data["dif"] = payload[10]
        data["svh"] = payload[11]
        data["svl"] = payload[12]
        data["room_temp_adj"] = (
            int.from_bytes(payload[13:15], "big", signed=True) / 10.0
        )
        data["fre"] = payload[15]
        data["poweron"] = payload[16]
        data["unknown"] = payload[17]
        data["external_temp"] = self._decode_temp(payload, 18)
        data["hour"] = payload[19]
        data["min"] = payload[20]
        data["sec"] = payload[21]
        data["dayofweek"] = payload[22]

        weekday = []
        for i in range(0, 6):
            weekday.append(
                {
                    "start_hour": payload[2 * i + 23],
                    "start_minute": payload[2 * i + 24],
                    "temp": payload[i + 39] / 2.0,
                }
            )

        data["weekday"] = weekday
        weekend = []
        for i in range(6, 8):
            weekend.append(
                {
                    "start_hour": payload[2 * i + 23],
                    "start_minute": payload[2 * i + 24],
                    "temp": payload[i + 39] / 2.0,
                }
            )

        data["weekend"] = weekend
        return data

    # Change controller mode
    # auto_mode = 1 for auto (scheduled/timed) mode, 0 for manual mode.
    # Manual mode will activate last used temperature.
    # In typical usage call set_temp to activate manual control and set temp.
    # loop_mode refers to index in [ "12345,67", "123456,7", "1234567" ]
    # E.g. loop_mode = 0 ("12345,67") means Saturday and Sunday (weekend schedule)
    # loop_mode = 2 ("1234567") means every day, including Saturday and Sunday (weekday schedule)
    # The sensor command is currently experimental
    def set_mode(
        self, auto_mode: int, loop_mode: int, sensor: int = 0
    ) -> None:
        """Set the mode of the device."""
        mode_byte = ((loop_mode + 1) << 4) + auto_mode
        self.send_request([0x01, 0x06, 0x00, 0x02, mode_byte, sensor])

    # Advanced settings
    # Sensor mode (SEN) sensor = 0 for internal sensor, 1 for external sensor,
    # 2 for internal control temperature, external limit temperature. Factory default: 0.
    # Set temperature range for external sensor (OSV) osv = 5..99. Factory default: 42C
    # Deadzone for floor temprature (dIF) dif = 1..9. Factory default: 2C
    # Upper temperature limit for internal sensor (SVH) svh = 5..99. Factory default: 35C
    # Lower temperature limit for internal sensor (SVL) svl = 5..99. Factory default: 5C
    # Actual temperature calibration (AdJ) adj = -0.5. Precision 0.1C
    # Anti-freezing function (FrE) fre = 0 for anti-freezing function shut down,
    #  1 for anti-freezing function open. Factory default: 0
    # Power on memory (POn) poweron = 0 for off, 1 for on. Default: 0
    def set_advanced(
        self,
        loop_mode: int,
        sensor: int,
        osv: int,
        dif: int,
        svh: int,
        svl: int,
        adj: float,
        fre: int,
        poweron: int,
    ) -> None:
        """Set advanced options."""
        self.send_request(
            [
                0x01,
                0x10,
                0x00,
                0x02,
                0x00,
                0x05,
                0x0A,
                loop_mode,
                sensor,
                osv,
                dif,
                svh,
                svl,
                int(adj * 10) >> 8 & 0xFF,
                int(adj * 10) & 0xFF,
                fre,
                poweron,
            ]
        )

    # For backwards compatibility only.  Prefer calling set_mode directly.
    # Note this function invokes loop_mode=0 and sensor=0.
    def switch_to_auto(self) -> None:
        """Switch mode to auto."""
        self.set_mode(auto_mode=1, loop_mode=0)

    def switch_to_manual(self) -> None:
        """Switch mode to manual."""
        self.set_mode(auto_mode=0, loop_mode=0)

    # Set temperature for manual mode (also activates manual mode if currently in automatic)
    def set_temp(self, temp: float) -> None:
        """Set the target temperature."""
        self.send_request([0x01, 0x06, 0x00, 0x01, 0x00, int(temp * 2)])

    # Set device on(1) or off(0), does not deactivate Wifi connectivity.
    # Remote lock disables control by buttons on thermostat.
    # heating_cooling: heating(0) cooling(1)
    def set_power(
        self, power: int = 1, remote_lock: int = 0, heating_cooling: int = 0
    ) -> None:
        """Set the power state of the device."""
        state = (heating_cooling << 7) + power
        self.send_request([0x01, 0x06, 0x00, 0x00, remote_lock, state])

    # set time on device
    # n.b. day=1 is Monday, ..., day=7 is Sunday
    def set_time(self, hour: int, minute: int, second: int, day: int) -> None:
        """Set the time."""
        self.send_request(
            [
                0x01,
                0x10,
                0x00,
                0x08,
                0x00,
                0x02,
                0x04,
                hour,
                minute,
                second,
                day
            ]
        )

    # Set timer schedule
    # Format is the same as you get from get_full_status.
    # weekday is a list (ordered) of 6 dicts like:
    # {'start_hour':17, 'start_minute':30, 'temp': 22 }
    # Each one specifies the thermostat temp that will become effective at start_hour:start_minute
    # weekend is similar but only has 2 (e.g. switch on in morning and off in afternoon)
    def set_schedule(self, weekday: List[dict], weekend: List[dict]) -> None:
        """Set timer schedule."""
        request = [0x01, 0x10, 0x00, 0x0A, 0x00, 0x0C, 0x18]

        # weekday times
        for i in range(0, 6):
            request.append(weekday[i]["start_hour"])
            request.append(weekday[i]["start_minute"])

        # weekend times
        for i in range(0, 2):
            request.append(weekend[i]["start_hour"])
            request.append(weekend[i]["start_minute"])

        # weekday temperatures
        for i in range(0, 6):
            request.append(int(weekday[i]["temp"] * 2))

        # weekend temperatures
        for i in range(0, 2):
            request.append(int(weekend[i]["temp"] * 2))

        self.send_request(request)


class hvac(Device):
    """Controls a HVAC.

    Supported models:
    - Tornado SMART X SQ series
    - Aux ASW-H12U3/JIR1DI-US
    - Aux ASW-H36U2/LFR1DI-US
    """

    TYPE = "HVAC"

    @enum.unique
    class Mode(enum.IntEnum):
        """Enumerates modes."""

        AUTO = 0
        COOL = 1
        DRY = 2
        HEAT = 3
        FAN = 4

    @enum.unique
    class Speed(enum.IntEnum):
        """Enumerates fan speed."""

        HIGH = 1
        MID = 2
        LOW = 3
        AUTO = 5

    @enum.unique
    class Preset(enum.IntEnum):
        """Enumerates presets."""

        NORMAL = 0
        TURBO = 1
        MUTE = 2

    @enum.unique
    class SwHoriz(enum.IntEnum):
        """Enumerates horizontal swing."""

        ON = 0
        OFF = 7

    @enum.unique
    class SwVert(enum.IntEnum):
        """Enumerates vertical swing."""

        ON = 0
        POS1 = 1
        POS2 = 2
        POS3 = 3
        POS4 = 4
        POS5 = 5
        OFF = 7

    def _encode(self, data: bytes) -> bytes:
        """Encode data for transport."""
        packet = bytearray(10)
        p_len = 10 + len(data)
        struct.pack_into(
            "<HHHHH", packet, 0, p_len, 0x00BB, 0x8006, 0, len(data)
        )
        packet += data
        crc = CRC16.calculate(packet[0x02:], polynomial=0x9BE4)
        packet += crc.to_bytes(2, "little")
        return packet

    def _decode(self, response: bytes) -> bytes:
        """Decode data from transport."""
        # payload[0x2:0x8] == bytes([0xbb, 0x00, 0x07, 0x00, 0x00, 0x00])
        payload = self.decrypt(response[0x38:])
        p_len = int.from_bytes(payload[:0x02], "little")
        nom_crc = int.from_bytes(payload[p_len:p_len+2], "little")
        real_crc = CRC16.calculate(payload[0x02:p_len], polynomial=0x9BE4)

        if nom_crc != real_crc:
            raise e.DataValidationError(
                -4008,
                "Received data packet check error",
                f"Expected a checksum of {nom_crc} and received {real_crc}",
            )

        d_len = int.from_bytes(payload[0x08:0x0A], "little")
        return payload[0x0A:0x0A+d_len]

    def _send(self, command: int, data: bytes = b"") -> bytes:
        """Send a command to the unit."""
        prefix = bytes([((command << 4) | 1), 1])
        packet = self._encode(prefix + data)
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        return self._decode(response)[0x02:]

    def _parse_state(self, data: bytes) -> dict:
        """Parse state."""
        state = {}
        state["power"] = bool(data[0x08] & 1 << 5)
        state["target_temp"] = 8 + (data[0x00] >> 3) + (data[0x04] >> 7) * 0.5
        state["swing_v"] = self.SwVert(data[0x00] & 0b111)
        state["swing_h"] = self.SwHoriz(data[0x01] >> 5)
        state["mode"] = self.Mode(data[0x05] >> 5)
        state["speed"] = self.Speed(data[0x03] >> 5)
        state["preset"] = self.Preset(data[0x04] >> 6)
        state["sleep"] = bool(data[0x05] & 1 << 2)
        state["ifeel"] = bool(data[0x05] & 1 << 3)
        state["health"] = bool(data[0x08] & 1 << 1)
        state["clean"] = bool(data[0x08] & 1 << 2)
        state["display"] = bool(data[0x0A] & 1 << 4)
        state["mildew"] = bool(data[0x0A] & 1 << 3)
        return state

    def set_state(
        self,
        power: bool,
        target_temp: float,  # 16<=target_temp<=32
        mode: Mode,
        speed: Speed,
        preset: Preset,
        swing_h: SwHoriz,
        swing_v: SwVert,
        sleep: bool,
        ifeel: bool,
        display: bool,
        health: bool,
        clean: bool,
        mildew: bool,
    ) -> dict:
        """Set the state of the device."""
        # TODO: decode unknown bits
        UNK0 = 0b100
        UNK1 = 0b1101
        UNK2 = 0b101

        target_temp = round(target_temp * 2) / 2

        if preset == self.Preset.MUTE:
            if mode != self.Mode.FAN:
                raise ValueError("mute is only available in fan mode")
            speed = self.Speed.LOW

        elif preset == self.Preset.TURBO:
            if mode not in {self.Mode.COOL, self.Mode.HEAT}:
                raise ValueError("turbo is only available in cooling/heating")
            speed = self.Speed.HIGH

        data = bytearray(0x0D)
        data[0x00] = (int(target_temp) - 8 << 3) | swing_v
        data[0x01] = (swing_h << 5) | UNK0
        data[0x02] = ((target_temp % 1 == 0.5) << 7) | UNK1
        data[0x03] = speed << 5
        data[0x04] = preset << 6
        data[0x05] = mode << 5 | sleep << 2 | ifeel << 3
        data[0x08] = power << 5 | clean << 2 | (health and 0b11)
        data[0x0A] = display << 4 | mildew << 3
        data[0x0C] = UNK2

        resp = self._send(0, data)
        return self._parse_state(resp)

    def get_state(self) -> dict:
        """Returns a dictionary with the unit's parameters.

        Returns:
            dict:
                power (bool):
                target_temp (float): temperature set point 16<n<32
                mode (hvac.Mode):
                speed (hvac.Speed):
                preset (hvac.Preset):
                swing_h (hvac.SwHoriz):
                swing_v (hvac.SwVert):
                sleep (bool):
                ifeel (bool):
                display (bool):
                health (bool):
                clean (bool):
                mildew (bool):
        """
        resp = self._send(1)

        if len(resp) < 13:
            raise e.DataValidationError(
                -4007,
                "Received data packet length error",
                f"Expected at least 15 bytes and received {len(resp) + 2}",
            )

        return self._parse_state(resp)

    def get_ac_info(self) -> dict:
        """Returns dictionary with AC info.

        Returns:
            dict:
                power (bool): power
                ambient_temp (float): ambient temperature
        """
        resp = self._send(2)

        if len(resp) < 22:
            raise e.DataValidationError(
                -4007,
                "Received data packet length error",
                f"Expected at least 24 bytes and received {len(resp) + 2}",
            )

        ac_info = {}
        ac_info["power"] = resp[0x1] & 1

        ambient_temp = resp[0x05] & 0b11111, resp[0x15] & 0b11111
        if any(ambient_temp):
            ac_info["ambient_temp"] = ambient_temp[0] + ambient_temp[1] / 10.0

        return ac_info



Package: broadlink
File: broadlink-0.19.0/broadlink/switch.py

"""Support for switches."""
import json
import struct
from typing import Optional

from . import exceptions as e
from .device import Device


class sp1(Device):
    """Controls a Broadlink SP1."""

    TYPE = "SP1"

    def set_power(self, pwr: bool) -> None:
        """Set the power state of the device."""
        packet = bytearray(4)
        packet[0] = bool(pwr)
        response = self.send_packet(0x66, packet)
        e.check_error(response[0x22:0x24])


class sp2(Device):
    """Controls a Broadlink SP2."""

    TYPE = "SP2"

    def set_power(self, pwr: bool) -> None:
        """Set the power state of the device."""
        packet = bytearray(16)
        packet[0] = 2
        packet[4] = bool(pwr)
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])

    def check_power(self) -> bool:
        """Return the power state of the device."""
        packet = bytearray(16)
        packet[0] = 1
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        payload = self.decrypt(response[0x38:])
        return bool(payload[0x4])


class sp2s(sp2):
    """Controls a Broadlink SP2S."""

    TYPE = "SP2S"

    def get_energy(self) -> float:
        """Return the power consumption in W."""
        packet = bytearray(16)
        packet[0] = 4
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        payload = self.decrypt(response[0x38:])
        return int.from_bytes(payload[0x4:0x7], "little") / 1000


class sp3(Device):
    """Controls a Broadlink SP3."""

    TYPE = "SP3"

    def set_power(self, pwr: bool) -> None:
        """Set the power state of the device."""
        packet = bytearray(16)
        packet[0] = 2
        packet[4] = self.check_nightlight() << 1 | bool(pwr)
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])

    def set_nightlight(self, ntlight: bool) -> None:
        """Set the night light state of the device."""
        packet = bytearray(16)
        packet[0] = 2
        packet[4] = bool(ntlight) << 1 | self.check_power()
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])

    def check_power(self) -> bool:
        """Return the power state of the device."""
        packet = bytearray(16)
        packet[0] = 1
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        payload = self.decrypt(response[0x38:])
        return bool(payload[0x4] & 1)

    def check_nightlight(self) -> bool:
        """Return the state of the night light."""
        packet = bytearray(16)
        packet[0] = 1
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        payload = self.decrypt(response[0x38:])
        return bool(payload[0x4] & 2)


class sp3s(sp2):
    """Controls a Broadlink SP3S."""

    TYPE = "SP3S"

    def get_energy(self) -> float:
        """Return the power consumption in W."""
        packet = bytearray([8, 0, 254, 1, 5, 1, 0, 0, 0, 45])
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        payload = self.decrypt(response[0x38:])
        energy = payload[0x7:0x4:-1].hex()
        return int(energy) / 100


class sp4(Device):
    """Controls a Broadlink SP4."""

    TYPE = "SP4"

    def set_power(self, pwr: bool) -> None:
        """Set the power state of the device."""
        self.set_state(pwr=pwr)

    def set_nightlight(self, ntlight: bool) -> None:
        """Set the night light state of the device."""
        self.set_state(ntlight=ntlight)

    def set_state(
        self,
        pwr: Optional[bool] = None,
        ntlight: Optional[bool] = None,
        indicator: Optional[bool] = None,
        ntlbrightness: Optional[int] = None,
        maxworktime: Optional[int] = None,
        childlock: Optional[bool] = None,
    ) -> dict:
        """Set state of device."""
        state = {}
        if pwr is not None:
            state["pwr"] = int(bool(pwr))
        if ntlight is not None:
            state["ntlight"] = int(bool(ntlight))
        if indicator is not None:
            state["indicator"] = int(bool(indicator))
        if ntlbrightness is not None:
            state["ntlbrightness"] = ntlbrightness
        if maxworktime is not None:
            state["maxworktime"] = maxworktime
        if childlock is not None:
            state["childlock"] = int(bool(childlock))

        packet = self._encode(2, state)
        response = self.send_packet(0x6A, packet)
        return self._decode(response)

    def check_power(self) -> bool:
        """Return the power state of the device."""
        state = self.get_state()
        return bool(state["pwr"])

    def check_nightlight(self) -> bool:
        """Return the state of the night light."""
        state = self.get_state()
        return bool(state["ntlight"])

    def get_state(self) -> dict:
        """Get full state of device."""
        packet = self._encode(1, {})
        response = self.send_packet(0x6A, packet)
        return self._decode(response)

    def _encode(self, flag: int, state: dict) -> bytes:
        """Encode a message."""
        packet = bytearray(12)
        data = json.dumps(state, separators=(",", ":")).encode()
        struct.pack_into(
            "<HHHBBI", packet, 0, 0xA5A5, 0x5A5A, 0x0000, flag, 0x0B, len(data)
        )
        packet.extend(data)
        checksum = sum(packet, 0xBEAF) & 0xFFFF
        packet[0x04:0x06] = checksum.to_bytes(2, "little")
        return packet

    def _decode(self, response: bytes) -> dict:
        """Decode a message."""
        e.check_error(response[0x22:0x24])
        payload = self.decrypt(response[0x38:])
        js_len = struct.unpack_from("<I", payload, 0x08)[0]
        state = json.loads(payload[0x0C:0x0C+js_len])
        return state


class sp4b(sp4):
    """Controls a Broadlink SP4 (type B)."""

    TYPE = "SP4B"

    def get_state(self) -> dict:
        """Get full state of device."""
        state = super().get_state()

        # Convert sensor data to float. Remove keys if sensors are not supported.
        sensor_attrs = ["current", "volt", "power", "totalconsum", "overload"]
        for attr in sensor_attrs:
            value = state.pop(attr, -1)
            if value != -1:
                state[attr] = value / 1000
        return state

    def _encode(self, flag: int, state: dict) -> bytes:
        """Encode a message."""
        packet = bytearray(14)
        data = json.dumps(state, separators=(",", ":")).encode()
        length = 12 + len(data)
        struct.pack_into(
            "<HHHHBBI",
            packet,
            0,
            length,
            0xA5A5,
            0x5A5A,
            0x0000,
            flag,
            0x0B,
            len(data),
        )
        packet.extend(data)
        checksum = sum(packet[0x02:], 0xBEAF) & 0xFFFF
        packet[0x06:0x08] = checksum.to_bytes(2, "little")
        return packet

    def _decode(self, response: bytes) -> dict:
        """Decode a message."""
        e.check_error(response[0x22:0x24])
        payload = self.decrypt(response[0x38:])
        js_len = struct.unpack_from("<I", payload, 0xA)[0]
        state = json.loads(payload[0x0E:0x0E+js_len])
        return state


class bg1(Device):
    """Controls a BG Electrical smart outlet."""

    TYPE = "BG1"

    def get_state(self) -> dict:
        """Return the power state of the device.

        Example: `{"pwr":1,"pwr1":1,"pwr2":0,"maxworktime":60,"maxworktime1":60,"maxworktime2":0,"idcbrightness":50}`
        """
        packet = self._encode(1, {})
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        return self._decode(response)

    def set_state(
        self,
        pwr: Optional[bool] = None,
        pwr1: Optional[bool] = None,
        pwr2: Optional[bool] = None,
        maxworktime: Optional[int] = None,
        maxworktime1: Optional[int] = None,
        maxworktime2: Optional[int] = None,
        idcbrightness: Optional[int] = None,
    ) -> dict:
        """Set the power state of the device."""
        state = {}
        if pwr is not None:
            state["pwr"] = int(bool(pwr))
        if pwr1 is not None:
            state["pwr1"] = int(bool(pwr1))
        if pwr2 is not None:
            state["pwr2"] = int(bool(pwr2))
        if maxworktime is not None:
            state["maxworktime"] = maxworktime
        if maxworktime1 is not None:
            state["maxworktime1"] = maxworktime1
        if maxworktime2 is not None:
            state["maxworktime2"] = maxworktime2
        if idcbrightness is not None:
            state["idcbrightness"] = idcbrightness

        packet = self._encode(2, state)
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        return self._decode(response)

    def _encode(self, flag: int, state: dict) -> bytes:
        """Encode a message."""
        packet = bytearray(14)
        data = json.dumps(state).encode()
        length = 12 + len(data)
        struct.pack_into(
            "<HHHHBBI",
            packet,
            0,
            length,
            0xA5A5,
            0x5A5A,
            0x0000,
            flag,
            0x0B,
            len(data),
        )
        packet.extend(data)
        checksum = sum(packet[0x2:], 0xBEAF) & 0xFFFF
        packet[0x06:0x08] = checksum.to_bytes(2, "little")
        return packet

    def _decode(self, response: bytes) -> dict:
        """Decode a message."""
        payload = self.decrypt(response[0x38:])
        js_len = struct.unpack_from("<I", payload, 0x0A)[0]
        state = json.loads(payload[0x0E:0x0E+js_len])
        return state


class ehc31(bg1):
    """Controls a BG Electrical smart extension lead."""

    TYPE = "EHC31"

    def set_state(
        self,
        pwr: Optional[bool] = None,
        pwr1: Optional[bool] = None,
        pwr2: Optional[bool] = None,
        pwr3: Optional[bool] = None,
        maxworktime1: Optional[int] = None,
        maxworktime2: Optional[int] = None,
        maxworktime3: Optional[int] = None,
        idcbrightness: Optional[int] = None,
        childlock: Optional[bool] = None,
        childlock1: Optional[bool] = None,
        childlock2: Optional[bool] = None,
        childlock3: Optional[bool] = None,
        childlock4: Optional[bool] = None,
    ) -> dict:
        """Set the power state of the device."""
        state = {}
        if pwr is not None:
            state["pwr"] = int(bool(pwr))
        if pwr1 is not None:
            state["pwr1"] = int(bool(pwr1))
        if pwr2 is not None:
            state["pwr2"] = int(bool(pwr2))
        if pwr3 is not None:
            state["pwr3"] = int(bool(pwr3))
        if maxworktime1 is not None:
            state["maxworktime1"] = maxworktime1
        if maxworktime2 is not None:
            state["maxworktime2"] = maxworktime2
        if maxworktime3 is not None:
            state["maxworktime3"] = maxworktime3
        if idcbrightness is not None:
            state["idcbrightness"] = idcbrightness
        if childlock is not None:
            state["childlock"] = int(bool(childlock))
        if childlock1 is not None:
            state["childlock1"] = int(bool(childlock1))
        if childlock2 is not None:
            state["childlock2"] = int(bool(childlock2))
        if childlock3 is not None:
            state["childlock3"] = int(bool(childlock3))
        if childlock4 is not None:
            state["childlock4"] = int(bool(childlock4))

        packet = self._encode(2, state)
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        return self._decode(response)


class mp1(Device):
    """Controls a Broadlink MP1."""

    TYPE = "MP1"

    def set_power_mask(self, sid_mask: int, pwr: bool) -> None:
        """Set the power state of the device."""
        packet = bytearray(16)
        packet[0x00] = 0x0D
        packet[0x02] = 0xA5
        packet[0x03] = 0xA5
        packet[0x04] = 0x5A
        packet[0x05] = 0x5A
        packet[0x06] = 0xB2 + ((sid_mask << 1) if pwr else sid_mask)
        packet[0x07] = 0xC0
        packet[0x08] = 0x02
        packet[0x0A] = 0x03
        packet[0x0D] = sid_mask
        packet[0x0E] = sid_mask if pwr else 0

        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])

    def set_power(self, sid: int, pwr: bool) -> None:
        """Set the power state of the device."""
        sid_mask = 0x01 << (sid - 1)
        self.set_power_mask(sid_mask, pwr)

    def check_power_raw(self) -> int:
        """Return the power state of the device in raw format."""
        packet = bytearray(16)
        packet[0x00] = 0x0A
        packet[0x02] = 0xA5
        packet[0x03] = 0xA5
        packet[0x04] = 0x5A
        packet[0x05] = 0x5A
        packet[0x06] = 0xAE
        packet[0x07] = 0xC0
        packet[0x08] = 0x01

        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        payload = self.decrypt(response[0x38:])
        return payload[0x0E]

    def check_power(self) -> dict:
        """Return the power state of the device."""
        data = self.check_power_raw()
        return {
            "s1": bool(data & 1),
            "s2": bool(data & 2),
            "s3": bool(data & 4),
            "s4": bool(data & 8),
        }


class mp1s(mp1):
    """Controls a Broadlink MP1S."""

    TYPE = "MP1S"

    def get_state(self) -> dict:
        """Return the power state of the device.

        voltage in V.
        current in A.
        power in W.
        power consumption in kW·h.
        """
        packet = bytearray(16)
        packet[0x00] = 0x0E
        packet[0x02] = 0xA5
        packet[0x03] = 0xA5
        packet[0x04] = 0x5A
        packet[0x05] = 0x5A
        packet[0x06] = 0xB2
        packet[0x07] = 0xC0
        packet[0x08] = 0x01
        packet[0x0A] = 0x04

        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        payload = self.decrypt(response[0x38:])
        payload_str = payload.hex()[4:-6]

        def get_value(start, end, factors):
            value = sum(
                int(payload_str[i-2:i]) * factor
                for i, factor in zip(range(start, end, -2), factors)
            )
            return value

        return {
            "volt": get_value(34, 30, [10, 0.1]),
            "current": get_value(40, 34, [1, 0.01, 0.0001]),
            "power": get_value(46, 40, [100, 1, 0.01]),
            "totalconsum": get_value(54, 46, [10000, 100, 1, 0.01]),
        }



Package: broadlink
File: broadlink-0.19.0/broadlink/alarm.py

"""Support for alarm kits."""
from . import exceptions as e
from .device import Device


class S1C(Device):
    """Controls a Broadlink S1C."""

    TYPE = "S1C"

    _SENSORS_TYPES = {
        0x31: "Door Sensor",
        0x91: "Key Fob",
        0x21: "Motion Sensor",
    }

    def get_sensors_status(self) -> dict:
        """Return the state of the sensors."""
        packet = bytearray(16)
        packet[0] = 0x06
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        payload = self.decrypt(response[0x38:])
        count = payload[0x4]
        sensor_data = payload[0x6:]
        sensors = [
            bytearray(sensor_data[i * 83 : (i + 1) * 83])
            for i in range(len(sensor_data) // 83)
        ]
        return {
            "count": count,
            "sensors": [
                {
                    "status": sensor[0],
                    "name": sensor[4:26].decode().strip("\x00"),
                    "type": self._SENSORS_TYPES.get(sensor[3], "Unknown"),
                    "order": sensor[1],
                    "serial": sensor[26:30].hex(),
                }
                for sensor in sensors
                if any(sensor[26:30])
            ],
        }



Package: broadlink
File: broadlink-0.19.0/broadlink/exceptions.py

"""Exceptions for Broadlink devices."""
import collections
import struct


class BroadlinkException(Exception):
    """Base class common to all Broadlink exceptions."""

    def __init__(self, *args, **kwargs):
        """Initialize the exception."""
        super().__init__(*args, **kwargs)
        if len(args) >= 2:
            self.errno = args[0]
            self.strerror = ": ".join(str(arg) for arg in args[1:])
        elif len(args) == 1:
            self.errno = None
            self.strerror = str(args[0])
        else:
            self.errno = None
            self.strerror = ""

    def __str__(self):
        """Return str(self)."""
        if self.errno is not None:
            return "[Errno %s] %s" % (self.errno, self.strerror)
        return self.strerror

    def __eq__(self, other):
        """Return self==value."""
        # pylint: disable=unidiomatic-typecheck
        return type(self) == type(other) and self.args == other.args

    def __hash__(self):
        """Return hash(self)."""
        return hash((type(self), self.args))


class MultipleErrors(BroadlinkException):
    """Multiple errors."""

    def __init__(self, *args, **kwargs):
        """Initialize the exception."""
        errors = args[0][:] if args else []
        counter = collections.Counter(errors)
        strerror = "Multiple errors occurred: %s" % counter
        super().__init__(strerror, **kwargs)
        self.errors = errors

    def __repr__(self):
        """Return repr(self)."""
        return "MultipleErrors(%r)" % self.errors

    def __str__(self):
        """Return str(self)."""
        return self.strerror


class AuthenticationError(BroadlinkException):
    """Authentication error."""


class AuthorizationError(BroadlinkException):
    """Authorization error."""


class CommandNotSupportedError(BroadlinkException):
    """Command not supported error."""


class ConnectionClosedError(BroadlinkException):
    """Connection closed error."""


class StructureAbnormalError(BroadlinkException):
    """Structure abnormal error."""


class DeviceOfflineError(BroadlinkException):
    """Device offline error."""


class ReadError(BroadlinkException):
    """Read error."""


class SendError(BroadlinkException):
    """Send error."""


class SSIDNotFoundError(BroadlinkException):
    """SSID not found error."""


class StorageError(BroadlinkException):
    """Storage error."""


class WriteError(BroadlinkException):
    """Write error."""


class NetworkTimeoutError(BroadlinkException):
    """Network timeout error."""


class DataValidationError(BroadlinkException):
    """Data validation error."""


class UnknownError(BroadlinkException):
    """Unknown error."""


BROADLINK_EXCEPTIONS = {
    # Firmware-related errors are generated by the device.
    -1: (AuthenticationError, "Authentication failed"),
    -2: (ConnectionClosedError, "You have been logged out"),
    -3: (DeviceOfflineError, "The device is offline"),
    -4: (CommandNotSupportedError, "Command not supported"),
    -5: (StorageError, "The device storage is full"),
    -6: (StructureAbnormalError, "Structure is abnormal"),
    -7: (AuthorizationError, "Control key is expired"),
    -8: (SendError, "Send error"),
    -9: (WriteError, "Write error"),
    -10: (ReadError, "Read error"),
    -11: (SSIDNotFoundError, "SSID could not be found in AP configuration"),
    # SDK related errors are generated by this module.
    -2040: (DataValidationError, "Device information is not intact"),
    -4000: (NetworkTimeoutError, "Network timeout"),
    -4007: (DataValidationError, "Received data packet length error"),
    -4008: (DataValidationError, "Received data packet check error"),
    -4009: (DataValidationError, "Received data packet information type error"),
    -4010: (DataValidationError, "Received encrypted data packet length error"),
    -4011: (DataValidationError, "Received encrypted data packet check error"),
    -4012: (AuthorizationError, "Device control ID error"),
}


def exception(err_code: int) -> BroadlinkException:
    """Return exception corresponding to an error code."""
    try:
        exc, msg = BROADLINK_EXCEPTIONS[err_code]
        return exc(err_code, msg)
    except KeyError:
        return UnknownError(err_code, "Unknown error")


def check_error(error: bytes) -> None:
    """Raise exception if an error occurred."""
    error_code = struct.unpack("h", error)[0]
    if error_code:
        raise exception(error_code)



Package: broadlink
File: broadlink-0.19.0/broadlink/__init__.py

#!/usr/bin/env python3
"""The python-broadlink library."""
import socket
from typing import Generator, List, Optional, Tuple, Union

from . import exceptions as e
from .const import DEFAULT_BCAST_ADDR, DEFAULT_PORT, DEFAULT_TIMEOUT
from .alarm import S1C
from .climate import hvac, hysen
from .cover import dooya, dooya2, wser
from .device import Device, ping, scan
from .hub import s3
from .light import lb1, lb2
from .remote import rm, rm4, rm4mini, rm4pro, rmmini, rmminib, rmpro
from .sensor import a1, a2
from .switch import bg1, ehc31, mp1, mp1s, sp1, sp2, sp2s, sp3, sp3s, sp4, sp4b

SUPPORTED_TYPES = {
    sp1: {
        0x0000: ("SP1", "Broadlink"),
    },
    sp2: {
        0x2717: ("NEO", "Ankuoo"),
        0x2719: ("SP2-compatible", "Honeywell"),
        0x271A: ("SP2-compatible", "Honeywell"),
        0x2720: ("SP mini", "Broadlink"),
        0x2728: ("SP2-compatible", "URANT"),
        0x273E: ("SP mini", "Broadlink"),
        0x7530: ("SP2", "Broadlink (OEM)"),
        0x7539: ("SP2-IL", "Broadlink (OEM)"),
        0x753E: ("SP mini 3", "Broadlink"),
        0x7540: ("MP2", "Broadlink"),
        0x7544: ("SP2-CL", "Broadlink"),
        0x7546: ("SP2-UK/BR/IN", "Broadlink (OEM)"),
        0x7547: ("SC1", "Broadlink"),
        0x7549: ("SP mini 3", "Broadlink (OEM)"),
        0x7918: ("SP2", "Broadlink (OEM)"),
        0x7919: ("SP2-compatible", "Honeywell"),
        0x791A: ("SP2-compatible", "Honeywell"),
        0x7D0D: ("SP mini 3", "Broadlink (OEM)"),
    },
    sp2s: {
        0x2711: ("SP2", "Broadlink"),
        0x2716: ("NEO PRO", "Ankuoo"),
        0x271D: ("Ego", "Efergy"),
        0x2736: ("SP mini+", "Broadlink"),
    },
    sp3: {
        0x2733: ("SP3", "Broadlink"),
        0x7D00: ("SP3-EU", "Broadlink (OEM)"),
    },
    sp3s: {
        0x9479: ("SP3S-US", "Broadlink"),
        0x947A: ("SP3S-EU", "Broadlink"),
    },
    sp4: {
        0x7568: ("SP4L-CN", "Broadlink"),
        0x756B: ("SP4M-JP", "Broadlink"),
        0x756C: ("SP4M", "Broadlink"),
        0x756F: ("MCB1", "Broadlink"),
        0x7579: ("SP4L-EU", "Broadlink"),
        0x757B: ("SP4L-AU", "Broadlink"),
        0x7583: ("SP mini 3", "Broadlink"),
        0x7587: ("SP4L-UK", "Broadlink"),
        0x7D11: ("SP mini 3", "Broadlink"),
        0xA4F9: ("WS4", "Broadlink (OEM)"),
        0xA569: ("SP4L-UK", "Broadlink"),
        0xA56A: ("MCB1", "Broadlink"),
        0xA56B: ("SCB1E", "Broadlink"),
        0xA56C: ("SP4L-EU", "Broadlink"),
        0xA576: ("SP4L-AU", "Broadlink"),
        0xA589: ("SP4L-UK", "Broadlink"),
        0xA5D3: ("SP4L-EU", "Broadlink"),
        0xA6F4: ("SP4D-US", "Broadlink"),
    },
    sp4b: {
        0x5115: ("SCB1E", "Broadlink"),
        0x51E2: ("AHC/U-01", "BG Electrical"),
        0x6111: ("MCB1", "Broadlink"),
        0x6113: ("SCB1E", "Broadlink"),
        0x618B: ("SP4L-EU", "Broadlink"),
        0x6489: ("SP4L-AU", "Broadlink"),
        0x648B: ("SP4M-US", "Broadlink"),
        0x648C: ("SP4L-US", "Broadlink"),
        0x6494: ("SCB2", "Broadlink"),
    },
    rmmini: {
        0x2737: ("RM mini 3", "Broadlink"),
        0x278F: ("RM mini", "Broadlink"),
        0x27B7: ("RM mini 3", "Broadlink"),
        0x27C2: ("RM mini 3", "Broadlink"),
        0x27C7: ("RM mini 3", "Broadlink"),
        0x27CC: ("RM mini 3", "Broadlink"),
        0x27CD: ("RM mini 3", "Broadlink"),
        0x27D0: ("RM mini 3", "Broadlink"),
        0x27D1: ("RM mini 3", "Broadlink"),
        0x27D3: ("RM mini 3", "Broadlink"),
        0x27DC: ("RM mini 3", "Broadlink"),
        0x27DE: ("RM mini 3", "Broadlink"),
    },
    rmpro: {
        0x2712: ("RM pro/pro+", "Broadlink"),
        0x272A: ("RM pro", "Broadlink"),
        0x273D: ("RM pro", "Broadlink"),
        0x277C: ("RM home", "Broadlink"),
        0x2783: ("RM home", "Broadlink"),
        0x2787: ("RM pro", "Broadlink"),
        0x278B: ("RM plus", "Broadlink"),
        0x2797: ("RM pro+", "Broadlink"),
        0x279D: ("RM pro+", "Broadlink"),
        0x27A1: ("RM plus", "Broadlink"),
        0x27A6: ("RM plus", "Broadlink"),
        0x27A9: ("RM pro+", "Broadlink"),
        0x27C3: ("RM pro+", "Broadlink"),
    },
    rmminib: {
        0x5F36: ("RM mini 3", "Broadlink"),
        0x6507: ("RM mini 3", "Broadlink"),
        0x6508: ("RM mini 3", "Broadlink"),
    },
    rm4mini: {
        0x51DA: ("RM4 mini", "Broadlink"),
        0x5209: ("RM4 TV mate", "Broadlink"),
        0x520C: ("RM4 mini", "Broadlink"),
        0x520D: ("RM4C mini", "Broadlink"),
        0x5211: ("RM4C mate", "Broadlink"),
        0x5212: ("RM4 TV mate", "Broadlink"),
        0x5216: ("RM4 mini", "Broadlink"),
        0x521C: ("RM4 mini", "Broadlink"),
        0x6070: ("RM4C mini", "Broadlink"),
        0x610E: ("RM4 mini", "Broadlink"),
        0x610F: ("RM4C mini", "Broadlink"),
        0x62BC: ("RM4 mini", "Broadlink"),
        0x62BE: ("RM4C mini", "Broadlink"),
        0x6364: ("RM4S", "Broadlink"),
        0x648D: ("RM4 mini", "Broadlink"),
        0x6539: ("RM4C mini", "Broadlink"),
        0x653A: ("RM4 mini", "Broadlink"),
    },
    rm4pro: {
        0x520B: ("RM4 pro", "Broadlink"),
        0x5213: ("RM4 pro", "Broadlink"),
        0x5218: ("RM4C pro", "Broadlink"),
        0x6026: ("RM4 pro", "Broadlink"),
        0x6184: ("RM4C pro", "Broadlink"),
        0x61A2: ("RM4 pro", "Broadlink"),
        0x649B: ("RM4 pro", "Broadlink"),
        0x653C: ("RM4 pro", "Broadlink"),
    },
    a1: {
        0x2714: ("A1", "Broadlink"),
    },
    a2: {
        0x4F60: ("A2", "Broadlink"),
    },
    mp1: {
        0x4EB5: ("MP1-1K4S", "Broadlink"),
        0x4F1B: ("MP1-1K3S2U", "Broadlink (OEM)"),
        0x4F65: ("MP1-1K3S2U", "Broadlink"),
    },
    mp1s: {
        0x4EF7: ("MP1-1K4S", "Broadlink (OEM)"),
    },
    lb1: {
        0x5043: ("SB800TD", "Broadlink (OEM)"),
        0x504E: ("LB1", "Broadlink"),
        0x606D: ("SLA22RGB9W81/SLA27RGB9W81", "Luceco"),
        0x606E: ("SB500TD", "Broadlink (OEM)"),
        0x60C7: ("LB1", "Broadlink"),
        0x60C8: ("LB1", "Broadlink"),
        0x6112: ("LB1", "Broadlink"),
        0x644B: ("LB1", "Broadlink"),
        0x644C: ("LB27 R1", "Broadlink"),
        0x644E: ("LB26 R1", "Broadlink"),
        0x6488: ("LB27 C1", "Broadlink"),
    },
    lb2: {
        0xA4F4: ("LB27 R1", "Broadlink"),
        0xA5F7: ("LB27 R1", "Broadlink"),
        0xA6EF: ("EFCF60WSMT", "Luceco"),
    },
    S1C: {
        0x2722: ("S2KIT", "Broadlink"),
    },
    s3: {
        0xA59C: ("S3", "Broadlink"),
        0xA64D: ("S3", "Broadlink"),
    },
    hvac: {
        0x4E2A: ("HVAC", "Licensed manufacturer"),
    },
    hysen: {
        0x4EAD: ("HY02/HY03", "Hysen"),
    },
    dooya: {
        0x4E4D: ("DT360E-45/20", "Dooya"),
    },
    dooya2: {
        0x4F6E: ("DT360E-45/20", "Dooya"),
    },
    wser: {
        0x4F6C: ("WSER", "Wistar"),
    },
    bg1: {
        0x51E3: ("BG800/BG900", "BG Electrical"),
    },
    ehc31: {
        0x6480: ("EHC31", "BG Electrical"),
    },
}


def gendevice(
    dev_type: int,
    host: Tuple[str, int],
    mac: Union[bytes, str],
    name: str = "",
    is_locked: bool = False,
) -> Device:
    """Generate a device."""
    for dev_cls, products in SUPPORTED_TYPES.items():
        try:
            model, manufacturer = products[dev_type]

        except KeyError:
            continue

        return dev_cls(
            host,
            mac,
            dev_type,
            name=name,
            model=model,
            manufacturer=manufacturer,
            is_locked=is_locked,
        )

    return Device(host, mac, dev_type, name=name, is_locked=is_locked)


def hello(
    ip_address: str,
    port: int = DEFAULT_PORT,
    timeout: int = DEFAULT_TIMEOUT,
) -> Device:
    """Direct device discovery.

    Useful if the device is locked.
    """
    try:
        return next(
            xdiscover(
                timeout=timeout,
                discover_ip_address=ip_address,
                discover_ip_port=port,
            )
        )
    except StopIteration as err:
        raise e.NetworkTimeoutError(
            -4000,
            "Network timeout",
            f"No response received within {timeout}s",
        ) from err


def discover(
    timeout: int = DEFAULT_TIMEOUT,
    local_ip_address: Optional[str] = None,
    discover_ip_address: str = DEFAULT_BCAST_ADDR,
    discover_ip_port: int = DEFAULT_PORT,
) -> List[Device]:
    """Discover devices connected to the local network."""
    responses = scan(
        timeout, local_ip_address, discover_ip_address, discover_ip_port
    )
    return [gendevice(*resp) for resp in responses]


def xdiscover(
    timeout: int = DEFAULT_TIMEOUT,
    local_ip_address: Optional[str] = None,
    discover_ip_address: str = DEFAULT_BCAST_ADDR,
    discover_ip_port: int = DEFAULT_PORT,
) -> Generator[Device, None, None]:
    """Discover devices connected to the local network.

    This function returns a generator that yields devices instantly.
    """
    responses = scan(
        timeout, local_ip_address, discover_ip_address, discover_ip_port
    )
    for resp in responses:
        yield gendevice(*resp)


# Setup a new Broadlink device via AP Mode. Review the README to see how to enter AP Mode.
# Only tested with Broadlink RM3 Mini (Blackbean)
def setup(
    ssid: str,
    password: str,
    security_mode: int,
    ip_address: str = DEFAULT_BCAST_ADDR,
) -> None:
    """Set up a new Broadlink device via AP mode."""
    # Security mode options are (0 - none, 1 = WEP, 2 = WPA1, 3 = WPA2, 4 = WPA1/2)
    payload = bytearray(0x88)
    payload[0x26] = 0x14  # This seems to always be set to 14
    # Add the SSID to the payload
    ssid_start = 68
    ssid_length = 0
    for letter in ssid:
        payload[(ssid_start + ssid_length)] = ord(letter)
        ssid_length += 1
    # Add the WiFi password to the payload
    pass_start = 100
    pass_length = 0
    for letter in password:
        payload[(pass_start + pass_length)] = ord(letter)
        pass_length += 1

    payload[0x84] = ssid_length  # Character length of SSID
    payload[0x85] = pass_length  # Character length of password
    payload[0x86] = security_mode  # Type of encryption

    checksum = sum(payload, 0xBEAF) & 0xFFFF
    payload[0x20] = checksum & 0xFF  # Checksum 1 position
    payload[0x21] = checksum >> 8  # Checksum 2 position

    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)  # Internet  # UDP
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
    sock.sendto(payload, (ip_address, DEFAULT_PORT))
    sock.close()



Package: broadlink
File: broadlink-0.19.0/broadlink/sensor.py

"""Support for sensors."""
from typing import Sequence

from . import exceptions as e
from .device import Device


class a1(Device):
    """Controls a Broadlink A1."""

    TYPE = "A1"

    _SENSORS_AND_LEVELS = (
        ("light", ("dark", "dim", "normal", "bright")),
        ("air_quality", ("excellent", "good", "normal", "bad")),
        ("noise", ("quiet", "normal", "noisy")),
    )

    def check_sensors(self) -> dict:
        """Return the state of the sensors."""
        data = self.check_sensors_raw()
        for sensor, levels in self._SENSORS_AND_LEVELS:
            try:
                data[sensor] = levels[data[sensor]]
            except IndexError:
                data[sensor] = "unknown"
        return data

    def check_sensors_raw(self) -> dict:
        """Return the state of the sensors in raw format."""
        packet = bytearray([0x1])
        resp = self.send_packet(0x6A, packet)
        e.check_error(resp[0x22:0x24])
        data = self.decrypt(resp[0x38:])

        return {
            "temperature": data[0x04] + data[0x05] / 10.0,
            "humidity": data[0x06] + data[0x07] / 10.0,
            "light": data[0x08],
            "air_quality": data[0x0A],
            "noise": data[0x0C],
        }


class a2(Device):
    """Controls a Broadlink A2."""

    TYPE = "A2"

    def _send(self, operation: int, data: Sequence = b""):
        """Send a command to the device."""
        packet = bytearray(12)
        packet[0x02] = 0xA5
        packet[0x03] = 0xA5
        packet[0x04] = 0x5A
        packet[0x05] = 0x5A
        packet[0x08] = operation
        packet[0x09] = 0x0B

        if data:
            data_len = len(data)
            packet[0x0A] = data_len & 0xFF
            packet[0x0B] = data_len >> 8
            packet += bytes(2)
            packet.extend(data)

        checksum = sum(packet, 0xBEAF) & 0xFFFF
        packet[0x06] = checksum & 0xFF
        packet[0x07] = checksum >> 8

        packet_len = len(packet) - 2
        packet[0x00] = packet_len & 0xFF
        packet[0x01] = packet_len >> 8

        resp = self.send_packet(0x6A, packet)
        e.check_error(resp[0x22:0x24])
        payload = self.decrypt(resp[0x38:])
        return payload

    def check_sensors_raw(self) -> dict:
        """Return the state of the sensors in raw format."""
        data = self._send(1)

        return {
            "temperature": data[0x13] * 256 + data[0x14],
            "humidity": data[0x15] * 256 + data[0x16],
            "pm10": data[0x0D] * 256 + data[0x0E],
            "pm2_5": data[0x0F] * 256 + data[0x10],
            "pm1": data[0x11] * 256 + data[0x12],
        }



Package: broadlink
File: broadlink-0.19.0/broadlink/cover.py

"""Support for covers."""
import time
from typing import Sequence

from . import exceptions as e
from .device import Device


class dooya(Device):
    """Controls a Dooya curtain motor."""

    TYPE = "DT360E"

    def _send(self, command: int, attribute: int = 0) -> int:
        """Send a packet to the device."""
        packet = bytearray(16)
        packet[0x00] = 0x09
        packet[0x02] = 0xBB
        packet[0x03] = command
        packet[0x04] = attribute
        packet[0x09] = 0xFA
        packet[0x0A] = 0x44

        resp = self.send_packet(0x6A, packet)
        e.check_error(resp[0x22:0x24])
        payload = self.decrypt(resp[0x38:])
        return payload[4]

    def open(self) -> int:
        """Open the curtain."""
        return self._send(0x01)

    def close(self) -> int:
        """Close the curtain."""
        return self._send(0x02)

    def stop(self) -> int:
        """Stop the curtain."""
        return self._send(0x03)

    def get_percentage(self) -> int:
        """Return the position of the curtain."""
        return self._send(0x06, 0x5D)

    def set_percentage_and_wait(self, new_percentage: int) -> None:
        """Set the position of the curtain."""
        current = self.get_percentage()
        if current > new_percentage:
            self.close()
            while current is not None and current > new_percentage:
                time.sleep(0.2)
                current = self.get_percentage()

        elif current < new_percentage:
            self.open()
            while current is not None and current < new_percentage:
                time.sleep(0.2)
                current = self.get_percentage()
        self.stop()


class dooya2(Device):
    """Controls a Dooya curtain motor (version 2)."""

    TYPE = "DT360E-2"

    def _send(self, operation: int, data: Sequence = b""):
        """Send a command to the device."""
        packet = bytearray(12)
        packet[0x02] = 0xA5
        packet[0x03] = 0xA5
        packet[0x04] = 0x5A
        packet[0x05] = 0x5A
        packet[0x08] = operation
        packet[0x09] = 0x0B

        if data:
            data_len = len(data)
            packet[0x0A] = data_len & 0xFF
            packet[0x0B] = data_len >> 8
            packet += bytes(2)
            packet.extend(data)

        checksum = sum(packet, 0xBEAF) & 0xFFFF
        packet[0x06] = checksum & 0xFF
        packet[0x07] = checksum >> 8

        packet_len = len(packet) - 2
        packet[0x00] = packet_len & 0xFF
        packet[0x01] = packet_len >> 8

        resp = self.send_packet(0x6A, packet)
        e.check_error(resp[0x22:0x24])
        payload = self.decrypt(resp[0x38:])
        return payload

    def open(self) -> None:
        """Open the curtain."""
        self._send(2, [0x00, 0x01, 0x00])

    def close(self) -> None:
        """Close the curtain."""
        self._send(2, [0x00, 0x02, 0x00])

    def stop(self) -> None:
        """Stop the curtain."""
        self._send(2, [0x00, 0x03, 0x00])

    def get_percentage(self) -> int:
        """Return the position of the curtain."""
        resp = self._send(1, [0x00, 0x06, 0x00])
        return resp[0x11]

    def set_percentage(self, new_percentage: int) -> None:
        """Set the position of the curtain."""
        self._send(2, [0x00, 0x09, new_percentage])


class wser(Device):
    """Controls a Wistar curtain motor"""

    TYPE = "WSER"

    def _send(self, operation: int, data: Sequence = b""):
        """Send a command to the device."""
        packet = bytearray(12)
        packet[0x02] = 0xA5
        packet[0x03] = 0xA5
        packet[0x04] = 0x5A
        packet[0x05] = 0x5A
        packet[0x08] = operation
        packet[0x09] = 0x0B

        if data:
            data_len = len(data)
            packet[0x0A] = data_len & 0xFF
            packet[0x0B] = data_len >> 8
            packet += bytes(2)
            packet.extend(data)

        checksum = sum(packet, 0xBEAF) & 0xFFFF
        packet[0x06] = checksum & 0xFF
        packet[0x07] = checksum >> 8

        packet_len = len(packet) - 2
        packet[0x00] = packet_len & 0xFF
        packet[0x01] = packet_len >> 8

        resp = self.send_packet(0x6A, packet)
        e.check_error(resp[0x22:0x24])
        payload = self.decrypt(resp[0x38:])
        return payload

    def get_position(self) -> int:
        """Return the position of the curtain."""
        resp = self._send(1, [])
        position = resp[0x0E]
        return position

    def open(self) -> int:
        """Open the curtain."""
        resp = self._send(2, [0x4A, 0x31, 0xA0])
        position = resp[0x0E]
        return position

    def close(self) -> int:
        """Close the curtain."""
        resp = self._send(2, [0x61, 0x32, 0xA0])
        position = resp[0x0E]
        return position

    def stop(self) -> int:
        """Stop the curtain."""
        resp = self._send(2, [0x4C, 0x73, 0xA0])
        position = resp[0x0E]
        return position

    def set_position(self, position: int) -> int:
        """Set the position of the curtain."""
        resp = self._send(2, [position, 0x70, 0xA0])
        position = resp[0x0E]
        return position



Package: broadlink
File: broadlink-0.19.0/broadlink/helpers.py

"""Helper functions and classes."""
from typing import Dict, List, Sequence


class CRC16:
    """Helps with CRC-16 calculation.

    CRC tables are cached for performance.
    """

    _cache: Dict[int, List[int]] = {}

    @classmethod
    def get_table(cls, polynomial: int) -> List[int]:
        """Return the CRC-16 table for a polynomial."""
        try:
            crc_table = cls._cache[polynomial]
        except KeyError:
            crc_table = []
            for dividend in range(0, 256):
                remainder = dividend
                for _ in range(0, 8):
                    if remainder & 1:
                        remainder = remainder >> 1 ^ polynomial
                    else:
                        remainder = remainder >> 1
                crc_table.append(remainder)
            cls._cache[polynomial] = crc_table
        return crc_table

    @classmethod
    def calculate(
        cls,
        sequence: Sequence[int],
        polynomial: int = 0xA001,  # CRC-16-ANSI.
        init_value: int = 0xFFFF,
    ) -> int:
        """Calculate the CRC-16 of a sequence of integers."""
        crc_table = cls.get_table(polynomial)
        crc = init_value
        for item in sequence:
            crc = crc >> 8 ^ crc_table[(crc ^ item) & 0xFF]
        return crc



Package: broadlink
File: broadlink-0.19.0/broadlink/const.py

"""Constants."""
DEFAULT_BCAST_ADDR = "255.255.255.255"
DEFAULT_PORT = 80
DEFAULT_RETRY_INTVL = 1
DEFAULT_TIMEOUT = 10



Package: broadlink
File: broadlink-0.19.0/broadlink/remote.py

"""Support for universal remotes."""
import struct
from typing import List, Optional, Tuple

from . import exceptions as e
from .device import Device


def pulses_to_data(pulses: List[int], tick: float = 32.84) -> bytes:
    """Convert a microsecond duration sequence into a Broadlink IR packet."""
    result = bytearray(4)
    result[0x00] = 0x26

    for pulse in pulses:
        div, mod = divmod(int(pulse // tick), 256)
        if div:
            result.append(0)
            result.append(div)
        result.append(mod)

    data_len = len(result) - 4
    result[0x02] = data_len & 0xFF
    result[0x03] = data_len >> 8

    return result


def data_to_pulses(data: bytes, tick: float = 32.84) -> List[int]:
    """Parse a Broadlink packet into a microsecond duration sequence."""
    result = []
    index = 4
    end = min(256 * data[0x03] + data[0x02] + 4, len(data))

    while index < end:
        chunk = data[index]
        index += 1

        if chunk == 0:
            try:
                chunk = 256 * data[index] + data[index + 1]
            except IndexError as err:
                raise ValueError("Malformed data.") from err
            index += 2

        result.append(int(chunk * tick))

    return result


class rmmini(Device):
    """Controls a Broadlink RM mini 3."""

    TYPE = "RMMINI"

    def _send(self, command: int, data: bytes = b"") -> bytes:
        """Send a packet to the device."""
        packet = struct.pack("<I", command) + data
        resp = self.send_packet(0x6A, packet)
        e.check_error(resp[0x22:0x24])
        payload = self.decrypt(resp[0x38:])
        return payload[0x4:]

    def update(self) -> None:
        """Update device name and lock status."""
        resp = self._send(0x1)
        self.name = resp[0x48:].split(b"\x00")[0].decode()
        self.is_locked = bool(resp[0x87])

    def send_data(self, data: bytes) -> None:
        """Send a code to the device."""
        self._send(0x2, data)

    def enter_learning(self) -> None:
        """Enter infrared learning mode."""
        self._send(0x3)

    def check_data(self) -> bytes:
        """Return the last captured code."""
        return self._send(0x4)


class rmpro(rmmini):
    """Controls a Broadlink RM pro."""

    TYPE = "RMPRO"

    def sweep_frequency(self) -> None:
        """Sweep frequency."""
        self._send(0x19)

    def check_frequency(self) -> Tuple[bool, float]:
        """Return True if the frequency was identified successfully."""
        resp = self._send(0x1A)
        is_found = bool(resp[0])
        frequency = struct.unpack("<I", resp[1:5])[0] / 1000.0
        return is_found, frequency

    def find_rf_packet(self, frequency: Optional[float] = None) -> None:
        """Enter radiofrequency learning mode."""
        payload = bytearray()
        if frequency:
            payload += struct.pack("<I", int(frequency * 1000))
        self._send(0x1B, payload)

    def cancel_sweep_frequency(self) -> None:
        """Cancel sweep frequency."""
        self._send(0x1E)

    def check_sensors(self) -> dict:
        """Return the state of the sensors."""
        resp = self._send(0x1)
        temp = struct.unpack("<bb", resp[:0x2])
        return {"temperature": temp[0x0] + temp[0x1] / 10.0}

    def check_temperature(self) -> float:
        """Return the temperature."""
        return self.check_sensors()["temperature"]


class rmminib(rmmini):
    """Controls a Broadlink RM mini 3 (new firmware)."""

    TYPE = "RMMINIB"

    def _send(self, command: int, data: bytes = b"") -> bytes:
        """Send a packet to the device."""
        packet = struct.pack("<HI", len(data) + 4, command) + data
        resp = self.send_packet(0x6A, packet)
        e.check_error(resp[0x22:0x24])
        payload = self.decrypt(resp[0x38:])
        p_len = struct.unpack("<H", payload[:0x2])[0]
        return payload[0x6:p_len+2]


class rm4mini(rmminib):
    """Controls a Broadlink RM4 mini."""

    TYPE = "RM4MINI"

    def check_sensors(self) -> dict:
        """Return the state of the sensors."""
        resp = self._send(0x24)
        temp = struct.unpack("<bb", resp[:0x2])
        return {
            "temperature": temp[0x0] + temp[0x1] / 100.0,
            "humidity": resp[0x2] + resp[0x3] / 100.0,
        }

    def check_temperature(self) -> float:
        """Return the temperature."""
        return self.check_sensors()["temperature"]

    def check_humidity(self) -> float:
        """Return the humidity."""
        return self.check_sensors()["humidity"]


class rm4pro(rm4mini, rmpro):
    """Controls a Broadlink RM4 pro."""

    TYPE = "RM4PRO"


class rm(rmpro):
    """For backwards compatibility."""

    TYPE = "RM2"


class rm4(rm4pro):
    """For backwards compatibility."""

    TYPE = "RM4"



Package: broadlink
File: broadlink-0.19.0/broadlink/light.py

"""Support for lights."""
import enum
import json
import struct
from typing import Optional

from . import exceptions as e
from .device import Device


class lb1(Device):
    """Controls a Broadlink LB1."""

    TYPE = "LB1"

    @enum.unique
    class ColorMode(enum.IntEnum):
        """Enumerates color modes."""

        RGB = 0
        WHITE = 1
        SCENE = 2

    def get_state(self) -> dict:
        """Return the power state of the device.

        Example: `{'red': 128, 'blue': 255, 'green': 128, 'pwr': 1, 'brightness': 75, 'colortemp': 2700, 'hue': 240, 'saturation': 50, 'transitionduration': 1500, 'maxworktime': 0, 'bulb_colormode': 1, 'bulb_scenes': '["@01686464,0,0,0", "#ffffff,10,0,#000000,190,0,0", "2700+100,0,0,0", "#ff0000,500,2500,#00FF00,500,2500,#0000FF,500,2500,0", "@01686464,100,2400,@01686401,100,2400,0", "@01686464,100,2400,@01686401,100,2400,@005a6464,100,2400,@005a6401,100,2400,0", "@01686464,10,0,@00000000,190,0,0", "@01686464,200,0,@005a6464,200,0,0"]', 'bulb_scene': '', 'bulb_sceneidx': 255}`
        """
        packet = self._encode(1, {})
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        return self._decode(response)

    def set_state(
        self,
        pwr: Optional[bool] = None,
        red: Optional[int] = None,
        blue: Optional[int] = None,
        green: Optional[int] = None,
        brightness: Optional[int] = None,
        colortemp: Optional[int] = None,
        hue: Optional[int] = None,
        saturation: Optional[int] = None,
        transitionduration: Optional[int] = None,
        maxworktime: Optional[int] = None,
        bulb_colormode: Optional[int] = None,
        bulb_scenes: Optional[str] = None,
        bulb_scene: Optional[str] = None,
        bulb_sceneidx: Optional[int] = None,
    ) -> dict:
        """Set the power state of the device."""
        state = {}
        if pwr is not None:
            state["pwr"] = int(bool(pwr))
        if red is not None:
            state["red"] = int(red)
        if blue is not None:
            state["blue"] = int(blue)
        if green is not None:
            state["green"] = int(green)
        if brightness is not None:
            state["brightness"] = int(brightness)
        if colortemp is not None:
            state["colortemp"] = int(colortemp)
        if hue is not None:
            state["hue"] = int(hue)
        if saturation is not None:
            state["saturation"] = int(saturation)
        if transitionduration is not None:
            state["transitionduration"] = int(transitionduration)
        if maxworktime is not None:
            state["maxworktime"] = int(maxworktime)
        if bulb_colormode is not None:
            state["bulb_colormode"] = int(bulb_colormode)
        if bulb_scenes is not None:
            state["bulb_scenes"] = str(bulb_scenes)
        if bulb_scene is not None:
            state["bulb_scene"] = str(bulb_scene)
        if bulb_sceneidx is not None:
            state["bulb_sceneidx"] = int(bulb_sceneidx)

        packet = self._encode(2, state)
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        return self._decode(response)

    def _encode(self, flag: int, state: dict) -> bytes:
        """Encode a JSON packet."""
        # flag: 1 for reading, 2 for writing.
        packet = bytearray(14)
        data = json.dumps(state, separators=(",", ":")).encode()
        p_len = 12 + len(data)
        struct.pack_into(
            "<HHHHBBI", packet, 0, p_len, 0xA5A5, 0x5A5A, 0, flag, 0x0B, len(data)
        )
        packet.extend(data)
        checksum = sum(packet[0x02:], 0xBEAF) & 0xFFFF
        packet[0x06:0x08] = checksum.to_bytes(2, "little")
        return packet

    def _decode(self, response: bytes) -> dict:
        """Decode a JSON packet."""
        payload = self.decrypt(response[0x38:])
        js_len = struct.unpack_from("<I", payload, 0xA)[0]
        state = json.loads(payload[0xE:0xE+js_len])
        return state


class lb2(Device):
    """Controls a Broadlink LB26/LB27."""

    TYPE = "LB2"

    @enum.unique
    class ColorMode(enum.IntEnum):
        """Enumerates color modes."""

        RGB = 0
        WHITE = 1
        SCENE = 2

    def get_state(self) -> dict:
        """Return the power state of the device.

        Example: `{'red': 128, 'blue': 255, 'green': 128, 'pwr': 1, 'brightness': 75, 'colortemp': 2700, 'hue': 240, 'saturation': 50, 'transitionduration': 1500, 'maxworktime': 0, 'bulb_colormode': 1, 'bulb_scenes': '["@01686464,0,0,0", "#ffffff,10,0,#000000,190,0,0", "2700+100,0,0,0", "#ff0000,500,2500,#00FF00,500,2500,#0000FF,500,2500,0", "@01686464,100,2400,@01686401,100,2400,0", "@01686464,100,2400,@01686401,100,2400,@005a6464,100,2400,@005a6401,100,2400,0", "@01686464,10,0,@00000000,190,0,0", "@01686464,200,0,@005a6464,200,0,0"]', 'bulb_scene': ''}`
        """
        packet = self._encode(1, {})
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        return self._decode(response)

    def set_state(
        self,
        pwr: Optional[bool] = None,
        red: Optional[int] = None,
        blue: Optional[int] = None,
        green: Optional[int] = None,
        brightness: Optional[int] = None,
        colortemp: Optional[int] = None,
        hue: Optional[int] = None,
        saturation: Optional[int] = None,
        transitionduration: Optional[int] = None,
        maxworktime: Optional[int] = None,
        bulb_colormode: Optional[int] = None,
        bulb_scenes: Optional[str] = None,
        bulb_scene: Optional[str] = None,
    ) -> dict:
        """Set the power state of the device."""
        state = {}
        if pwr is not None:
            state["pwr"] = int(bool(pwr))
        if red is not None:
            state["red"] = int(red)
        if blue is not None:
            state["blue"] = int(blue)
        if green is not None:
            state["green"] = int(green)
        if brightness is not None:
            state["brightness"] = int(brightness)
        if colortemp is not None:
            state["colortemp"] = int(colortemp)
        if hue is not None:
            state["hue"] = int(hue)
        if saturation is not None:
            state["saturation"] = int(saturation)
        if transitionduration is not None:
            state["transitionduration"] = int(transitionduration)
        if maxworktime is not None:
            state["maxworktime"] = int(maxworktime)
        if bulb_colormode is not None:
            state["bulb_colormode"] = int(bulb_colormode)
        if bulb_scenes is not None:
            state["bulb_scenes"] = str(bulb_scenes)
        if bulb_scene is not None:
            state["bulb_scene"] = str(bulb_scene)

        packet = self._encode(2, state)
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        return self._decode(response)

    def _encode(self, flag: int, state: dict) -> bytes:
        """Encode a JSON packet."""
        # flag: 1 for reading, 2 for writing.
        packet = bytearray(12)
        data = json.dumps(state, separators=(",", ":")).encode()
        struct.pack_into(
            "<HHHBBI", packet, 0, 0xA5A5, 0x5A5A, 0, flag, 0x0B, len(data)
        )
        packet.extend(data)
        checksum = sum(packet, 0xBEAF) & 0xFFFF
        packet[0x04:0x06] = checksum.to_bytes(2, "little")
        return packet

    def _decode(self, response: bytes) -> dict:
        """Decode a JSON packet."""
        payload = self.decrypt(response[0x38:])
        js_len = struct.unpack_from("<I", payload, 0x08)[0]
        state = json.loads(payload[0x0C:0x0C+js_len])
        return state



Package: broadlink
File: broadlink-0.19.0/broadlink/hub.py

"""Support for hubs."""
import struct
import json
from typing import Optional

from . import exceptions as e
from .device import Device


class s3(Device):
    """Controls a Broadlink S3."""

    TYPE = "S3"
    MAX_SUBDEVICES = 8

    def get_subdevices(self, step: int = 5) -> list:
        """Return a list of sub devices."""
        total = self.MAX_SUBDEVICES
        sub_devices = []
        seen = set()
        index = 0

        while index < total:
            state = {"count": step, "index": index}
            packet = self._encode(14, state)
            resp = self.send_packet(0x6A, packet)
            e.check_error(resp[0x22:0x24])
            resp = self._decode(resp)

            for device in resp["list"]:
                did = device["did"]
                if did in seen:
                    continue

                seen.add(did)
                sub_devices.append(device)

            total = resp["total"]
            if len(seen) >= total:
                break

            index += step

        return sub_devices

    def get_state(self, did: Optional[str] = None) -> dict:
        """Return the power state of the device."""
        state = {}
        if did is not None:
            state["did"] = did

        packet = self._encode(1, state)
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        return self._decode(response)

    def set_state(
        self,
        did: Optional[str] = None,
        pwr1: Optional[bool] = None,
        pwr2: Optional[bool] = None,
        pwr3: Optional[bool] = None,
    ) -> dict:
        """Set the power state of the device."""
        state = {}
        if did is not None:
            state["did"] = did
        if pwr1 is not None:
            state["pwr1"] = int(bool(pwr1))
        if pwr2 is not None:
            state["pwr2"] = int(bool(pwr2))
        if pwr3 is not None:
            state["pwr3"] = int(bool(pwr3))

        packet = self._encode(2, state)
        response = self.send_packet(0x6A, packet)
        e.check_error(response[0x22:0x24])
        return self._decode(response)

    def _encode(self, flag: int, state: dict) -> bytes:
        """Encode a JSON packet."""
        # flag: 1 for reading, 2 for writing.
        packet = bytearray(12)
        data = json.dumps(state, separators=(",", ":")).encode()
        struct.pack_into(
            "<HHHBBI", packet, 0, 0xA5A5, 0x5A5A, 0, flag, 0x0B, len(data)
        )
        packet.extend(data)
        checksum = sum(packet, 0xBEAF) & 0xFFFF
        packet[0x04:0x06] = checksum.to_bytes(2, "little")
        return packet

    def _decode(self, response: bytes) -> dict:
        """Decode a JSON packet."""
        payload = self.decrypt(response[0x38:])
        js_len = struct.unpack_from("<I", payload, 0x08)[0]
        state = json.loads(payload[0x0C:0x0C+js_len])
        return state



Package: broadlink
File: broadlink-0.19.0/README.md

# python-broadlink

A Python module and CLI for controlling Broadlink devices locally. The following devices are supported:

- **Universal remotes**: RM home, RM mini 3, RM plus, RM pro, RM pro+, RM4 mini, RM4 pro, RM4C mini, RM4S, RM4 TV mate
- **Smart plugs**: SP mini, SP mini 3, SP mini+, SP1, SP2, SP2-BR, SP2-CL, SP2-IN, SP2-UK, SP3, SP3-EU, SP3S-EU, SP3S-US, SP4L-AU, SP4L-EU, SP4L-UK, SP4M, SP4M-US, Ankuoo NEO, Ankuoo NEO PRO, Efergy Ego, BG AHC/U-01
- **Switches**: MCB1, SC1, SCB1E, SCB2
- **Outlets**: BG 800, BG 900
- **Power strips**: MP1-1K3S2U, MP1-1K4S, MP2
- **Environment sensors**: A1
- **Alarm kits**: S1C, S2KIT
- **Light bulbs**: LB1, LB26 R1, LB27 R1, SB800TD
- **Curtain motors**: Dooya DT360E-45/20
- **Thermostats**: Hysen HY02B05H
- **Hubs**: S3

## Installation

Use pip3 to install the latest version of this module.

```
pip3 install broadlink
```

## Basic functions

First, open Python 3 and import this module.

```
python3
```
```python3
import broadlink
```

Now let's try some functions...

### Setup

In order to control the device, you need to connect it to your local network. If you have already configured the device with the Broadlink app, this step is not necessary.

1. Put the device into AP Mode.
  - Long press the reset button until the blue LED is blinking quickly.
  - Long press again until blue LED is blinking slowly.
  - Manually connect to the WiFi SSID named BroadlinkProv.
2. Connect the device to your local network with the setup function.
```python3
broadlink.setup('myssid', 'mynetworkpass', 3)
```

Security mode options are (0 = none, 1 = WEP, 2 = WPA1, 3 = WPA2, 4 = WPA1/2)

#### Advanced options

You may need to specify a broadcast address if setup is not working.
```python3
broadlink.setup('myssid', 'mynetworkpass', 3, ip_address='192.168.0.255')
```

### Discovery

Use this function to discover devices:

```python3
devices = broadlink.discover()
```

#### Advanced options
You may need to specify `local_ip_address` or `discover_ip_address` if discovery does not return any devices.

Using the IP address of your local machine:
```python3
devices = broadlink.discover(local_ip_address='192.168.0.100')
```

Using the broadcast address of your subnet:
```python3
devices = broadlink.discover(discover_ip_address='192.168.0.255')
```

If the device is locked, it may not be discoverable with broadcast. In such cases, you can use the unicast version `broadlink.hello()` for direct discovery:
```python3
device = broadlink.hello('192.168.0.16')
```

If you are a perfomance freak, use `broadlink.xdiscover()` to create devices instantly:
```python3
for device in broadlink.xdiscover():
    print(device)  # Example action. Do whatever you want here.
```

### Authentication
After discovering the device, call the `auth()` method to obtain the authentication key required for further communication:
```python3
device.auth()
```

The next steps depend on the type of device you want to control.

## Universal remotes

### Learning IR codes

Learning IR codes takes place in three steps.

1. Enter learning mode:
```python3
device.enter_learning()
```
2. When the LED blinks, point the remote at the Broadlink device and press the button you want to learn.
3. Get the IR packet.
```python3
packet = device.check_data()
```

### Learning RF codes

Learning RF codes takes place in six steps.

1. Sweep the frequency:
```python3
device.sweep_frequency()
```
2. When the LED blinks, point the remote at the Broadlink device for the first time and long press the button you want to learn.
3. Check if the frequency was successfully identified:
```python3
ok = device.check_frequency()
if ok:
    print('Frequency found!')
```
4. Enter learning mode:
```python3
device.find_rf_packet()
```
5. When the LED blinks, point the remote at the Broadlink device for the second time and short press the button you want to learn.
6. Get the RF packet:
```python3
packet = device.check_data()
```

#### Notes

Universal remotes with product id 0x2712 use the same method for learning IR and RF codes. They don't need to sweep frequency. Just call `device.enter_learning()` and `device.check_data()`.

### Canceling learning

You can exit the learning mode in the middle of the process by calling this method:
```python3
device.cancel_sweep_frequency()
```

### Sending IR/RF packets
```python3
device.send_data(packet)
```

### Fetching sensor data
```python3
data = device.check_sensors()
```

## Switches

### Setting power state
```python3
device.set_power(True)
device.set_power(False)
```

### Checking power state
```python3
state = device.check_power()
```

### Checking energy consumption
```python3
state = device.get_energy()
```

## Power strips

### Setting power state
```python3
device.set_power(1, True)  # Example socket. It could be 2 or 3.
device.set_power(1, False)
```

### Checking power state
```python3
state = device.check_power()
```

## Light bulbs

### Fetching data
```python3
state = device.get_state()
```

### Setting state attributes
```python3
devices[0].set_state(pwr=0)
devices[0].set_state(pwr=1)
devices[0].set_state(brightness=75)
devices[0].set_state(bulb_colormode=0)
devices[0].set_state(blue=255)
devices[0].set_state(red=0)
devices[0].set_state(green=128)
devices[0].set_state(bulb_colormode=1)
```

## Environment sensors

### Fetching sensor data
```python3
data = device.check_sensors()
```

## Hubs

### Discovering subdevices
```python3
device.get_subdevices()
```

### Fetching data
Use the DID obtained from get_subdevices() for the input parameter to query specific sub-device.

```python3
device.get_state(did="00000000000000000000a043b0d06963")
```

### Setting state attributes
The parameters depend on the type of subdevice that is being controlled. In this example, we are controlling LC-1 switches:

#### Turn on
```python3
device.set_state(did="00000000000000000000a043b0d0783a", pwr=1)
device.set_state(did="00000000000000000000a043b0d0783a", pwr1=1)
device.set_state(did="00000000000000000000a043b0d0783a", pwr2=1)
```
#### Turn off
```python3
device.set_state(did="00000000000000000000a043b0d0783a", pwr=0)
device.set_state(did="00000000000000000000a043b0d0783a", pwr1=0)
device.set_state(did="00000000000000000000a043b0d0783a", pwr2=0)
```



Package: dlna
File: dlna-0.2.2/PKG-INFO

Metadata-Version: 2.1
Name: dlna
Version: 0.2.2
Summary: A UPnP/DLNA client, support local file and online resource cast to screen.
Home-page: https://leesoar.com/real
Author: leesoar
Author-email: core@111.com
License: MIT
Description: 
        ## dlna
        
        A UPnP/DLNA client, support local file and online resource cast to screen.
        
        **Support Online Resource.**
        
        Thanks for nanodlna!
        
        
        ### Release Note
        * 0.2.2: Fixed some tips.
        * 0.2: Fixed multiple scans of the same.
        * 0.1.7: Fixed an error caused by custom xml attributes.
        * 0.1.6: Timeout expanded to 10s.
        * 0.1.5: Optimize the tips content.
        * 0.1.4:
            * Now you can select device by name.
            * Add 'How to use'
            * Fix bugs.
        * 0.1.3: Fix bugs.
        
        
        ### How to use
        * scan DLNA-enabled devices
        ```shell
        >>> dlna device
        ```
        
        * resource cast to screen
        ```shell
        >>> dlna play "your_link_or_path"
        ```
        
        * select device by url
        ```shell
        >>> dlna play "your_link_or_path" -d "your_device_url"
        
        # device url like 'http://host:port/'
        ```
        
        * select device by name
        ```shell
        >>> dlna play "your_link_or_path" -q "your_device_name"
        
        # device name like 'room tv'
        ```
Keywords: dlna
Platform: UNKNOWN
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.5
Classifier: Programming Language :: Python :: 3.6
Classifier: Programming Language :: Python :: 3.7
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Programming Language :: Python :: Implementation :: PyPy
Requires-Python: >=3.3.0
Description-Content-Type: text/markdown



Package: dlna
File: dlna-0.2.2/dlna.egg-info/PKG-INFO

Metadata-Version: 2.1
Name: dlna
Version: 0.2.2
Summary: A UPnP/DLNA client, support local file and online resource cast to screen.
Home-page: https://leesoar.com/real
Author: leesoar
Author-email: core@111.com
License: MIT
Description: 
        ## dlna
        
        A UPnP/DLNA client, support local file and online resource cast to screen.
        
        **Support Online Resource.**
        
        Thanks for nanodlna!
        
        
        ### Release Note
        * 0.2.2: Fixed some tips.
        * 0.2: Fixed multiple scans of the same.
        * 0.1.7: Fixed an error caused by custom xml attributes.
        * 0.1.6: Timeout expanded to 10s.
        * 0.1.5: Optimize the tips content.
        * 0.1.4:
            * Now you can select device by name.
            * Add 'How to use'
            * Fix bugs.
        * 0.1.3: Fix bugs.
        
        
        ### How to use
        * scan DLNA-enabled devices
        ```shell
        >>> dlna device
        ```
        
        * resource cast to screen
        ```shell
        >>> dlna play "your_link_or_path"
        ```
        
        * select device by url
        ```shell
        >>> dlna play "your_link_or_path" -d "your_device_url"
        
        # device url like 'http://host:port/'
        ```
        
        * select device by name
        ```shell
        >>> dlna play "your_link_or_path" -q "your_device_name"
        
        # device name like 'room tv'
        ```
Keywords: dlna
Platform: UNKNOWN
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.5
Classifier: Programming Language :: Python :: 3.6
Classifier: Programming Language :: Python :: 3.7
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Programming Language :: Python :: Implementation :: PyPy
Requires-Python: >=3.3.0
Description-Content-Type: text/markdown



Package: dlna
File: dlna-0.2.2/dlna.egg-info/SOURCES.txt

LICENSE
MANIFEST.in
README.md
requirements.txt
setup.py
dlna/__init__.py
dlna/client.py
dlna/util.py
dlna.egg-info/PKG-INFO
dlna.egg-info/SOURCES.txt
dlna.egg-info/dependency_links.txt
dlna.egg-info/entry_points.txt
dlna.egg-info/requires.txt
dlna.egg-info/top_level.txt


Package: dlna
File: dlna-0.2.2/dlna.egg-info/entry_points.txt

[console_scripts]
dlna = dlna.client:run




Package: dlna
File: dlna-0.2.2/dlna.egg-info/requires.txt

lxml
requests
Twisted
urllib3



Package: dlna
File: dlna-0.2.2/dlna.egg-info/top_level.txt

dlna



Package: dlna
File: dlna-0.2.2/dlna.egg-info/dependency_links.txt





Package: dlna
File: dlna-0.2.2/LICENSE

The MIT License (MIT)

Copyright (c) 2020 leesoar.com

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.




Package: dlna
File: dlna-0.2.2/requirements.txt

lxml
requests
Twisted
urllib3


Package: dlna
File: dlna-0.2.2/dlna/util.py

# -*- coding: utf-8 -*-
# @author: leesoar

"""something."""
import os
import pkgutil
import re
import socket
import sys
import threading
from urllib.parse import urljoin, urlparse

if sys.version_info.major == 3:
    import urllib.request as urllibreq
else:
    import urllib2 as urllibreq

import requests
from lxml import etree
from twisted.internet import reactor
from twisted.web.resource import Resource
from twisted.web.server import Site
from twisted.web.static import File

SSDP_BROADCAST_ADDR = "239.255.255.250"
SSDP_BROADCAST_PORT = 1900
SSDP_BROADCAST_PARAMS = [
    "M-SEARCH * HTTP/1.1",
    f"HOST: {SSDP_BROADCAST_ADDR}:{SSDP_BROADCAST_PORT}",
    "MAN: \"ssdp:discover\"",
    "MX: 10",
    "ST: ssdp:all",
    "", ""
]
SSDP_BROADCAST_MSG = "\r\n".join(SSDP_BROADCAST_PARAMS)
UPNP_DEFAULT_SERVICE_TYPE = "urn:schemas-upnp-org:service:AVTransport:1"


def get_serve_ip(target_ip, target_port=80):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.connect((target_ip, target_port))
    serve_ip = s.getsockname()[0]
    s.close()
    return serve_ip


def set_files(files, serve_ip, serve_port):

    files_index = {file_key: (os.path.basename(file_path),
                              os.path.abspath(file_path),
                              os.path.dirname(os.path.abspath(file_path)))
                   for file_key, file_path in files.items()}

    files_serve = {file_name: file_path
                   for file_name, file_path, file_dir in files_index.values()}

    files_urls = {
        file_key: "http://{0}:{1}/{2}/{3}".format(
            serve_ip, serve_port, file_key, file_name)
        for file_key, (file_name, file_path, file_dir)
        in files_index.items()}

    return files_index, files_serve, files_urls


def start_server(files, serve_ip, serve_port=9000):

    # import sys
    # log.startLogging(sys.stdout)

    files_index, files_serve, files_urls = set_files(
        files, serve_ip, serve_port)

    root = Resource()
    for file_key, (file_name, file_path, file_dir) in files_index.items():
        root.putChild(file_key.encode("utf-8"), Resource())
        root.children[file_key.encode("utf-8")].putChild(
            file_name.encode("utf-8"), File(file_path))

    reactor.listenTCP(serve_port, Site(root))
    threading.Thread(
        target=reactor.run, kwargs={"installSignalHandlers": False}).start()

    return files_urls


def parse_xml(url):
    node = etree.XML(re.sub(" xmlns=\"[^\"]+\"", "", requests.get(url).content.decode(), count=1).encode(), parser=etree.XMLParser(recover=True))
    friendly_name = node.xpath("//friendlyName/text()")
    manufacturer_url = node.xpath("//manufacturerURL/text()")
    control_url = node.xpath("//serviceType[contains(text(), 'AVTransport')]/../controlURL/text()")
    manufacturer = node.xpath("//manufacturer/text()")
    data = {
        "location": url,
        "host": urlparse(url).hostname,
        "friendly_name": friendly_name and friendly_name[0],
        "action_url": urljoin(url, control_url and control_url[0]),
        "manufacturer": manufacturer and manufacturer[0],
        "manufacturer_url": manufacturer_url and manufacturer_url[0],
        "st": UPNP_DEFAULT_SERVICE_TYPE,
        "help": "leesoar.com/real | Wechat MP: GMapi | core@111.com, chat with me.",
    }
    return data


def get_device(timeout=10):
    print("Scanning...")
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
    s.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 4)
    s.bind(("", SSDP_BROADCAST_PORT + 10))

    s.sendto(SSDP_BROADCAST_MSG.encode(), (SSDP_BROADCAST_ADDR, SSDP_BROADCAST_PORT))
    s.settimeout(timeout)

    scanned_devices = []
    devices = {}
    while 1:
        try:
            data, addr = s.recvfrom(1024)
        except socket.timeout:
            break

        def serialize(x):
            try:
                k, v = x.split(":", maxsplit=1)
            except ValueError:
                pass
            else:
                return k.lower(), v.lstrip()

        device = dict(map(serialize, filter(lambda x: x.count(":") >= 1, data.decode().split("\r\n"))))
        if device in scanned_devices:
            continue
        if "AVTransport" in device["st"]:
            devices.update({device["location"]: parse_xml(device["location"])})
        scanned_devices.append(device)
    return list(devices.values())


def send_dlna_action(device, data, action):

    action_data = pkgutil.get_data(
        "dlna", "templates/action-{0}.xml".format(action)).decode("UTF-8")
    action_data = action_data.format(**data).encode("UTF-8")

    headers = {
        "Content-Type": "text/xml; charset=\"utf-8\"",
        "Content-Length": "{0}".format(len(action_data)),
        "Connection": "close",
        "SOAPACTION": "\"{0}#{1}\"".format(device["st"], action)
    }

    request = urllibreq.Request(device["action_url"], action_data, headers)
    urllibreq.urlopen(request)


def play(files_urls, device, is_net=False):
    if is_net:
        video_data = {
            "uri_video": files_urls,
            "type_video": "leesoar.com",
            "metadata": "",
        }
    else:
        video_data = {
            "uri_video": files_urls["file_video"],
            "type_video": os.path.splitext(files_urls["file_video"])[1][1:],
            "metadata": "",
        }

    send_dlna_action(device, video_data, "SetAVTransportURI")
    send_dlna_action(device, video_data, "Play")



Package: dlna
File: dlna-0.2.2/dlna/client.py

# -*- coding: utf-8 -*-
# @author: leesoar

"""something."""
import argparse
import json
import sys

from . import util, __version__


def show_devices(args):
    devices = util.get_device(args.timeout)

    if not devices:
        return print("No online devices.")

    for i, d in enumerate(devices, start=1):
        print(f"=> Device {i}:\n{json.dumps(d, ensure_ascii=False, indent=4)}\n")


def play(args):
    src = {"file_video": args.src}

    try:
        if args.device_location:
            device = util.parse_xml(args.device_location)
        elif args.query_device:
            device = list(filter(lambda x: args.query_device.lower() in x["friendly_name"].lower(), util.get_device(args.timeout)))[0]
        else:
            device = util.get_device(args.timeout)[0]
    except Exception:
        device = None

    if not device:
        print("No online devices.")
        sys.exit()

    print(f"Current play device: {device['friendly_name']}")

    if args.src.startswith("http"):
        util.play(args.src.replace("&", "&amp;").replace("\\", ""), device, True)
    else:
        serve_ip = util.get_serve_ip(util.get_serve_ip(device["host"]))
        files_urls = util.start_server(src, serve_ip)
        util.play(files_urls, device)


def run():
    parser = argparse.ArgumentParser(
        description=f"A UPnP/DLNA client, support local file and online resource cast to screen.",
        prog="dlna", add_help=False)

    subparsers = parser.add_subparsers(dest='cmd', title='Available commands')
    parser.add_argument('-v', '--version', action='version', version=__version__, help='Get version of dlna')
    parser.add_argument('-h', '--help', action='help', help='Show help message')

    p_device = subparsers.add_parser('device')
    p_device.set_defaults(func=show_devices)
    p_device.add_argument("-t", "--timeout", type=float, default=5)

    p_play = subparsers.add_parser('play')
    p_play.set_defaults(func=play)
    p_play.add_argument("-d", "--device", dest="device_location")
    p_play.add_argument("-q", "--query", dest="query_device")
    p_play.add_argument("-t", "--timeout", type=float, default=5)
    p_play.add_argument('src', type=str, help="media src")

    try:
        args = parser.parse_args()
        args.func(args)
    except Exception:
        print("Caught an exception, please try again.")



Package: dlna
File: dlna-0.2.2/dlna/__init__.py

# -*- coding: utf-8 -*-
# @author: leesoar

"""something."""

__version__ = "0.2.2"



Package: dlna
File: dlna-0.2.2/MANIFEST.in

include requirements.txt
include README.md
include LICENSE


Package: dlna
File: dlna-0.2.2/README.md

## dlna

A UPnP/DLNA client, support local file and online resource cast to screen.

**Support Online Resource.**

Thanks for nanodlna!


### Release Note
* 0.2.2: Fixed some tips.
* 0.2: Fixed multiple scans of the same.
* 0.1.7: Fixed an error caused by custom xml attributes.
* 0.1.6: Timeout expanded to 10s.
* 0.1.5: Optimize the tips content.
* 0.1.4:
    * Now you can select device by name.
    * Add 'How to use'
    * Fix bugs.
* 0.1.3: Fix bugs.


### How to use
* scan DLNA-enabled devices
```shell
>>> dlna device
```

* resource cast to screen
```shell
>>> dlna play "your_link_or_path"
```

* select device by url
```shell
>>> dlna play "your_link_or_path" -d "your_device_url"

# device url like 'http://host:port/'
```

* select device by name
```shell
>>> dlna play "your_link_or_path" -q "your_device_name"

# device name like 'room tv'
```


Package: dlna
File: dlna-0.2.2/setup.py

# -*- coding: utf-8 -*-
# @author: leesoar
# @email: secure@tom.com
# @email2: employ@aliyun.com

from os.path import join, isfile
from os import walk
import io
import os
import sys
from shutil import rmtree
from setuptools import setup, Command


def read_file(filename):
    with open(filename) as fp:
        return fp.read().strip()


def read_requirements(filename):
    return [line.strip() for line in read_file(filename).splitlines()
            if not line.startswith('#')]


NAME = 'dlna'
FOLDER = 'dlna'
DESCRIPTION = 'A UPnP/DLNA client, support local file and online resource cast to screen.'
URL = 'https://leesoar.com/real'
EMAIL = 'core@111.com'
AUTHOR = 'leesoar'
REQUIRES_PYTHON = '>=3.3.0'

REQUIRED = read_requirements('requirements.txt')

here = os.path.abspath(os.path.dirname(__file__))

try:
    with io.open(os.path.join(here, 'README.md'), encoding='utf-8') as f:
        long_description = '\n' + f.read()
except FileNotFoundError:
    long_description = DESCRIPTION


def package_files(directories):
    paths = []
    for item in directories:
        if isfile(item):
            paths.append(join('..', item))
            continue
        for (path, directories, filenames) in walk(item):
            for filename in filenames:
                paths.append(join('..', path, filename))
    return paths


class UploadCommand(Command):
    description = 'Build and publish the package.'
    user_options = []

    @staticmethod
    def status(s):
        """Prints things in bold."""
        print('\033[1m{0}\033[0m'.format(s))

    def initialize_options(self):
        pass

    def finalize_options(self):
        pass

    def run(self):
        try:
            self.status('Removing previous builds…')
            rmtree(os.path.join(here, 'dist'))
        except OSError:
            pass

        self.status('Building Source and Wheel (universal) distribution…')
        os.system('{0} setup.py sdist bdist_wheel --universal'.format(sys.executable))

        self.status('Uploading the package to PyPI via Twine…')
        os.system('twine upload dist/*')

        sys.exit()


setup(
    name=NAME,
    version="0.2.2",
    description=DESCRIPTION,
    long_description=long_description,
    long_description_content_type='text/markdown',
    author=AUTHOR,
    author_email=EMAIL,
    python_requires=REQUIRES_PYTHON,
    url=URL,
    packages=['dlna'],
    package_data={'dlna': ['templates/*.xml']},
    install_requires=REQUIRED,
    include_package_data=True,
    license='MIT',
    entry_points={
        'console_scripts': ['dlna = dlna.client:run']
    },
    classifiers=[
        'License :: OSI Approved :: MIT License',
        'Programming Language :: Python :: 3.5',
        'Programming Language :: Python :: 3.6',
        'Programming Language :: Python :: 3.7',
        'Programming Language :: Python :: Implementation :: CPython',
        'Programming Language :: Python :: Implementation :: PyPy'
    ],
    cmdclass={
        'upload': UploadCommand,
    },
    keywords=[
        'dlna'
    ],
)



Package: dlna
File: dlna-0.2.2/setup.cfg

[egg_info]
tag_build = 
tag_date = 0