bridge.py

""" LimitlessLED Bridge. """

import queue
import socket
import select
import time
import threading
from datetime import datetime, timedelta

from limitlessled import MIN_WAIT, REPS
from limitlessled.group.rgbw import RgbwGroup, RGBW, BRIDGE_LED
from limitlessled.group.wrgb import WrgbGroup, WRGB
from limitlessled.group.rgbww import RgbwwGroup, RGBWW
from limitlessled.group.eightzone import EightzoneGroup, EIGHTZONE
from limitlessled.group.white import WhiteGroup, WHITE
from limitlessled.group.dimmer import DimmerGroup, DIMMER


BRIDGE_PORT = 5987
BRIDGE_VERSION = 6
BRIDGE_LED_GROUP = 1
BRIDGE_LED_NAME = 'bridge'
SELECT_WAIT = 0.025
BRIDGE_INITIALIZATION_COMMAND = [0x20, 0x00, 0x00, 0x00, 0x16, 0x02, 0x62,
                                 0x3a, 0xd5, 0xed, 0xa3, 0x01, 0xae, 0x08,
                                 0x2d, 0x46, 0x61, 0x41, 0xa7, 0xf6, 0xdc,
                                 0xaf, 0xfe, 0xf7, 0x00, 0x00, 0x1e]
KEEP_ALIVE_COMMAND_PREAMBLE = [0xD0, 0x00, 0x00, 0x00, 0x02]
KEEP_ALIVE_RESPONSE_PREAMBLE = [0xd8, 0x0, 0x0, 0x0, 0x07]
KEEP_ALIVE_TIME = 5
RECONNECT_TIME = 5
SOCKET_TIMEOUT = 5
STARTING_SEQUENTIAL_BYTE = 0x02


def group_factory(bridge, number, name, led_type):
    """ Make a group.

    :param bridge: Member of this bridge.
    :param number: Group number (1-4).
    :param name: Name of group.
    :param led_type: Either `RGBW`, `WRGB`, `RGBWW`, `WHITE`, `EIGHTZONE`, `DIMMER` or `BRIDGE_LED`.
    :returns: New group.
    """
    if led_type in [RGBW, BRIDGE_LED]:
        return RgbwGroup(bridge, number, name, led_type)
    elif led_type == RGBWW:
        return RgbwwGroup(bridge, number, name)
    elif led_type == EIGHTZONE:
        return EightzoneGroup(bridge, number, name)
    elif led_type == WHITE:
        return WhiteGroup(bridge, number, name)
    elif led_type == DIMMER:
        return DimmerGroup(bridge, number, name)
    elif led_type == WRGB:
        return WrgbGroup(bridge, number, name)
    else:
        raise ValueError('Invalid LED type: %s', led_type)


