wizlight.py

'''
     MIT License

     Copyright (c) 2020 Stephan Traub

     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.
'''
import socket
import json
import asyncio
import asyncio_dgram
import logging
from time import time
from .scenes import SCENES

_LOGGER = logging.getLogger(__name__)

class PilotBuilder:
    def __init__ (self, warm_white = None,
        cold_white = None,
        speed = None,
        scene = None,
        rgb = None,
        brightness = None,
        colortemp = None):

        self.pilot_params = {"state": True}

        if warm_white != None:
            self._set_warm_white(warm_white)
        if cold_white != None:
            self._set_cold_white(cold_white)
        if speed != None:
            self._set_speed(speed)
        if scene != None:
            self._set_scene(scene)
        if rgb != None:
            self._set_rgb(rgb)
        if brightness != None:
            self._set_brightness(brightness)
        if colortemp != None:
            self._set_colortemp(colortemp)

    def get_pilot_message(self):
        return json.dumps({"method": "setPilot",
                  "params": self.pilot_params })
    
    def _set_warm_white(self, value: int):
        '''
            set the value of the cold white led
        '''
        if value > 0 and value < 256:
            self.pilot_params['w'] = value


    def _set_cold_white(self, value: int):
        '''
            set the value of the cold white led
        '''
        if value > 0 and value < 256:
            self.pilot_params['c'] = value
        else:
            raise IndexError("Value must be between 1 and 255")

    def _set_speed(self, value: int):
        '''
            set the color changing speed in precent (0-100)
            This applies only to changing effects!
        '''
        if value > 0 and value < 101:
            self.pilot_params['speed'] = value
        else:
            raise IndexError("Value must be between 0 and 100")

    def _set_scene(self, scene_id: int):
        '''
            set the scene by id
        '''
        if scene_id in SCENES:
            self.pilot_params['sceneId'] = scene_id
        else:
            # id not in SCENES !
            raise IndexError("Scene is not available - only 0 to 32 are supported")

    def _set_rgb(self, values):
        '''
            set the rgb color state of the bulb
        '''
        r, g, b = values
        self.pilot_params['r'] = r
        self.pilot_params['g'] = g
        self.pilot_params['b'] = b

    def _set_brightness(self, value: int):
        '''
            set the value of the brightness 0-255
        '''
        percent = self.hex_to_percent(value)
        # lamp doesn't supports lower than 10%
        if percent < 10: percent = 10
        self.pilot_params['dimming'] = percent
    
    def _set_colortemp(self, kelvin: int):
        '''
            sets the color temperature for the white led in the bulb
        '''
        # normalize the kelvin values - should be removed
        if kelvin < 2500: kelvin = 2500
        if kelvin > 6500: kelvin = 6500 
        
        self.pilot_params['temp'] = kelvin

    def hex_to_percent(self, hex):
        ''' converts hex 0-255 values to percent '''
        return round((hex/255)*100)

class PilotParser:
    def __init__ (self, pilotResult):
        self.pilotResult = pilotResult

    def get_state(self) -> str:
        return self.pilotResult['state']

    def get_warm_white(self) -> int:
        ''' 
            get the value of the warm white led 
        '''
        if "w" in self.pilotResult:
            return self.pilotResult['w']
        else:
            return None

    def get_speed(self) -> int:
        ''' 
            get the color changing speed
        '''
        if "speed" in self.pilotResult:
            return self.pilotResult['speed']
        else:
            return None

    def get_scene(self) -> str:
        '''
            get the current scene name
        '''
        if 'schdPsetId' in self.pilotResult: #rhythm
            return SCENES[1000]

        id = self.pilotResult['sceneId']
        if id in SCENES:
            return SCENES[id]
        else:
            return None

    def get_cold_white(self) -> int:
        '''
            get the value of the cold white led
        '''
        if "c" in self.pilotResult:
            return self.pilotResult['c']
        else:
            return None
            
    def get_rgb(self):
        '''
            get the rgb color state of the bulb and turns it on
        '''
        if "r" in self.pilotResult and "g" in self.pilotResult and "b" in self.pilotResult:
            r = self.pilotResult['r']
            g = self.pilotResult['g']
            b = self.pilotResult['b']
            return r, g, b
        else:
            # no RGB color value was set
            return None, None, None

    def get_brightness(self) -> int:
        '''
            gets the value of the brightness 0-255
        '''
        return self.percent_to_hex(self.pilotResult['dimming'])

    def get_colortemp(self) -> int:
        if "temp" in self.pilotResult:
            return self.pilotResult['temp']
        else:
            return None

    def percent_to_hex(self, percent):
        ''' converts percent values 0-100 into hex 0-255'''
        return round((percent / 100)*255)

