[WIP] Feature/typings

examples/blinds.py

@@ -1,30 +1,42 @@
-from sinric import SinricPro, SinricProConstants
+
 import asyncio
+from collections.abc import Callable
+from numbers import Real
+from typing import Final, Union
+
+from sinric import SinricPro, SinricProConstants
+
+CallbackFunctions = Union[
+    Callable[[str, str], tuple[bool, str]],  # Power state
+    Callable[[str, Real, str], tuple[bool, Real, str]]  # Range Value
+]
+
+APP_KEY: Final[str] = ''
+APP_SECRET: Final[str] = ''
+BLINDS_ID: Final[str] = ''
 
-APP_KEY = ''
-APP_SECRET = ''
-BLINDS_ID = ''
+# TODO: got this from _sinricpro_constants.py - POWER_STATE_ON is a str, could we consider changing it to a strEnum?
 
 
-def set_power_state(device_id, state):
+def set_power_state(device_id: str, state: str) -> tuple[bool, str]:
     print('device_id: {} state: {}'.format(device_id, state))
     return True, state
 
 
-def set_range_value(device_id, value, instance_id):
+def set_range_value(device_id: str, value: Real, instance_id: str) -> tuple[bool, Real, str]:
     print('device_id: {} set to: {}'.format(device_id, value))
     return True, value, instance_id
 
 