class Bridge(object):
    """ Represents a LimitlessLED bridge. """

    def __init__(self, ip, port=BRIDGE_PORT, version=BRIDGE_VERSION,
                 bridge_led_name=BRIDGE_LED_NAME):
        """ Initialize bridge.

        Bridge version 6 (latest as of this release)
        can use the default parameters. For lower versions,
        use port 8899 (3 to 5) or 50000 (lower then 3).
        Lower versions also require sending a larger payload
        to the bridge (slower).

        :param ip: IP address of bridge.
        :param port: Bridge port.
        :param version: Bridge version.
        :param bridge_led_name: Name of the bridge led group.
        """
        self.is_ready = True
        self.is_closed = False
        self.wait = MIN_WAIT
        self.reps = REPS
        self.groups = []
        self.ip = ip
        self.version = version
        self._sn = STARTING_SEQUENTIAL_BYTE
        self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self._socket.settimeout(SOCKET_TIMEOUT)
        self._socket.connect((ip, port))
        self._command_queue = queue.Queue()
        self._lock = threading.Lock()
        self.active = 0
        self._selected_number = None

        # Start queue consumer thread.
        consumer = threading.Thread(target=self._consume)
        consumer.daemon = True
        consumer.start()

        # Version specific stuff
        self._wb1 = None
        self._wb2 = None
        self._bridge_led = None
        if self.version >= 6:
            # Create bridge led group
            self._bridge_led = group_factory(self, BRIDGE_LED_GROUP,
                                             bridge_led_name, BRIDGE_LED)

            # Set is_ready to False before initing connection
            self.is_ready = False

            # Initialize connection to retrieve bridge session ids (wb1, wb2)
            self._init_connection()

            # Start keep alive thread.
            keep_alive_thread = threading.Thread(target=self._keep_alive)
            keep_alive_thread.daemon = True
            keep_alive_thread.start()

    @property
    def sn(self):
        """ Gets the current sequential byte. """
        return self._sn

    @property
    def wb1(self):
        """ Gets the bridge session id 1. """
        return self._wb1

    @property
    def wb2(self):
        """ Gets the bridge session id 2. """
        return self._wb2

    @property
    def bridge_led(self):
        """ Get the group to control the bridge led. """
        return self._bridge_led

    def incr_active(self):
        """ Increment number of active groups. """
        with self._lock:
            self.active += 1

    def decr_active(self):
        """ Decrement number of active groups. """
        with self._lock:
            self.active -= 1

    def add_group(self, number, name, led_type):
        """ Add a group.

        :param number: Group number (1-4).
        :param name: Group name.
        :param led_type: Either `RGBW`, `WRGB`, `RGBWW`, `WHITE`, `DIMMER` or `BRIDGE_LED`.
        :returns: Added group.
        """
        group = group_factory(self, number, name, led_type)
        self.groups.append(group)
        return group

    def send(self, command, reps=REPS, wait=MIN_WAIT):
        """ Send a command to the physical bridge.

        :param command: A Command instance.
        :param reps: Number of repetitions.
        :param wait: Wait time in seconds.
        """
        # Enqueue the command.
        self._command_queue.put((command, reps, wait))
        # Wait before accepting another command.
        # This keeps individual groups relatively synchronized.
        sleep = reps * wait * self.active
        if command.select and self._selected_number != command.group_number:
            sleep += SELECT_WAIT
        time.sleep(sleep)

    def _consume(self):
        """ Consume commands from the queue.

        The command is repeated according to the configured value.
        Wait after each command is sent.

        The bridge socket is a shared resource. It must only
        be used by one thread at a time. Note that this can and
        will delay commands if multiple groups are attempting
        to communicate at the same time on the same bridge.
        """
        while not self.is_closed:
            # Get command from queue.
            msg = self._command_queue.get()

            # Closed
            if msg is None:
                return

            # Use the lock so we are sure is_ready is not changed during execution
            # and the socket is not in use
            with self._lock:
                # Check if bridge is ready
                if self.is_ready:
                    (command, reps, wait) = msg

                    # Select group if a different group is currently selected.
                    if command.select and self._selected_number != command.group_number:
                        if self._send_raw(command.select_command.get_bytes(self)):
                            self._selected_number = command.group_number
                            time.sleep(SELECT_WAIT)
                        else:
                            # Stop sending on socket error
                            self.is_ready = False

                    # Repeat command as necessary.
                    for _ in range(reps):
                        if self.is_ready:
                            if self._send_raw(command.get_bytes(self)):
                                time.sleep(wait)
                            else:
                                # Stop sending on socket error
                                self.is_ready = False

            # Wait if bridge is not ready, we're only reading is_ready, no lock needed
            if not self.is_ready and not self.is_closed:
                # For older bridges, always try again, there's no keep-alive thread
                if self.version < 6:
                    # Give the reconnect some time
                    time.sleep(RECONNECT_TIME)
                    self.is_ready = True

    def _send_raw(self, command):
        """
        Sends an raw command directly to the physical bridge.
        :param command: A bytearray.
        """
        try:
            self._socket.send(bytearray(command))
            self._sn = (self._sn + 1) % 256
            return True
        except (socket.error, socket.timeout):
            # We can get a socket.error or timeout exception if the bridge is disconnected,
            # but we are still sending data. In that case, return False to indicate that data is not sent.
            return False

    def _init_connection(self):
        """
        Requests the session ids of the bridge.
        :returns: True, if initialization was successful. False, otherwise.
        """
        try:
            # We are changing self.is_ready: lock it up!
            self._lock.acquire()

            response = bytearray(22)
            self._send_raw(BRIDGE_INITIALIZATION_COMMAND)
            self._socket.recv_into(response)
            self._wb1 = response[19]
            self._wb2 = response[20]
            self.is_ready = True
        except (socket.error, socket.timeout):
            # Connection timed out, bridge is not ready for us
            self.is_ready = False
        finally:
            # Prevent deadlocks: always release the lock
            self._lock.release()

        return self.is_ready

    def _reconnect(self):
        """
        Try continuously to reconnect to the bridge.
        """
        while not self.is_closed:
            if self._init_connection():
                return

            time.sleep(RECONNECT_TIME)

    def _keep_alive(self):
        """
        Send keep alive messages continuously to bridge.
        """
        send_next_keep_alive_at = 0
        while not self.is_closed:
            if not self.is_ready:
                self._reconnect()
                continue

            if time.monotonic() > send_next_keep_alive_at:
                command = KEEP_ALIVE_COMMAND_PREAMBLE + [self.wb1, self.wb2]
                self._send_raw(command)
                need_response_by = time.monotonic() + KEEP_ALIVE_TIME

            # Wait for responses
            timeout = max(0, need_response_by - time.monotonic())
            ready = select.select([self._socket], [], [], timeout)
            if ready[0]:
                try:
                    response = bytearray(12)
                    self._socket.recv_into(response)

                    if response[:5] == bytearray(KEEP_ALIVE_RESPONSE_PREAMBLE):
                        send_next_keep_alive_at = need_response_by
                except (socket.error, socket.timeout):
                    with self._lock:
                        self.is_ready = False
            elif send_next_keep_alive_at < need_response_by:
                # Acquire the lock to make sure we don't change self.is_ready
                # while _consume() is sending commands
                with self._lock:
                    self.is_ready = False

    def close(self):
        """
        Closes the connection to the bridge.
        """
        self.is_closed = True
        self.is_ready = False
        self._command_queue.put(None)