feat: webRTC

Author: kieranmanning

.envrc

@@ -1,3 +1,7 @@
-export OAUTH_CLIENT_ID=7v4bnRCzelHg6DdEtw6K81ayjTP3GJV0seFtsRv8
-export TOKEN_FILE_PATH=".token-file.json"
-export API_URI="http://localhost:8000"
+export OAUTH_CLIENT_ID=K7asmp280fyVAfgx12uCcqauaVIFnb2Py9DpwCn1
+export VIDEO_SOURCE=0
+export TOKEN_FILE_PATH="~/.config/intercom-client/tokens.json"
+export WEBSOCKET_API_BASE_URL="ws://localhost:8000/ws/live_stream"
+export HTTP_API_BASE_URL="http://localhost:8000"
+
+source .venv/bin/activate

.token-file.json

@@ -0,0 +1,31 @@
+import cv2
+import logging
+from intercomclient.config import Config
+from aiortc import VideoStreamTrack
+import av
+import asyncio
+
+LOG = logging.getLogger(__name__)
+
+
+class CameraVideoStreamTrack(VideoStreamTrack):
+    def __init__(self, config=Config):
+        self.config = config
+        self.target_fps = config.target_fps
+        self.capture = cv2.VideoCapture(config.video_source)
+        if not self.capture.isOpened():
+            raise ValueError("Unable to open video source")
+        self.segment_number = 0
+        super().__init__()
+
+    async def recv(self):
+        pts, time_base = await self.next_timestamp()
+        ret, frame = self.capture.read()
+        if not ret:
+            await asyncio.sleep(1 / self.target_fps)
+            return await self.recv()
+
+        frame = av.VideoFrame.from_ndarray(frame, format="bgr24")
+        frame.pts = pts
+        frame.time_base = time_base
+        return frame

intercomclient/camera_video_stream_track.py

@@ -1,24 +1,28 @@
 from dataclasses import dataclass
 from typing import Tuple
+from pathlib import Path
 import logging
 import os
 
 
 @dataclass()
 class Config:
     segment_duration: int = 60
-    fps: int = 5
+    target_fps: int = 5
     resolution: Tuple[int, int] = (320, 240)
     debug_mode: bool = True
-    video_source: int = 0
+    video_source: int = int(os.getenv("VIDEO_SOURCE", 0))
     output_dir_path: str = "/tmp/intercom_videos"
     fourcc: str = "XVID"
     video_format: str = "avi"
     oauth_scope: str = "openid email profile"
     oauth_grant: str = "urn:ietf:params:oauth:grant-type:device_code"
     oauth_client_id: str = os.getenv("OAUTH_CLIENT_ID", "wrong")
-    token_file_path: str = os.getenv("TOKEN_FILE_PATH", "wrong")
-    api_uri: str = os.getenv("API_URI", "wrong")
+    token_file_path: Path = Path(
+        os.getenv("TOKEN_FILE_PATH", "~/.config/intercomclient/tokens.json")
+    ).expanduser()
+    http_api_base_url: str = os.getenv("HTTP_API_BASE_URL", "wrong")
+    websocket_api_base_url: str = os.getenv("WEBSOCKET_API_BASE_URL", "wrong")
     max_polling_time_mins: int = int(os.getenv("MAX_POLLING_TIME_MINS", 5))
 
 

intercomclient/config.py

@@ -2,8 +2,8 @@
 
 
 def get_os_info():
-    return platform.platform()
+    return str(platform.platform())[0:30]
 
 
 def get_device_type():
-    return platform.machine()
+    return str(platform.machine())[0:30]

intercomclient/device.py

@@ -9,7 +9,7 @@ def initiate_device_authorization(config: Config) -> dict:
     client_id = Config.oauth_client_id
     device_type = get_device_type()
     device_os = get_os_info()