class wizlight:
    '''
        Creates a instance of a WiZ Light Bulb
    '''
    # default port for WiZ lights

    def __init__ (self, ip, port=38899):
        ''' Constructor with ip of the bulb '''
        self.ip = ip
        self.port = port
        self.state = None

    @property
    def status(self) -> bool:
        '''
            returns true or false / true = on , false = off
        '''
        if self.state == None:
            return None
        return self.state.get_state()

    ## ------------------ Non properties --------------

    async def turn_off(self):
        '''
            turns the light off
        '''
        message = r'{"method":"setPilot","params":{"state":false}}'
        await self.sendUDPMessage(message)

    async def turn_on(self, pilot_builder = PilotBuilder()):
        '''
            turns the light on
        '''
        await self.sendUDPMessage(pilot_builder.get_pilot_message(), max_send_datagrams = 10)
    
    def get_id_from_scene_name(self, scene: str) -> int: 
        ''' gets the id from a scene name '''
        for id in SCENES:
            if SCENES[id] == scene:
                return id
        raise ValueError("Scene '%s' not in scene list." % scene)

    ### ---------- Helper Functions ------------
    async def updateState(self):
        '''
            Note: Call this method before getting any other property
            Also, call this method to update the current value for any property
            getPilot - gets the current bulb state - no paramters need to be included
            {"method": "getPilot", "id": 24}
        '''
        message = r'{"method":"getPilot","params":{}}'
        resp = await self.sendUDPMessage(message)
        if resp != None and 'result' in resp:
            self.state = PilotParser(resp['result'])
        else:
            self.state = None
        return self.state

    async def getBulbConfig(self):
        '''
            returns the configuration from the bulb
        '''
        message = r'{"method":"getSystemConfig","params":{}}'
        return await self.sendUDPMessage(message)
    
    async def lightSwitch(self):
        '''
            turns the light bulb on or off like a switch
        '''
        # first get the status
        state = await self.updateState()
        if state.get_state():
            # if the light is on - switch off
            await self.turn_off()
        else:
            # if the light is off - turn on
            await self.turn_on()


    async def receiveUDPwithTimeout(self, stream, timeout):
        data, remote_addr = await asyncio.wait_for(stream.recv(), timeout)
        return data

    async def sendUDPMessage(self, message, timeout = 60, send_interval = 0.5, max_send_datagrams = 100):
        '''
            send the udp message to the bulb
        '''
        connid = hex(int(time()*10000000))[2:]
        data = None

        try:
            _LOGGER.debug("[wizlight {}, connid {}] connecting to UDP port".format(self.ip, connid))
            stream = await asyncio.wait_for(asyncio_dgram.connect((self.ip, self.port)), timeout)
            _LOGGER.debug("[wizlight {}, connid {}] listening for response datagram".format(self.ip, connid))

            receive_task = asyncio.create_task(self.receiveUDPwithTimeout(stream, timeout))

            i = 0
            while not receive_task.done() and i < max_send_datagrams:
                _LOGGER.debug("[wizlight {}, connid {}] sending command datagram {}: {}".format(self.ip, connid, i, message))
                asyncio.create_task(stream.send(bytes(message, "utf-8")))
                await asyncio.sleep(send_interval)
                i += 1

            await receive_task
            data = receive_task.result()
            
        except asyncio.TimeoutError:
            _LOGGER.exception("[wizlight {}, connid {}] Failed to do UDP call(s) to wiz light".format(self.ip, connid))
        finally:
            stream.close()

        if data != None and len(data) is not None:
            resp = json.loads(data.decode())
            if "error" not in resp:
                _LOGGER.debug("[wizlight {}, connid {}] response received: {}".format(self.ip, connid, resp))
                return resp
            else:
                # exception should be created
                raise ValueError("Cant read response from the bulb. Debug:",resp)