chore: retooling for new web API

Author: kieranmanning

.envrc

@@ -0,0 +1,3 @@
+export OAUTH_CLIENT_ID=7v4bnRCzelHg6DdEtw6K81ayjTP3GJV0seFtsRv8
+export TOKEN_FILE_PATH=".token-file.json"
+export API_URI="http://localhost:8000"

.pre-commit-config.yaml

@@ -30,3 +30,5 @@ repos:
   rev: v1.19.1
   hooks:
   - id: mypy
+    additional_dependencies:
+    - 'types-requests'

.token-file.json

@@ -0,0 +1 @@
+{"access": {"token_value": "GuXYyK7iC0l4GOGE5N8pU8Q99z0aaf", "expiry_time": 1770967462.992627}, "refresh": {"token_value": "J8ZSFtUqKFPDKoer71bODFx4wyqXwR"}}

intercomclient/aws.py

@@ -1,19 +1,25 @@
 from dataclasses import dataclass
 from typing import Tuple
 import logging
+import os
 
 
 @dataclass()
 class Config:
     segment_duration: int = 60
     fps: int = 5
     resolution: Tuple[int, int] = (320, 240)
-    s3_bucket_name: str = "segment-uploads"
     debug_mode: bool = True
     video_source: int = 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")
+    max_polling_time_mins: int = int(os.getenv("MAX_POLLING_TIME_MINS", 5))
 
 
 logging.basicConfig(level=logging.DEBUG if Config.debug_mode else logging.INFO)

intercomclient/config.py

@@ -0,0 +1,9 @@
+import platform
+
+
+def get_os_info():
+    return platform.platform()
+
+
+def get_device_type():
+    return platform.machine()

intercomclient/device.py

@@ -0,0 +1,67 @@
+from .device import get_device_type, get_os_info
+from .config import Config
+from time import sleep
+from datetime import datetime, timedelta
+import requests
+
+
+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/"
+
+    api_response = requests.post(
+        url,
+        data={
+            "client_id": client_id,
+            "device_type": device_type,
+            "device_os": device_os,
+            "scope": "profile email",
+        },
+    )
+    return api_response.json()
+
+
+def poll_for_token(config: Config, device_code: str, interval: int) -> dict:
+    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/"
+    while continue_polling:
+        if datetime.now() > polling_end_time:
+            continue_polling = False
+            break
+        sleep(interval)
+        api_response = requests.post(
+            url,
+            data={
+                "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
+                "device_code": device_code,
+                "client_id": client_id,
+            },
+        )
+        resp_json = api_response.json()
+        if "error" in resp_json and resp_json["error"] == "authorization_pending":
+            continue
+        else:
+            return api_response.json()
+
+    return api_response.json()
+
+
+def refresh_tokens(config: Config, refresh_token) -> dict:
+    client_id = Config.oauth_client_id
+    url = f"{Config.api_uri}/oauth/token/"
+
+    api_response = requests.post(
+        url,
+        data={
+            "grant_type": "refresh_token",
+            "client_id": client_id,
+            "refresh_token": refresh_token,
+        },
+    )
+
+    return api_response.json()

intercomclient/device_authorization.py

@@ -0,0 +1,19 @@
+import json
+from .config import Config
+
+
+def load_tokens():
+    with open(Config.token_file_path, "r") as token_file_handle:
+        return json.load(token_file_handle)
+
+
+def store_tokens(tokens: dict[str, str], access_token_expiry):
+    data = {
+        "access": {
+            "token_value": tokens["access_token"],
+            "expiry_time": access_token_expiry,
+        },
+        "refresh": {"token_value": tokens["refresh_token"]},
+    }
+    with open(Config.token_file_path, "w") as token_file_handle:
+        token_file_handle.write(json.dumps(data))

intercomclient/local_store.py

@@ -1,6 +1,39 @@
-from intercomclient.video_pipeline import VideoPipeline
+from intercomclient.device_authorization import (
+    initiate_device_authorization,
+    poll_for_token,
+    refresh_tokens,
+)
 from intercomclient.config import Config
+import logging
+from datetime import datetime, timedelta, UTC
+from intercomclient.local_store import store_tokens, load_tokens
 
-if __name__ == "__main__":
-    pipeline = VideoPipeline(config=Config)
-    pipeline.record_segment()
+logging.basicConfig(level=logging.INFO)
+LOG = logging.getLogger(__name__)
+config = Config()
+
+authorization_response = initiate_device_authorization(config)
+LOG.info(f"User code: {authorization_response['user_code']}")
+
+token_response = poll_for_token(
+    config,
+    authorization_response["device_code"],
+    interval=authorization_response["interval"],
+)
+print(token_response)
+
+token_expiry = (
+    datetime.now(tz=UTC) + timedelta(seconds=token_response["expires_in"])
+).timestamp()
+
+store_tokens(
+    {
+        "access_token": token_response["access_token"],
+        "refresh_token": token_response["refresh_token"],
+    },
+    access_token_expiry=token_expiry,
+)
+
+refresh_token_response = refresh_tokens(config, load_tokens()["refresh"]["token_value"])
+
+print(refresh_token_response)

main.py

@@ -8,7 +8,10 @@ dependencies = [
     "boto3>=1.42.39",
     "numpy>=2.4.1",
     "opencv-python>=4.13.0.90",
+    "pip==26.0",
+    "requests>=2.32.5",
     "ruff>=0.14.14",
+    "transparenc-sdk",
 ]
 
 [dependency-groups]
@@ -17,3 +20,6 @@ dev = [
     "pytest>=9.0.2",
     "pytest-watch>=4.2.0",
 ]
+
+[tool.uv.sources]
+transparenc-sdk = { url = "https://github.com/kieranmanning/IntercomClientSDKs/releases/download/v0.9/transparenc_sdk-1.0.0-py3-none-any.whl" }