tina.schellander/elgato-cli
1
0
Fork 0
Code Pull Requests Actions Packages Releases Wiki Activity

Fix HTTP compatibility with Elgato Key Light firmware

Browse Source 
The light's TCP stack resets HTTP/1.1 GET connections but requires
HTTP/1.1 for PUT requests. Switch from requests to raw http.client
with HTTP/1.0 for GETs and HTTP/1.1 + Connection: close for PUTs.
Add retry logic (5 attempts, 1s delay) for intermittent resets.
This commit is contained in:
Tina Schellander
2026-02-07 11:33:11 +01:00
parent ca36b15af1
commit df7653dcba
2 changed files with 114 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/elgato_cli/api.py
+61 -13
View File
@@ -1,7 +1,9 @@
"""HTTP client for the Elgato Key Light REST API."""
import http.client
import json as _json
import time
from dataclasses import dataclass
import requests
@@ -21,14 +23,66 @@ def kelvin_to_mired(kelvin: int) -> int:
class ElgatoLight:
RETRIES = 5
RETRY_DELAY = 1.0
def __init__(self, host: str, port: int = 9123):
self.host = host
self.port = port
self.base_url = f"http://{host}:{port}/elgato"
def _request(self, method: str, path: str, **kwargs):
"""HTTP request with retries for flaky Elgato TCP stack.
The Elgato Key Light firmware resets HTTP/1.1 GET connections but
requires HTTP/1.1 for PUT. We use HTTP/1.0 for reads, HTTP/1.1 with
Connection: close for writes.
"""
timeout = kwargs.pop("timeout", 5)
json_body = kwargs.pop("json", None)
for attempt in range(self.RETRIES):
try:
conn = http.client.HTTPConnection(self.host, self.port, timeout=timeout)
# HTTP/1.0 for GET (light resets HTTP/1.1 GETs)
# HTTP/1.1 for PUT (light rejects HTTP/1.0 PUTs)
if method == "GET":
conn._http_vsn = 10
conn._http_vsn_str = "HTTP/1.0"
headers = {"Connection": "close"}
body = None
if json_body is not None:
body = _json.dumps(json_body)
headers["Content-Type"] = "application/json"
conn.request(method, f"/elgato/{path}", body=body, headers=headers)
resp = conn.getresponse()
data = resp.read()
conn.close()
if resp.status >= 400:
raise requests.HTTPError(f"{resp.status} {resp.reason}")
class _Response:
def __init__(self, raw, status):
self._data = raw
self.status_code = status
def json(self):
return _json.loads(self._data)
def raise_for_status(self):
pass
return _Response(data, resp.status)
except (requests.HTTPError, ValueError):
raise
except (ConnectionError, OSError, TimeoutError) as e:
if attempt == self.RETRIES - 1:
raise requests.ConnectionError(str(e))
time.sleep(self.RETRY_DELAY)
def get_state(self) -> LightState:
resp = requests.get(f"{self.base_url}/lights", timeout=5)
resp.raise_for_status()
resp = self._request("GET", "lights")
light = resp.json()["lights"][0]
return LightState(
on=bool(light["on"]),
@@ -48,12 +102,10 @@ class ElgatoLight:
"brightness": brightness if brightness is not None else current.brightness,
"temperature": temperature if temperature is not None else current.temperature,
}
resp = requests.put(
f"{self.base_url}/lights",
resp = self._request(
"PUT", "lights",
json={"numberOfLights": 1, "lights": [payload]},
timeout=5,
)
resp.raise_for_status()
light = resp.json()["lights"][0]
return LightState(
on=bool(light["on"]),
@@ -66,11 +118,7 @@ class ElgatoLight:
return self.set_state(on=not current.on)
def get_info(self) -> dict:
resp = requests.get(f"{self.base_url}/accessory-info", timeout=5)
resp.raise_for_status()
return resp.json()
return self._request("GET", "accessory-info").json()
def get_settings(self) -> dict:
resp = requests.get(f"{self.base_url}/lights/settings", timeout=5)
resp.raise_for_status()
return resp.json()
return self._request("GET", "lights/settings").json()
tests/test_api.py
+53 -34
View File
@@ -1,24 +1,30 @@
"""Tests for elgato_cli.api."""
import json
from unittest.mock import patch, MagicMock
import requests
from elgato_cli.api import ElgatoLight, LightState, mired_to_kelvin, kelvin_to_mired
def _mock_response(json_data, status_code=200):
resp = MagicMock()
resp.json.return_value = json_data
resp.status_code = status_code
resp.raise_for_status.return_value = None
return resp
LIGHT_RESPONSE = {
"numberOfLights": 1,
"lights": [{"on": 1, "brightness": 50, "temperature": 200}],
}
def _make_mock_conn(response_data, status=200):
"""Create a mock HTTPConnection that returns the given response."""
mock_conn = MagicMock()
mock_resp = MagicMock()
mock_resp.status = status
mock_resp.reason = "OK" if status == 200 else "Error"
mock_resp.read.return_value = json.dumps(response_data).encode()
mock_conn.getresponse.return_value = mock_resp
return mock_conn
class TestMiredKelvin:
def test_mired_to_kelvin(self):
assert mired_to_kelvin(143) == 6993
@@ -36,64 +42,77 @@ class TestMiredKelvin:
class TestElgatoLight:
@patch("elgato_cli.api.requests.get")
def test_get_state(self, mock_get):
mock_get.return_value = _mock_response(LIGHT_RESPONSE)
@patch("elgato_cli.api.http.client.HTTPConnection")
def test_get_state(self, mock_http_cls):
mock_http_cls.return_value = _make_mock_conn(LIGHT_RESPONSE)
light = ElgatoLight("192.168.1.100")
state = light.get_state()
assert state == LightState(on=True, brightness=50, temperature=200)
mock_get.assert_called_once_with(
"http://192.168.1.100:9123/elgato/lights", timeout=5
)
mock_http_cls.assert_called_with("192.168.1.100", 9123, timeout=5)
@patch("elgato_cli.api.requests.put")
@patch("elgato_cli.api.requests.get")
def test_set_state_partial(self, mock_get, mock_put):
mock_get.return_value = _mock_response(LIGHT_RESPONSE)
@patch("elgato_cli.api.http.client.HTTPConnection")
def test_set_state_partial(self, mock_http_cls):
updated = {
"numberOfLights": 1,
"lights": [{"on": 1, "brightness": 80, "temperature": 200}],
}
mock_put.return_value = _mock_response(updated)
# First call: get_state (GET), second call: put (PUT)
mock_http_cls.side_effect = [
_make_mock_conn(LIGHT_RESPONSE),
_make_mock_conn(updated),
]
light = ElgatoLight("192.168.1.100")
state = light.set_state(brightness=80)
assert state.brightness == 80
call_args = mock_put.call_args
payload = call_args[1]["json"]
assert payload["lights"][0]["brightness"] == 80
assert payload["lights"][0]["on"] == 1 # preserved from current state
assert mock_http_cls.call_count == 2
@patch("elgato_cli.api.requests.put")
@patch("elgato_cli.api.requests.get")
def test_toggle(self, mock_get, mock_put):
mock_get.return_value = _mock_response(LIGHT_RESPONSE) # on=1
@patch("elgato_cli.api.http.client.HTTPConnection")
def test_toggle(self, mock_http_cls):
toggled = {
"numberOfLights": 1,
"lights": [{"on": 0, "brightness": 50, "temperature": 200}],
}
mock_put.return_value = _mock_response(toggled)
# toggle calls: get_state, then set_state(get_state + put)
mock_http_cls.side_effect = [
_make_mock_conn(LIGHT_RESPONSE), # toggle -> get_state
_make_mock_conn(LIGHT_RESPONSE), # set_state -> get_state
_make_mock_conn(toggled), # set_state -> put
]
light = ElgatoLight("192.168.1.100")
state = light.toggle()
assert state.on is False
@patch("elgato_cli.api.requests.get")
def test_get_info(self, mock_get):
@patch("elgato_cli.api.http.client.HTTPConnection")
def test_get_info(self, mock_http_cls):
info_resp = {
"productName": "Elgato Key Light",
"serialNumber": "AB12CD34",
"firmwareVersion": "1.0.3",
}
mock_get.return_value = _mock_response(info_resp)
mock_http_cls.return_value = _make_mock_conn(info_resp)
light = ElgatoLight("192.168.1.100")
info = light.get_info()
assert info["productName"] == "Elgato Key Light"
mock_get.assert_called_once_with(
"http://192.168.1.100:9123/elgato/accessory-info", timeout=5
)
@patch("elgato_cli.api.time.sleep")
@patch("elgato_cli.api.http.client.HTTPConnection")
def test_retry_on_connection_error(self, mock_http_cls, mock_sleep):
fail_conn = MagicMock()
fail_conn.request.side_effect = ConnectionError("reset")
ok_conn = _make_mock_conn(LIGHT_RESPONSE)
mock_http_cls.side_effect = [fail_conn, ok_conn]
light = ElgatoLight("192.168.1.100")
state = light.get_state()
assert state == LightState(on=True, brightness=50, temperature=200)
assert mock_http_cls.call_count == 2
mock_sleep.assert_called_once_with(1.0)
def test_custom_port(self):
light = ElgatoLight("10.0.0.5", port=8080)