Initial implementation: CLI + KDE Plasma 6 widget for Elgato Key Light
- Python CLI (click): discover, status, on/off/toggle, brightness, temperature, info - mDNS discovery via zeroconf for automatic light detection - Shared config at ~/.config/elgato-cli/config.json - KDE Plasma 6 plasmoid with system tray icon, power/brightness/temperature controls - Plasmoid uses XMLHttpRequest directly to light API (no CLI dependency) - Unit tests for API client with mocked HTTP responses
This commit is contained in:
+10
@@ -0,0 +1,10 @@
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
*.egg-info/
|
||||
dist/
|
||||
build/
|
||||
*.egg
|
||||
.venv/
|
||||
venv/
|
||||
.pytest_cache/
|
||||
Executable
+12
@@ -0,0 +1,12 @@
|
||||
#!/bin/bash
|
||||
set -e
|
||||
|
||||
PLASMOID_DIR="$(dirname "$0")/plasmoid/at.sub-net.elgato"
|
||||
|
||||
# Remove old version if installed
|
||||
kpackagetool6 -t Plasma/Applet -r at.sub-net.elgato 2>/dev/null || true
|
||||
|
||||
# Install
|
||||
kpackagetool6 -t Plasma/Applet -i "$PLASMOID_DIR"
|
||||
|
||||
echo "Plasmoid installed. Add 'Elgato Key Light' to your panel or system tray."
|
||||
@@ -0,0 +1,21 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<kcfg xmlns="http://www.kde.org/standards/kcfg/1.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://www.kde.org/standards/kcfg/1.0
|
||||
http://www.kde.org/standards/kcfg/1.0/kcfg.xsd">
|
||||
<kcfgfile name=""/>
|
||||
<group name="General">
|
||||
<entry name="lightHost" type="String">
|
||||
<label>Elgato light IP address or hostname</label>
|
||||
<default></default>
|
||||
</entry>
|
||||
<entry name="lightPort" type="Int">
|
||||
<label>Elgato light port</label>
|
||||
<default>9123</default>
|
||||
</entry>
|
||||
<entry name="pollInterval" type="Int">
|
||||
<label>Polling interval in milliseconds</label>
|
||||
<default>3000</default>
|
||||
</entry>
|
||||
</group>
|
||||
</kcfg>
|
||||
@@ -0,0 +1,37 @@
|
||||
import QtQuick
|
||||
import org.kde.plasma.core as PlasmaCore
|
||||
import org.kde.kirigami as Kirigami
|
||||
|
||||
MouseArea {
|
||||
id: compactRoot
|
||||
|
||||
property bool wasExpanded: false
|
||||
|
||||
acceptedButtons: Qt.LeftButton | Qt.MiddleButton
|
||||
|
||||
Kirigami.Icon {
|
||||
anchors.fill: parent
|
||||
source: {
|
||||
if (root.lightHost === "" || !root.lightReachable)
|
||||
return "network-disconnect"
|
||||
return root.lightOn ? "brightness-high" : "brightness-low"
|
||||
}
|
||||
active: compactRoot.containsMouse
|
||||
}
|
||||
|
||||
hoverEnabled: true
|
||||
|
||||
onPressed: function(mouse) {
|
||||
if (mouse.button === Qt.MiddleButton) {
|
||||
root.toggleLight()
|
||||
} else {
|
||||
wasExpanded = root.expanded
|
||||
}
|
||||
}
|
||||
|
||||
onClicked: function(mouse) {
|
||||
if (mouse.button === Qt.LeftButton) {
|
||||
root.expanded = !wasExpanded
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
import QtQuick
|
||||
import QtQuick.Controls as QQC2
|
||||
import QtQuick.Layouts
|
||||
import org.kde.kirigami as Kirigami
|
||||
import org.kde.plasma.components as PC3
|
||||
|
||||
ColumnLayout {
|
||||
id: fullRoot
|
||||
|
||||
Layout.preferredWidth: Kirigami.Units.gridUnit * 18
|
||||
Layout.preferredHeight: Kirigami.Units.gridUnit * 10
|
||||
Layout.minimumWidth: Kirigami.Units.gridUnit * 14
|
||||
|
||||
spacing: Kirigami.Units.smallSpacing
|
||||
|
||||
// Not configured state
|
||||
PC3.Label {
|
||||
Layout.alignment: Qt.AlignCenter
|
||||
text: "Run 'elgato discover' or configure host in widget settings"
|
||||
visible: root.lightHost === ""
|
||||
wrapMode: Text.WordWrap
|
||||
Layout.fillWidth: true
|
||||
horizontalAlignment: Text.AlignHCenter
|
||||
}
|
||||
|
||||
// Unreachable state
|
||||
PC3.Label {
|
||||
Layout.alignment: Qt.AlignCenter
|
||||
text: "Light unreachable (" + root.lightHost + ")"
|
||||
visible: root.lightHost !== "" && !root.lightReachable
|
||||
wrapMode: Text.WordWrap
|
||||
Layout.fillWidth: true
|
||||
horizontalAlignment: Text.AlignHCenter
|
||||
}
|
||||
|
||||
// Controls (only when reachable)
|
||||
ColumnLayout {
|
||||
visible: root.lightHost !== "" && root.lightReachable
|
||||
Layout.fillWidth: true
|
||||
spacing: Kirigami.Units.smallSpacing
|
||||
Layout.margins: Kirigami.Units.smallSpacing
|
||||
|
||||
// Header: Power switch
|
||||
RowLayout {
|
||||
Layout.fillWidth: true
|
||||
|
||||
Kirigami.Icon {
|
||||
source: "video-television"
|
||||
Layout.preferredWidth: Kirigami.Units.iconSizes.medium
|
||||
Layout.preferredHeight: Kirigami.Units.iconSizes.medium
|
||||
}
|
||||
|
||||
PC3.Label {
|
||||
text: "Elgato Key Light"
|
||||
font.bold: true
|
||||
Layout.fillWidth: true
|
||||
}
|
||||
|
||||
QQC2.Switch {
|
||||
checked: root.lightOn
|
||||
onToggled: root.toggleLight()
|
||||
}
|
||||
}
|
||||
|
||||
Kirigami.Separator {
|
||||
Layout.fillWidth: true
|
||||
}
|
||||
|
||||
// Brightness slider
|
||||
RowLayout {
|
||||
Layout.fillWidth: true
|
||||
|
||||
Kirigami.Icon {
|
||||
source: "brightness-high"
|
||||
Layout.preferredWidth: Kirigami.Units.iconSizes.small
|
||||
Layout.preferredHeight: Kirigami.Units.iconSizes.small
|
||||
}
|
||||
|
||||
PC3.Slider {
|
||||
id: brightnessSlider
|
||||
Layout.fillWidth: true
|
||||
from: 3
|
||||
to: 100
|
||||
stepSize: 1
|
||||
value: root.lightBrightness
|
||||
onMoved: root.setBrightness(Math.round(value))
|
||||
}
|
||||
|
||||
PC3.Label {
|
||||
text: Math.round(brightnessSlider.value) + "%"
|
||||
Layout.preferredWidth: Kirigami.Units.gridUnit * 3
|
||||
horizontalAlignment: Text.AlignRight
|
||||
}
|
||||
}
|
||||
|
||||
// Temperature slider
|
||||
RowLayout {
|
||||
Layout.fillWidth: true
|
||||
|
||||
Kirigami.Icon {
|
||||
source: "color-management"
|
||||
Layout.preferredWidth: Kirigami.Units.iconSizes.small
|
||||
Layout.preferredHeight: Kirigami.Units.iconSizes.small
|
||||
}
|
||||
|
||||
PC3.Slider {
|
||||
id: temperatureSlider
|
||||
Layout.fillWidth: true
|
||||
from: 143
|
||||
to: 344
|
||||
stepSize: 1
|
||||
value: root.lightTemperature
|
||||
onMoved: root.setTemperature(Math.round(value))
|
||||
}
|
||||
|
||||
PC3.Label {
|
||||
text: Math.round(1000000 / temperatureSlider.value) + "K"
|
||||
Layout.preferredWidth: Kirigami.Units.gridUnit * 3
|
||||
horizontalAlignment: Text.AlignRight
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,150 @@
|
||||
import QtQuick
|
||||
import org.kde.plasma.plasmoid
|
||||
import org.kde.plasma.core as PlasmaCore
|
||||
|
||||
PlasmoidItem {
|
||||
id: root
|
||||
|
||||
// State
|
||||
property bool lightOn: false
|
||||
property int lightBrightness: 50
|
||||
property int lightTemperature: 200 // mired
|
||||
property bool lightReachable: false
|
||||
property string lightHost: ""
|
||||
property int lightPort: 9123
|
||||
|
||||
// Resolve host: widget config > shared CLI config
|
||||
function resolveHost() {
|
||||
var cfgHost = Plasmoid.configuration.lightHost || ""
|
||||
var cfgPort = Plasmoid.configuration.lightPort || 9123
|
||||
|
||||
if (cfgHost !== "") {
|
||||
root.lightHost = cfgHost
|
||||
root.lightPort = cfgPort
|
||||
return
|
||||
}
|
||||
|
||||
// Try reading shared CLI config
|
||||
var xhr = new XMLHttpRequest()
|
||||
var configPath = StandardPaths.writableLocation(StandardPaths.GenericConfigLocation) + "/../elgato-cli/config.json"
|
||||
// Use home dir approach
|
||||
try {
|
||||
xhr.open("GET", "file://" + StandardPaths.writableLocation(StandardPaths.HomeLocation) + "/.config/elgato-cli/config.json", false)
|
||||
xhr.send()
|
||||
if (xhr.status === 200 || xhr.readyState === XMLHttpRequest.DONE) {
|
||||
var config = JSON.parse(xhr.responseText)
|
||||
if (config.host) {
|
||||
root.lightHost = config.host
|
||||
root.lightPort = config.port || 9123
|
||||
}
|
||||
}
|
||||
} catch (e) {
|
||||
console.log("Elgato: Could not read shared config:", e)
|
||||
}
|
||||
}
|
||||
|
||||
function baseUrl() {
|
||||
return "http://" + root.lightHost + ":" + root.lightPort + "/elgato"
|
||||
}
|
||||
|
||||
function fetchState() {
|
||||
if (root.lightHost === "") return
|
||||
|
||||
var xhr = new XMLHttpRequest()
|
||||
xhr.open("GET", baseUrl() + "/lights")
|
||||
xhr.timeout = 3000
|
||||
xhr.onreadystatechange = function() {
|
||||
if (xhr.readyState !== XMLHttpRequest.DONE) return
|
||||
if (xhr.status === 200) {
|
||||
var data = JSON.parse(xhr.responseText)
|
||||
var light = data.lights[0]
|
||||
root.lightOn = light.on === 1
|
||||
root.lightBrightness = light.brightness
|
||||
root.lightTemperature = light.temperature
|
||||
root.lightReachable = true
|
||||
} else {
|
||||
root.lightReachable = false
|
||||
}
|
||||
}
|
||||
xhr.onerror = function() { root.lightReachable = false }
|
||||
xhr.send()
|
||||
}
|
||||
|
||||
function setLightState(payload) {
|
||||
if (root.lightHost === "") return
|
||||
|
||||
var xhr = new XMLHttpRequest()
|
||||
xhr.open("PUT", baseUrl() + "/lights")
|
||||
xhr.setRequestHeader("Content-Type", "application/json")
|
||||
xhr.onreadystatechange = function() {
|
||||
if (xhr.readyState !== XMLHttpRequest.DONE) return
|
||||
if (xhr.status === 200) {
|
||||
var data = JSON.parse(xhr.responseText)
|
||||
var light = data.lights[0]
|
||||
root.lightOn = light.on === 1
|
||||
root.lightBrightness = light.brightness
|
||||
root.lightTemperature = light.temperature
|
||||
root.lightReachable = true
|
||||
} else {
|
||||
root.lightReachable = false
|
||||
}
|
||||
}
|
||||
xhr.onerror = function() { root.lightReachable = false }
|
||||
xhr.send(JSON.stringify({numberOfLights: 1, lights: [payload]}))
|
||||
}
|
||||
|
||||
function toggleLight() {
|
||||
setLightState({
|
||||
on: root.lightOn ? 0 : 1,
|
||||
brightness: root.lightBrightness,
|
||||
temperature: root.lightTemperature
|
||||
})
|
||||
}
|
||||
|
||||
function setBrightness(val) {
|
||||
setLightState({
|
||||
on: root.lightOn ? 1 : 0,
|
||||
brightness: val,
|
||||
temperature: root.lightTemperature
|
||||
})
|
||||
}
|
||||
|
||||
function setTemperature(val) {
|
||||
setLightState({
|
||||
on: root.lightOn ? 1 : 0,
|
||||
brightness: root.lightBrightness,
|
||||
temperature: val
|
||||
})
|
||||
}
|
||||
|
||||
// Polling timer
|
||||
Timer {
|
||||
id: pollTimer
|
||||
interval: Plasmoid.configuration.pollInterval || 3000
|
||||
running: true
|
||||
repeat: true
|
||||
triggeredOnStart: true
|
||||
onTriggered: root.fetchState()
|
||||
}
|
||||
|
||||
Component.onCompleted: {
|
||||
resolveHost()
|
||||
fetchState()
|
||||
}
|
||||
|
||||
// Tray icon
|
||||
compactRepresentation: CompactRepresentation {}
|
||||
|
||||
// Popup
|
||||
fullRepresentation: FullRepresentation {}
|
||||
|
||||
// Prefer status area
|
||||
Plasmoid.status: root.lightReachable ? PlasmaCore.Types.ActiveStatus : PlasmaCore.Types.PassiveStatus
|
||||
preferredRepresentation: compactRepresentation
|
||||
toolTipMainText: "Elgato Key Light"
|
||||
toolTipSubText: {
|
||||
if (root.lightHost === "") return "Not configured"
|
||||
if (!root.lightReachable) return "Unreachable"
|
||||
return (root.lightOn ? "On" : "Off") + " - " + root.lightBrightness + "% - " + Math.round(1000000 / root.lightTemperature) + "K"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
{
|
||||
"KPlugin": {
|
||||
"Id": "at.sub-net.elgato",
|
||||
"Name": "Elgato Key Light",
|
||||
"Description": "Control an Elgato Key Light from the system tray",
|
||||
"Category": "Hardware",
|
||||
"Icon": "video-television",
|
||||
"Authors": [
|
||||
{
|
||||
"Name": "Tina Schellander",
|
||||
"Email": "tina@sub-net.at"
|
||||
}
|
||||
],
|
||||
"Version": "0.1.0",
|
||||
"License": "MIT"
|
||||
},
|
||||
"X-Plasma-API": "declarativeappletscript",
|
||||
"X-Plasma-API-Minimum-Version": "6.0",
|
||||
"X-Plasma-Provides": [
|
||||
"org.kde.plasma.systemtray"
|
||||
],
|
||||
"KPackageStructure": "Plasma/Applet"
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=68.0"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "elgato-cli"
|
||||
version = "0.1.0"
|
||||
description = "CLI tool to control Elgato Key Light"
|
||||
requires-python = ">=3.11"
|
||||
dependencies = [
|
||||
"click>=8.0",
|
||||
"requests>=2.28",
|
||||
"zeroconf>=0.80",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = ["pytest>=7.0"]
|
||||
|
||||
[project.scripts]
|
||||
elgato = "elgato_cli.cli:cli"
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["src"]
|
||||
@@ -0,0 +1,76 @@
|
||||
"""HTTP client for the Elgato Key Light REST API."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
@dataclass
|
||||
class LightState:
|
||||
on: bool
|
||||
brightness: int # 3-100
|
||||
temperature: int # 143-344 mired
|
||||
|
||||
|
||||
def mired_to_kelvin(mired: int) -> int:
|
||||
return round(1_000_000 / mired)
|
||||
|
||||
|
||||
def kelvin_to_mired(kelvin: int) -> int:
|
||||
return round(1_000_000 / kelvin)
|
||||
|
||||
|
||||
class ElgatoLight:
|
||||
def __init__(self, host: str, port: int = 9123):
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.base_url = f"http://{host}:{port}/elgato"
|
||||
|
||||
def get_state(self) -> LightState:
|
||||
resp = requests.get(f"{self.base_url}/lights", timeout=5)
|
||||
resp.raise_for_status()
|
||||
light = resp.json()["lights"][0]
|
||||
return LightState(
|
||||
on=bool(light["on"]),
|
||||
brightness=light["brightness"],
|
||||
temperature=light["temperature"],
|
||||
)
|
||||
|
||||
def set_state(
|
||||
self,
|
||||
on: bool | None = None,
|
||||
brightness: int | None = None,
|
||||
temperature: int | None = None,
|
||||
) -> LightState:
|
||||
current = self.get_state()
|
||||
payload = {
|
||||
"on": int(on) if on is not None else int(current.on),
|
||||
"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",
|
||||
json={"numberOfLights": 1, "lights": [payload]},
|
||||
timeout=5,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
light = resp.json()["lights"][0]
|
||||
return LightState(
|
||||
on=bool(light["on"]),
|
||||
brightness=light["brightness"],
|
||||
temperature=light["temperature"],
|
||||
)
|
||||
|
||||
def toggle(self) -> LightState:
|
||||
current = self.get_state()
|
||||
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()
|
||||
|
||||
def get_settings(self) -> dict:
|
||||
resp = requests.get(f"{self.base_url}/lights/settings", timeout=5)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
@@ -0,0 +1,152 @@
|
||||
"""Click CLI entry point for elgato-cli."""
|
||||
|
||||
import click
|
||||
import requests
|
||||
|
||||
from . import config
|
||||
from .api import ElgatoLight, mired_to_kelvin, kelvin_to_mired
|
||||
|
||||
|
||||
def get_light(host: str | None, port: int | None) -> ElgatoLight:
|
||||
"""Resolve light host/port from flags or config."""
|
||||
cfg = config.load()
|
||||
h = host or cfg.get("host")
|
||||
p = port or cfg.get("port", 9123)
|
||||
if not h:
|
||||
raise click.ClickException(
|
||||
"No light configured. Run 'elgato discover' first or use --host."
|
||||
)
|
||||
return ElgatoLight(h, p)
|
||||
|
||||
|
||||
@click.group()
|
||||
@click.option("--host", default=None, help="Light IP address (overrides config).")
|
||||
@click.option("--port", default=None, type=int, help="Light port (overrides config).")
|
||||
@click.pass_context
|
||||
def cli(ctx, host, port):
|
||||
"""Control Elgato Key Light from the command line."""
|
||||
ctx.ensure_object(dict)
|
||||
ctx.obj["host"] = host
|
||||
ctx.obj["port"] = port
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.option("--timeout", default=5.0, help="Discovery timeout in seconds.")
|
||||
def discover(timeout):
|
||||
"""Discover Elgato lights on the network and save to config."""
|
||||
from .discovery import discover as do_discover
|
||||
|
||||
click.echo(f"Searching for Elgato lights ({timeout}s)...")
|
||||
lights = do_discover(timeout=timeout)
|
||||
|
||||
if not lights:
|
||||
raise click.ClickException("No Elgato lights found.")
|
||||
|
||||
for i, light in enumerate(lights):
|
||||
click.echo(f" [{i + 1}] {light['name']} ({light['host']}:{light['port']})")
|
||||
|
||||
chosen = lights[0]
|
||||
if len(lights) > 1:
|
||||
idx = click.prompt("Select light", type=int, default=1)
|
||||
chosen = lights[idx - 1]
|
||||
|
||||
config.save({
|
||||
"host": chosen["host"],
|
||||
"port": chosen["port"],
|
||||
"name": chosen["name"],
|
||||
})
|
||||
click.echo(f"Saved: {chosen['name']} ({chosen['host']}:{chosen['port']})")
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.pass_context
|
||||
def status(ctx):
|
||||
"""Show current light state."""
|
||||
light = get_light(ctx.obj["host"], ctx.obj["port"])
|
||||
try:
|
||||
state = light.get_state()
|
||||
except requests.ConnectionError:
|
||||
raise click.ClickException("Light unreachable.")
|
||||
click.echo(f"Power: {'on' if state.on else 'off'}")
|
||||
click.echo(f"Brightness: {state.brightness}%")
|
||||
click.echo(f"Temperature: {state.temperature} mired ({mired_to_kelvin(state.temperature)}K)")
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.pass_context
|
||||
def on(ctx):
|
||||
"""Turn the light on."""
|
||||
light = get_light(ctx.obj["host"], ctx.obj["port"])
|
||||
light.set_state(on=True)
|
||||
click.echo("Light on.")
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.pass_context
|
||||
def off(ctx):
|
||||
"""Turn the light off."""
|
||||
light = get_light(ctx.obj["host"], ctx.obj["port"])
|
||||
light.set_state(on=False)
|
||||
click.echo("Light off.")
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.pass_context
|
||||
def toggle(ctx):
|
||||
"""Toggle the light on/off."""
|
||||
light = get_light(ctx.obj["host"], ctx.obj["port"])
|
||||
state = light.toggle()
|
||||
click.echo(f"Light {'on' if state.on else 'off'}.")
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.argument("value", required=False, type=int)
|
||||
@click.pass_context
|
||||
def brightness(ctx, value):
|
||||
"""Get or set brightness (3-100)."""
|
||||
light = get_light(ctx.obj["host"], ctx.obj["port"])
|
||||
if value is None:
|
||||
state = light.get_state()
|
||||
click.echo(f"{state.brightness}%")
|
||||
else:
|
||||
if not 3 <= value <= 100:
|
||||
raise click.ClickException("Brightness must be 3-100.")
|
||||
state = light.set_state(brightness=value)
|
||||
click.echo(f"Brightness set to {state.brightness}%.")
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.argument("value", required=False, type=int)
|
||||
@click.option("-k", "--kelvin", is_flag=True, help="Interpret value as Kelvin.")
|
||||
@click.pass_context
|
||||
def temperature(ctx, value, kelvin):
|
||||
"""Get or set color temperature (143-344 mired, or use -k for Kelvin)."""
|
||||
light = get_light(ctx.obj["host"], ctx.obj["port"])
|
||||
if value is None:
|
||||
state = light.get_state()
|
||||
click.echo(f"{state.temperature} mired ({mired_to_kelvin(state.temperature)}K)")
|
||||
else:
|
||||
mired = kelvin_to_mired(value) if kelvin else value
|
||||
if not 143 <= mired <= 344:
|
||||
raise click.ClickException(
|
||||
f"Temperature must be 143-344 mired (2900-7000K). Got {mired} mired."
|
||||
)
|
||||
state = light.set_state(temperature=mired)
|
||||
click.echo(
|
||||
f"Temperature set to {state.temperature} mired ({mired_to_kelvin(state.temperature)}K)."
|
||||
)
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.pass_context
|
||||
def info(ctx):
|
||||
"""Show light hardware info."""
|
||||
light = get_light(ctx.obj["host"], ctx.obj["port"])
|
||||
try:
|
||||
data = light.get_info()
|
||||
except requests.ConnectionError:
|
||||
raise click.ClickException("Light unreachable.")
|
||||
click.echo(f"Product: {data.get('productName', 'N/A')}")
|
||||
click.echo(f"Serial: {data.get('serialNumber', 'N/A')}")
|
||||
click.echo(f"Firmware: {data.get('firmwareVersion', 'N/A')}")
|
||||
click.echo(f"Display: {data.get('displayName', 'N/A')}")
|
||||
@@ -0,0 +1,18 @@
|
||||
"""Shared configuration for elgato-cli (~/.config/elgato-cli/config.json)."""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
CONFIG_DIR = Path.home() / ".config" / "elgato-cli"
|
||||
CONFIG_FILE = CONFIG_DIR / "config.json"
|
||||
|
||||
|
||||
def load() -> dict:
|
||||
if CONFIG_FILE.exists():
|
||||
return json.loads(CONFIG_FILE.read_text())
|
||||
return {}
|
||||
|
||||
|
||||
def save(config: dict) -> None:
|
||||
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
||||
CONFIG_FILE.write_text(json.dumps(config, indent=2) + "\n")
|
||||
@@ -0,0 +1,42 @@
|
||||
"""mDNS discovery for Elgato Key Light devices."""
|
||||
|
||||
import socket
|
||||
import time
|
||||
|
||||
from zeroconf import ServiceBrowser, ServiceStateChange, Zeroconf
|
||||
|
||||
SERVICE_TYPE = "_elg._tcp.local."
|
||||
|
||||
|
||||
def discover(timeout: float = 5.0) -> list[dict]:
|
||||
"""Discover Elgato lights on the network. Returns list of {name, host, port}."""
|
||||
lights = []
|
||||
|
||||
def on_state_change(zeroconf, service_type, name, state_change):
|
||||
if state_change is not ServiceStateChange.Added:
|
||||
return
|
||||
info = zeroconf.get_service_info(service_type, name)
|
||||
if info is None:
|
||||
return
|
||||
addresses = info.parsed_addresses()
|
||||
if not addresses:
|
||||
return
|
||||
# Prefer IPv4
|
||||
host = next(
|
||||
(a for a in addresses if ":" not in a),
|
||||
addresses[0],
|
||||
)
|
||||
lights.append({
|
||||
"name": info.server.rstrip(".") if info.server else name,
|
||||
"host": host,
|
||||
"port": info.port,
|
||||
})
|
||||
|
||||
zc = Zeroconf()
|
||||
try:
|
||||
browser = ServiceBrowser(zc, SERVICE_TYPE, handlers=[on_state_change])
|
||||
time.sleep(timeout)
|
||||
finally:
|
||||
zc.close()
|
||||
|
||||
return lights
|
||||
@@ -0,0 +1,100 @@
|
||||
"""Tests for elgato_cli.api."""
|
||||
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
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}],
|
||||
}
|
||||
|
||||
|
||||
class TestMiredKelvin:
|
||||
def test_mired_to_kelvin(self):
|
||||
assert mired_to_kelvin(143) == 6993
|
||||
assert mired_to_kelvin(344) == 2907
|
||||
assert mired_to_kelvin(200) == 5000
|
||||
|
||||
def test_kelvin_to_mired(self):
|
||||
assert kelvin_to_mired(5000) == 200
|
||||
assert kelvin_to_mired(7000) == 143
|
||||
assert kelvin_to_mired(2900) == 345
|
||||
|
||||
def test_roundtrip(self):
|
||||
for mired in [143, 200, 250, 300, 344]:
|
||||
assert abs(kelvin_to_mired(mired_to_kelvin(mired)) - mired) <= 1
|
||||
|
||||
|
||||
class TestElgatoLight:
|
||||
@patch("elgato_cli.api.requests.get")
|
||||
def test_get_state(self, mock_get):
|
||||
mock_get.return_value = _mock_response(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
|
||||
)
|
||||
|
||||
@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)
|
||||
updated = {
|
||||
"numberOfLights": 1,
|
||||
"lights": [{"on": 1, "brightness": 80, "temperature": 200}],
|
||||
}
|
||||
mock_put.return_value = _mock_response(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
|
||||
|
||||
@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
|
||||
toggled = {
|
||||
"numberOfLights": 1,
|
||||
"lights": [{"on": 0, "brightness": 50, "temperature": 200}],
|
||||
}
|
||||
mock_put.return_value = _mock_response(toggled)
|
||||
|
||||
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):
|
||||
info_resp = {
|
||||
"productName": "Elgato Key Light",
|
||||
"serialNumber": "AB12CD34",
|
||||
"firmwareVersion": "1.0.3",
|
||||
}
|
||||
mock_get.return_value = _mock_response(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
|
||||
)
|
||||
|
||||
def test_custom_port(self):
|
||||
light = ElgatoLight("10.0.0.5", port=8080)
|
||||
assert light.base_url == "http://10.0.0.5:8080/elgato"
|
||||
Reference in New Issue
Block a user