-callbacks = {
+callbacks: dict[str, CallbackFunctions] = {
     SinricProConstants.SET_POWER_STATE: set_power_state,
     SinricProConstants.SET_RANGE_VALUE: set_range_value
 }
 
 if __name__ == '__main__':
-    loop = asyncio.get_event_loop()
-    client = SinricPro(APP_KEY, [BLINDS_ID], callbacks,
-                       enable_log=False, restore_states=False, secret_key=APP_SECRET)
+    loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
+    client: SinricPro = SinricPro(APP_KEY, [BLINDS_ID], callbacks,
+                                  enable_log=False, restore_states=False, secret_key=APP_SECRET)
     loop.run_until_complete(client.connect())
 
 # client.event_handler.raise_event(BLINDS_ID, SinricProConstants.SET_RANGE_VALUE, data = {SinricProConstants.RANGE_VALUE: 30 })

examples/camera.py

@@ -1,51 +1,66 @@
+from collections.abc import Callable
+from typing import ByteString, Final, Optional, Union
 from sinric import SinricPro, SinricProConstants
 import requests
 import asyncio
 import base64
 
-APP_KEY = ""
-APP_SECRET = ""
-CAMERA_ID = ''
+APP_KEY: Final[str] = ''
+APP_SECRET: Final[str] = ''
+CAMERA_ID: Final[str] = ''
 
-def get_webrtc_answer(device_id, offer):
-    sdp_offer = base64.b64decode(offer)
+OfferType = Union[str, ByteString]
+
+CallbackFunctions = Union[
+    Callable[[str, str], tuple[bool, str]],  # Power state / Camera Stream Url
+    Callable[[str, OfferType], tuple[bool, Optional[bytes]]],  # WEBRTC Answer
+
+]
+
+
+def get_webrtc_answer(device_id: str, offer: OfferType) -> tuple[bool, Optional[bytes]]:
+    sdp_offer: Final[bytes] = base64.b64decode(offer)
     print('device_id: {} offer: {}'.format(device_id, offer))
 
     # PORT 8889 for WebRTC. eg: for PiCam, use http://<mediamtx-hostname>:8889/cam/whep
-    mediamtx_url = "http://<mediamtx-hostname>:8889/<device>/whep"  
-    headers = {"Content-Type": "application/sdp"}
-    response = requests.post(mediamtx_url, headers=headers, data=sdp_offer)
+    mediamtx_url: Final[str] = "http://<mediamtx-hostname>:8889/<device>/whep"
+    headers: dict[str, str] = {"Content-Type": "application/sdp"}
+    response: requests.Response = requests.post(
+        mediamtx_url, headers=headers, data=sdp_offer)
 
     if response.status_code == 201:
-        answer = base64.b64encode(response.content).decode("utf-8")
+        answer: bytes = base64.b64encode(response.content).decode("utf-8")
         return True, answer
     else:
-        return False
+        return False, None
 
 
-def power_state(device_id, state):
+def power_state(device_id: str, state: str) -> tuple[bool, str]:
     print('device_id: {} power state: {}'.format(device_id, state))
     return True, state
 
-def get_camera_stream_url(device_id, protocol):
+
+def get_camera_stream_url(device_id: str, protocol: str) -> tuple[bool, str]:
+    # TODO: Should protocol be a string literal?
     # Google Home: RTSP protocol not supported. Requires a Chromecast TV or Google Nest Hub
     # Alexa: RTSP url must be interleaved TCP on port 443 (for both RTP and RTSP) over TLS 1.2 port 443
 
     print('device_id: {} protocol: {}'.format(device_id, protocol))
 
     if protocol == "rstp":
-        return True, 'rtsp://rtspurl:443'   # RSTP. 
+        return True, 'rtsp://rtspurl:443'   # RSTP.
     else:
-        return True, 'https://cph-p2p-msl.akamaized.net/hls/live/2000341/test/master.m3u8' # HLS
+        return True, 'https://cph-p2p-msl.akamaized.net/hls/live/2000341/test/master.m3u8'  # HLS
+
 
-callbacks = {
+callbacks: dict[str, CallbackFunctions] = {
     SinricProConstants.GET_WEBRTC_ANSWER: get_webrtc_answer,
     SinricProConstants.GET_CAMERA_STREAM_URL: get_camera_stream_url,
     SinricProConstants.SET_POWER_STATE: power_state
 }
 
 if __name__ == '__main__':
-    loop = asyncio.get_event_loop()
-    client = SinricPro(APP_KEY, [CAMERA_ID], callbacks,
-                       enable_log=False, restore_states=False, secret_key=APP_SECRET)
+    loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
+    client: SinricPro = SinricPro(APP_KEY, [CAMERA_ID], callbacks,
+                                  enable_log=False, restore_states=False, secret_key=APP_SECRET)
     loop.run_until_complete(client.connect())

examples/custom_device.py

@@ -1,34 +1,44 @@
+from collections.abc import Callable
+from numbers import Real
+from typing import Any, Final, Union
 from sinric import SinricPro, SinricProConstants
 import asyncio
 
-APP_KEY = ''
-APP_SECRET = ''
-DEVICE_ID = ''
+APP_KEY: Final[str] = ''
+APP_SECRET: Final[str] = ''
+DEVICE_ID: Final[str] = ''
 
+CallbackFunctions = Union[
+    Callable[[str, str], tuple[bool, str]],  # Power state
+    Callable[[str, Real, str], tuple[bool, Real, str]],  # Range Value
+    Callable[[str, Any, str], tuple[bool, Any, str]]  # Mode Value
+]
 
-def power_state(device_id, state):
+
+def power_state(device_id: str, state: str) -> tuple[bool, str]:
     print(device_id, state)
     return True, state
 
 
-def range_value(device_id, range_value, instance_id):
+def range_value(device_id: str, range_value: Real, instance_id: str) -> tuple[bool, Real, str]:
     print(device_id, range_value, instance_id)
     return True, range_value, instance_id
 
 
-def mode_value(device_id, mode_value, instance_id):
+def mode_value(device_id: str, mode_value: Any, instance_id: str) -> tuple[bool, Any, str]:
+    # TODO: what is mode_Value? str?Enum?
     print(device_id, mode_value, instance_id)
     return True, mode_value, instance_id
 
 
-callbacks = {
+callbacks: dict[str, CallbackFunctions] = {
     SinricProConstants.SET_POWER_STATE: power_state,
     SinricProConstants.SET_RANGE_VALUE: range_value,
     SinricProConstants.SET_MODE: mode_value
 }
 
 if __name__ == '__main__':
-    loop = asyncio.get_event_loop()
-    client = SinricPro(APP_KEY, [DEVICE_ID], callbacks,
-                       enable_log=False, restore_states=False, secret_key=APP_SECRET)
+    loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
+    client: SinricPro = SinricPro(APP_KEY, [DEVICE_ID], callbacks,
+                                  enable_log=False, restore_states=False, secret_key=APP_SECRET)
     loop.run_until_complete(client.connect())

examples/dim_switch.py

@@ -1,30 +1,38 @@
+from collections.abc import Callable
+from numbers import Real
+from typing import Final, Union
 from sinric import SinricPro, SinricProConstants
 import asyncio
 
-APP_KEY = ''
-APP_SECRET = ''
-DIM_SWITCH_ID = ''
+APP_KEY: Final[str] = ''
+APP_SECRET: Final[str] = ''
+DIM_SWITCH_ID: Final[str] = ''
 
+CallbackFunctions = Union[
+    Callable[[str, str], tuple[bool, str]],  # Power state
+    Callable[[str, Real], tuple[bool, Real]]  # Power Level
+]
 
-def power_state(device_id, state):
+
+def power_state(device_id: str, state: str) -> tuple[bool, str]:
     print('device_id: {} state: {}'.format(device_id, state))
     return True, state
 
 
-def power_level(device_id, power_level):
+def power_level(device_id: str, power_level: Real) -> tuple[bool, Real]:
     print('device_id: {} power level: {}'.format(device_id, power_level))
     return True, power_level
 
 
-callbacks = {
+callbacks: dict[str, CallbackFunctions] = {
     SinricProConstants.SET_POWER_STATE: power_state,
     SinricProConstants.SET_POWER_LEVEL: power_level
 }
 
 if __name__ == '__main__':
-    loop = asyncio.get_event_loop()
-    client = SinricPro(APP_KEY, [DIM_SWITCH_ID], callbacks,
-                       enable_log=False, restore_states=False, secret_key=APP_SECRET)
+    loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
+    client: SinricPro = SinricPro(APP_KEY, [DIM_SWITCH_ID], callbacks,
+                                  enable_log=False, restore_states=False, secret_key=APP_SECRET)
     loop.run_until_complete(client.connect())
 
     # client.event_handler.raise_event(DIM_SWITCH_ID, SinricProConstants.SET_POWER_LEVEL, data = {SinricProConstants.POWER_LEVEL: 50 })

examples/doorbell.py

@@ -1,9 +1,10 @@
+from typing import Final
 from sinric import SinricPro, SinricProConstants
 import asyncio
 
-APP_KEY = ''
-APP_SECRET = ''
-DOORBELL_ID = ''
+APP_KEY: Final[str] = ''
+APP_SECRET: Final[str] = ''
+DOORBELL_ID: Final[str] = ''
 
 '''
 DON'T FORGET TO TURN ON 'Doorbell Press' IN ALEXA APP
@@ -19,9 +20,9 @@ async def events():
 callbacks = {}
 
 if __name__ == '__main__':
-    loop = asyncio.get_event_loop()
-    client = SinricPro(APP_KEY, [DOORBELL_ID], callbacks, event_callbacks=events,
-                       enable_log=True, restore_states=False, secret_key=APP_SECRET)
+    loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
+    client: SinricPro = SinricPro(APP_KEY, [DOORBELL_ID], callbacks, event_callbacks=events,
+                                  enable_log=True, restore_states=False, secret_key=APP_SECRET)
     loop.run_until_complete(client.connect())
 
     # client.event_handler.raise_event(DOORBELL_ID, SinricProConstants.DOORBELLPRESS)

examples/garagedoor.py

@@ -1,30 +1,38 @@
-from sinric import SinricPro, SinricProConstants
 import asyncio
+from collections.abc import Callable
+from typing import Any, Final, Union
+
+from sinric import SinricPro, SinricProConstants
+
+APP_KEY: Final[str] = ''
+APP_SECRET: Final[str] = ''
+GARAGEDOOR_ID: Final[str] = ''
 
-APP_KEY = ''
-APP_SECRET = ''
-GARAGEDOOR_ID = ''
+CallbackFunctions = Union[
+    Callable[[str, str], tuple[bool, str]],  # Power state
+    Callable[[str, Any, str], tuple[bool, Any, str]]  # Mode Value
+]
 
 
-def power_state(device_id, state):
+def power_state(device_id: str, state: str) -> tuple[bool, str]:
     print('device_id: {} state: {}'.format(device_id, state))
     return True, device_id
 
 
-def set_mode(device_id, state, instance_id):
+def set_mode(device_id: str, state: Any, instance_id: str) -> tuple[bool, Any, str]:
     print('device_id: {} mode: {}'.format(device_id, state))
     return True, state, instance_id
 
 
-callbacks = {
+callbacks:  dict[str, CallbackFunctions] = {
     SinricProConstants.SET_MODE: set_mode,
     SinricProConstants.SET_POWER_STATE: power_state
 }
 
 if __name__ == '__main__':
-    loop = asyncio.get_event_loop()
-    client = SinricPro(APP_KEY, [GARAGEDOOR_ID], callbacks,
-                       enable_log=False, restore_states=False, secret_key=APP_SECRET)
+    loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
+    client: SinricPro = SinricPro(APP_KEY, [GARAGEDOOR_ID], callbacks,
+                                  enable_log=False, restore_states=False, secret_key=APP_SECRET)
     loop.run_until_complete(client.connect())
 
     # client.event_handler.raise_event(GARAGEDOOR_ID, SinricProConstants.SET_MODE, data = {SinricProConstants.MODE: SinricProConstants.OPEN })