-    url = f"{Config.api_uri}/api/v1/oauth/device-authorization/"
+    url = f"{config.http_api_base_url}/oauth/device-authorization/"
 
     api_response = requests.post(
         url,
@@ -24,23 +24,24 @@ def initiate_device_authorization(config: Config) -> dict:
 
 
 def poll_for_token(config: Config, device_code: str, interval: int) -> dict:
-    client_id = Config.oauth_client_id
+    client_id = config.oauth_client_id
 
     continue_polling = True
     polling_end_time = datetime.now() + timedelta(minutes=config.max_polling_time_mins)
-    url = f"{Config.api_uri}/oauth/token/"
+    url = f"{config.http_api_base_url}/oauth/token/"
     while continue_polling:
         if datetime.now() > polling_end_time:
             continue_polling = False
             break
         sleep(interval)
+        post_data = {
+            "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
+            "device_code": device_code,
+            "client_id": client_id,
+        }
         api_response = requests.post(
             url,
-            data={
-                "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
-                "device_code": device_code,
-                "client_id": client_id,
-            },
+            data=post_data,
         )
         resp_json = api_response.json()
         if "error" in resp_json and resp_json["error"] == "authorization_pending":
@@ -53,7 +54,7 @@ def poll_for_token(config: Config, device_code: str, interval: int) -> dict:
 
 def refresh_tokens(config: Config, refresh_token) -> dict:
     client_id = Config.oauth_client_id
-    url = f"{Config.api_uri}/oauth/token/"
+    url = f"{config.http_api_base_url}/oauth/token/"
 
     api_response = requests.post(
         url,

intercomclient/device_authorization.py

@@ -0,0 +1,107 @@
+import json
+import logging
+import os
+from .config import Config
+import datetime
+from enum import Enum
+
+LOG = logging.getLogger(__name__)
+
+
+class TokenStatus(Enum):
+    VALID = 1
+    EXPIRED = 2
+    MISSING = 3
+    EXPIRING = 4
+
+
+class CannotLoadTokensException(Exception):
+    pass
+
+
+class TokenStore:
+    """
+    Class to manage token storage and access
+    """
+
+    def __init__(self, config: Config, verify=True):
+        self.config = config
+
+        if verify:
+            self.verify_token_file_path()
+
+    @classmethod
+    def check_token_expiry_delta(cls, token_expiry_time) -> datetime.timedelta:
+        current_time = datetime.datetime.now().timestamp()
+        return datetime.timedelta(seconds=current_time - token_expiry_time)
+
+    def load_tokens(self):
+        self.verify_token_file_path()
+        with open(self.config.token_file_path, "r") as token_file_handle:
+            token_file_json = json.load(token_file_handle)
+            try:
+                access_token = token_file_json["access"]["token_value"]
+                refresh_token = token_file_json["refresh"]["token_value"]
+                access_token_expiry = token_file_json["access"]["expiry_time"]
+                device_code = token_file_json["device_code"]
+
+                return {
+                    "access": {
+                        "token_value": access_token,
+                        "expiry_time": access_token_expiry,
+                    },
+                    "refresh": {"token_value": refresh_token},
+                    "device_code": device_code,
+                }
+
+            except KeyError as e:
+                error_message = f"Token file is missing keys, will assume empty {self.config.token_file_path}"
+                raise CannotLoadTokensException(error_message) from e
+
+    def store_tokens(self, tokens: dict[str, str], access_token_expiry, device_code):
+        self.verify_token_file_path()
+
+        data = {
+            "access": {
+                "token_value": tokens["access_token"],
+                "expiry_time": access_token_expiry,
+            },
+            "refresh": {"token_value": tokens["refresh_token"]},
+            "device_code": device_code,
+        }
+        with open(self.config.token_file_path, "w") as token_file_handle:
+            token_file_handle.write(json.dumps(data))
+
+    def verify_token_file_path(self) -> bool:
+        token_file_path = self.config.token_file_path
+        token_parent_dir_path = os.path.split(token_file_path)[0]
+        # Check our dir path exists, and if not attempt to create it
+        if not os.path.isdir(token_parent_dir_path):
+            try:
+                os.makedirs(token_parent_dir_path, exist_ok=True)
+            except OSError as e:
+                error_message = f"Error: could not find or create token file parent directory: {token_file_path}"
+                raise OSError(error_message) from e
+
+        # Check our file exists and has the correct permissions
+        if not os.path.exists(token_file_path):
+            token_fd = os.open(
+                token_file_path,
+                os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
+                0o600,
+            )
+            example_file_outline = {
+                "access": {"token_value": None, "expiry_time": None},
+                "refresh": {"token_value": None},
+                "device_code": None,
+            }
+            with os.fdopen(token_fd, "w") as file_handle:
+                file_handle.write(json.dumps(example_file_outline))
+        else:
+            # If the file already exists, we want to check it has the correct permissions
+            current_permissions = oct(os.stat(token_file_path).st_mode)[-3:]
+            if current_permissions != "600":
+                error_message = f"Error: token file has incorrect permissions: {token_file_path}, current permissions: {current_permissions}"
+                raise OSError(error_message)
+
+        return True