Add Kobra X integration with initial components and configuration

This commit is contained in:
Gangoke
2026-05-16 22:26:45 -10:00
parent 1982605cef
commit 9f4a94da24
17 changed files with 738 additions and 0 deletions

3
.gitignore vendored Normal file
View File

@@ -0,0 +1,3 @@
__pycache__/
*.pyc
.pytest_cache/

View File

@@ -0,0 +1,47 @@
from __future__ import annotations
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .api import KobraXApiClient
from .const import CONF_HOST, DOMAIN, PLATFORMS
from .coordinator import KobraXCoordinator
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass: HomeAssistant, config: dict) -> bool:
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass.data.setdefault(DOMAIN, {})
host = entry.data[CONF_HOST]
base_url = host if host.startswith("http") else f"http://{host}"
session = async_get_clientsession(hass)
api = KobraXApiClient(session, base_url)
hass.data[DOMAIN]["logger"] = _LOGGER
coordinator = KobraXCoordinator(hass, api)
await coordinator.async_config_entry_first_refresh()
hass.data[DOMAIN][entry.entry_id] = {
"coordinator": coordinator,
"api": api,
"entry": entry,
}
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

View File

@@ -0,0 +1,78 @@
from __future__ import annotations
from typing import Any
import aiohttp
class KobraXApiError(Exception):
"""Raised when communication with KX-Bridge fails."""
class KobraXApiClient:
def __init__(self, session: aiohttp.ClientSession, base_url: str) -> None:
self._session = session
self._base_url = base_url.rstrip("/")
def _url(self, path: str) -> str:
return f"{self._base_url}{path}"
async def _get_json(self, path: str) -> dict[str, Any]:
try:
async with self._session.get(self._url(path)) as response:
response.raise_for_status()
return await response.json()
except Exception as err:
raise KobraXApiError(err) from err
async def _post_json(self, path: str, body: dict[str, Any]) -> dict[str, Any]:
try:
async with self._session.post(self._url(path), json=body) as response:
response.raise_for_status()
return await response.json()
except Exception as err:
raise KobraXApiError(err) from err
async def async_check_version(self) -> dict[str, Any]:
return await self._get_json("/api/version")
async def async_get_state(self) -> dict[str, Any]:
return await self._get_json("/api/state")
async def async_pause_print(self) -> None:
await self._post_json("/printer/print/pause", {})
async def async_resume_print(self) -> None:
await self._post_json("/printer/print/resume", {})
async def async_cancel_print(self) -> None:
await self._post_json("/printer/print/cancel", {})
async def async_set_light(self, is_on: bool, brightness: int = 80) -> None:
await self._post_json("/api/light", {"on": is_on, "brightness": brightness})
async def async_set_speed_mode(self, mode: int) -> None:
await self._post_json("/api/speed", {"mode": mode})
async def async_set_temperature(self, nozzle: float | None, bed: float | None) -> None:
payload: dict[str, Any] = {}
if nozzle is not None:
payload["nozzle"] = nozzle
if bed is not None:
payload["bed"] = bed
if payload:
await self._post_json("/api/temperature", payload)
async def async_connect(self) -> None:
await self._post_json("/api/connect", {})
async def async_disconnect(self) -> None:
await self._post_json("/api/disconnect", {})
async def async_get_camera_snapshot(self) -> bytes:
try:
async with self._session.get(self._url("/api/camera/snapshot")) as response:
response.raise_for_status()
return await response.read()
except Exception as err:
raise KobraXApiError(err) from err

View File

@@ -0,0 +1,62 @@
from __future__ import annotations
from dataclasses import dataclass
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from .const import DOMAIN
from .entity import KobraXEntity
@dataclass(frozen=True, kw_only=True)
class KobraXBinaryDescription(BinarySensorEntityDescription):
value_key: str
BINARY_SENSORS: tuple[KobraXBinaryDescription, ...] = (
KobraXBinaryDescription(
key="online",
name="Online",
value_key="kobra_state",
device_class=BinarySensorDeviceClass.CONNECTIVITY,
),
KobraXBinaryDescription(
key="printing",
name="Printing",
value_key="print_state",
device_class=BinarySensorDeviceClass.RUNNING,
),
KobraXBinaryDescription(
key="light",
name="Light State",
value_key="light_on",
icon="mdi:lightbulb",
entity_registry_enabled_default=False,
),
)
class KobraXBinarySensor(KobraXEntity, BinarySensorEntity):
entity_description: KobraXBinaryDescription
@property
def is_on(self) -> bool:
if self.entity_description.value_key == "kobra_state":
return str(self.state_data.get("kobra_state", "")).lower() != "offline"
if self.entity_description.value_key == "print_state":
return str(self.state_data.get("print_state", "")).lower() == "printing"
return bool(self.state_data.get(self.entity_description.value_key, False))
async def async_setup_entry(hass, entry, async_add_entities):
coordinator = hass.data[DOMAIN][entry.entry_id]["coordinator"]
async_add_entities(
[
KobraXBinarySensor(coordinator, entry, description.key, description.name)
for description in BINARY_SENSORS
]
)

View File

@@ -0,0 +1,82 @@
from __future__ import annotations
from dataclasses import dataclass
from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
from homeassistant.exceptions import ServiceValidationError
from .api import KobraXApiError
from .const import DOMAIN
from .entity import KobraXEntity
@dataclass(frozen=True, kw_only=True)
class KobraXButtonDescription(ButtonEntityDescription):
action: str
BUTTONS: tuple[KobraXButtonDescription, ...] = (
KobraXButtonDescription(
key="pause_print",
name="Pause Print",
icon="mdi:pause",
action="pause",
),
KobraXButtonDescription(
key="resume_print",
name="Resume Print",
icon="mdi:play",
action="resume",
),
KobraXButtonDescription(
key="cancel_print",
name="Cancel Print",
icon="mdi:stop",
action="cancel",
),
KobraXButtonDescription(
key="connect",
name="Connect Bridge",
icon="mdi:lan-connect",
action="connect",
entity_registry_enabled_default=False,
),
KobraXButtonDescription(
key="disconnect",
name="Disconnect Bridge",
icon="mdi:lan-disconnect",
action="disconnect",
entity_registry_enabled_default=False,
),
)
class KobraXActionButton(KobraXEntity, ButtonEntity):
entity_description: KobraXButtonDescription
async def async_press(self) -> None:
api = self.hass.data[DOMAIN][self._entry.entry_id]["api"]
try:
if self.entity_description.action == "pause":
await api.async_pause_print()
elif self.entity_description.action == "resume":
await api.async_resume_print()
elif self.entity_description.action == "cancel":
await api.async_cancel_print()
elif self.entity_description.action == "connect":
await api.async_connect()
elif self.entity_description.action == "disconnect":
await api.async_disconnect()
await self.coordinator.async_request_refresh()
except KobraXApiError as err:
raise ServiceValidationError(str(err)) from err
async def async_setup_entry(hass, entry, async_add_entities):
coordinator = hass.data[DOMAIN][entry.entry_id]["coordinator"]
async_add_entities(
[
KobraXActionButton(coordinator, entry, description.key, description.name)
for description in BUTTONS
]
)

View File

@@ -0,0 +1,29 @@
from __future__ import annotations
from homeassistant.components.camera import Camera
from .api import KobraXApiError
from .const import DOMAIN
from .entity import KobraXEntity
class KobraXCamera(KobraXEntity, Camera):
def __init__(self, coordinator, entry, key: str, name: str) -> None:
KobraXEntity.__init__(self, coordinator, entry, key, name)
Camera.__init__(self)
@property
def is_streaming(self) -> bool:
return bool(self.state_data.get("camera_url"))
async def async_camera_image(self, width: int | None = None, height: int | None = None) -> bytes | None:
api = self.hass.data[DOMAIN][self._entry.entry_id]["api"]
try:
return await api.async_get_camera_snapshot()
except KobraXApiError:
return None
async def async_setup_entry(hass, entry, async_add_entities):
coordinator = hass.data[DOMAIN][entry.entry_id]["coordinator"]
async_add_entities([KobraXCamera(coordinator, entry, "camera", "Camera")])

View File

@@ -0,0 +1,102 @@
from __future__ import annotations
import re
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.core import callback
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .api import KobraXApiClient, KobraXApiError
from .const import CONF_HOST, CONF_PRINTER_NAME, DEFAULT_HOST, DEFAULT_PRINTER_NAME, DOMAIN
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST, default=DEFAULT_HOST): str,
vol.Required(CONF_PRINTER_NAME, default=DEFAULT_PRINTER_NAME): str,
}
)
def _normalize_host(host: str) -> str:
cleaned = host.strip()
if not re.match(r"https?://", cleaned):
cleaned = f"http://{cleaned}"
cleaned = re.sub(r"/+$", "", cleaned)
return cleaned
class KobraXConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
async def async_step_user(self, user_input=None):
errors: dict[str, str] = {}
if user_input is not None:
host = _normalize_host(user_input[CONF_HOST])
printer_name = user_input[CONF_PRINTER_NAME].strip() or DEFAULT_PRINTER_NAME
await self.async_set_unique_id(host)
self._abort_if_unique_id_configured()
session = async_get_clientsession(self.hass)
api = KobraXApiClient(session, host)
try:
await api.async_check_version()
except KobraXApiError:
errors["base"] = "cannot_connect"
if not errors:
return self.async_create_entry(
title=printer_name,
data={
CONF_HOST: host,
CONF_PRINTER_NAME: printer_name,
},
)
return self.async_show_form(
step_id="user",
data_schema=STEP_USER_DATA_SCHEMA,
errors=errors,
)
@staticmethod
@callback
def async_get_options_flow(config_entry):
return KobraXOptionsFlow(config_entry)
class KobraXOptionsFlow(config_entries.OptionsFlow):
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
self.config_entry = config_entry
async def async_step_init(self, user_input=None):
if user_input is not None:
host = _normalize_host(user_input[CONF_HOST])
printer_name = user_input[CONF_PRINTER_NAME].strip() or DEFAULT_PRINTER_NAME
return self.async_create_entry(
title="",
data={
CONF_HOST: host,
CONF_PRINTER_NAME: printer_name,
},
)
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Required(
CONF_HOST,
default=self.config_entry.data.get(CONF_HOST, DEFAULT_HOST),
): str,
vol.Required(
CONF_PRINTER_NAME,
default=self.config_entry.data.get(
CONF_PRINTER_NAME, DEFAULT_PRINTER_NAME
),
): str,
}
),
)

View File

@@ -0,0 +1,22 @@
from datetime import timedelta
NAME = "Kobra X"
DOMAIN = "kobrax"
MANUFACTURER = "Anycubic"
CONF_HOST = "host"
CONF_PRINTER_NAME = "printer_name"
DEFAULT_HOST = "localhost:7125"
DEFAULT_PRINTER_NAME = "Anycubic Kobra X"
UPDATE_INTERVAL = timedelta(seconds=5)
PLATFORMS = [
"sensor",
"binary_sensor",
"light",
"select",
"button",
"camera",
]

View File

@@ -0,0 +1,26 @@
from __future__ import annotations
from typing import Any
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .api import KobraXApiClient, KobraXApiError
from .const import DOMAIN, UPDATE_INTERVAL
class KobraXCoordinator(DataUpdateCoordinator[dict[str, Any]]):
def __init__(self, hass: HomeAssistant, api: KobraXApiClient) -> None:
super().__init__(
hass,
logger=hass.data[DOMAIN]["logger"],
name=DOMAIN,
update_interval=UPDATE_INTERVAL,
)
self.api = api
async def _async_update_data(self) -> dict[str, Any]:
try:
return await self.api.async_get_state()
except KobraXApiError as err:
raise UpdateFailed(str(err)) from err

View File

@@ -0,0 +1,25 @@
from __future__ import annotations
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, MANUFACTURER
class KobraXEntity(CoordinatorEntity):
def __init__(self, coordinator, entry, key: str, name: str) -> None:
super().__init__(coordinator)
self._entry = entry
self._attr_unique_id = f"{entry.entry_id}_{key}"
self._attr_name = f"{entry.data['printer_name']} {name}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, entry.entry_id)},
manufacturer=MANUFACTURER,
model="Kobra X",
name=entry.data["printer_name"],
configuration_url=entry.data.get("host"),
)
@property
def state_data(self) -> dict:
return self.coordinator.data or {}

View File

@@ -0,0 +1,38 @@
from __future__ import annotations
from homeassistant.components.light import ColorMode, LightEntity
from homeassistant.exceptions import ServiceValidationError
from .api import KobraXApiError
from .const import DOMAIN
from .entity import KobraXEntity
class KobraXLight(KobraXEntity, LightEntity):
_attr_color_mode = ColorMode.ONOFF
_attr_supported_color_modes = {ColorMode.ONOFF}
@property
def is_on(self) -> bool:
return bool(self.state_data.get("light_on", False))
async def async_turn_on(self, **kwargs) -> None:
api = self.hass.data[DOMAIN][self._entry.entry_id]["api"]
try:
await api.async_set_light(True)
await self.coordinator.async_request_refresh()
except KobraXApiError as err:
raise ServiceValidationError(str(err)) from err
async def async_turn_off(self, **kwargs) -> None:
api = self.hass.data[DOMAIN][self._entry.entry_id]["api"]
try:
await api.async_set_light(False)
await self.coordinator.async_request_refresh()
except KobraXApiError as err:
raise ServiceValidationError(str(err)) from err
async def async_setup_entry(hass, entry, async_add_entities):
coordinator = hass.data[DOMAIN][entry.entry_id]["coordinator"]
async_add_entities([KobraXLight(coordinator, entry, "light", "Light")])

View File

@@ -0,0 +1,12 @@
{
"domain": "kobrax",
"name": "Kobra X",
"codeowners": [
"@Gangoke"
],
"config_flow": true,
"documentation": "https://gitea.it-drui.de/viewit/KX-Bridge-Release",
"iot_class": "local_polling",
"issue_tracker": "https://gitea.it-drui.de/viewit/KX-Bridge-Release/issues",
"version": "0.1.0"
}

View File

@@ -0,0 +1,41 @@
from __future__ import annotations
from homeassistant.components.select import SelectEntity
from homeassistant.exceptions import ServiceValidationError
from .api import KobraXApiError
from .const import DOMAIN
from .entity import KobraXEntity
class KobraXPrintSpeedSelect(KobraXEntity, SelectEntity):
_attr_options = ["Slow (1)", "Normal (2)", "Fast (3)"]
def _mode_to_option(self, mode: int | None) -> str:
mapping = {
1: "Slow (1)",
2: "Normal (2)",
3: "Fast (3)",
}
return mapping.get(mode or 2, "Normal (2)")
@property
def current_option(self) -> str:
mode = self.state_data.get("print_speed_mode")
return self._mode_to_option(int(mode) if mode is not None else None)
async def async_select_option(self, option: str) -> None:
mode = int(option.split("(")[-1].replace(")", ""))
api = self.hass.data[DOMAIN][self._entry.entry_id]["api"]
try:
await api.async_set_speed_mode(mode)
await self.coordinator.async_request_refresh()
except KobraXApiError as err:
raise ServiceValidationError(str(err)) from err
async def async_setup_entry(hass, entry, async_add_entities):
coordinator = hass.data[DOMAIN][entry.entry_id]["coordinator"]
async_add_entities(
[KobraXPrintSpeedSelect(coordinator, entry, "print_speed_mode", "Print Speed")]
)

View File

@@ -0,0 +1,121 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity, SensorEntityDescription
from homeassistant.const import PERCENTAGE, UnitOfTemperature
from .const import DOMAIN
from .entity import KobraXEntity
@dataclass(frozen=True, kw_only=True)
class KobraXSensorDescription(SensorEntityDescription):
value_key: str
SENSORS: tuple[KobraXSensorDescription, ...] = (
KobraXSensorDescription(
key="state",
name="State",
value_key="kobra_state",
icon="mdi:printer-3d",
),
KobraXSensorDescription(
key="print_state",
name="Print State",
value_key="print_state",
icon="mdi:state-machine",
),
KobraXSensorDescription(
key="progress",
name="Progress",
value_key="progress",
native_unit_of_measurement=PERCENTAGE,
icon="mdi:percent",
),
KobraXSensorDescription(
key="hotend_temp",
name="Hotend Temperature",
value_key="nozzle_temp",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
),
KobraXSensorDescription(
key="target_hotend_temp",
name="Target Hotend Temperature",
value_key="nozzle_target",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
entity_registry_enabled_default=False,
),
KobraXSensorDescription(
key="bed_temp",
name="Bed Temperature",
value_key="bed_temp",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
),
KobraXSensorDescription(
key="target_bed_temp",
name="Target Bed Temperature",
value_key="bed_target",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
entity_registry_enabled_default=False,
),
KobraXSensorDescription(
key="filename",
name="Filename",
value_key="filename",
icon="mdi:file",
),
KobraXSensorDescription(
key="current_layer",
name="Current Layer",
value_key="curr_layer",
icon="mdi:layers-triple",
),
KobraXSensorDescription(
key="total_layers",
name="Total Layers",
value_key="total_layers",
icon="mdi:layers",
entity_registry_enabled_default=False,
),
KobraXSensorDescription(
key="remaining_time",
name="Remaining Time",
value_key="remain_time",
icon="mdi:timer-sand",
),
KobraXSensorDescription(
key="print_duration",
name="Print Duration",
value_key="print_duration",
icon="mdi:timer-outline",
entity_registry_enabled_default=False,
),
)
class KobraXSensor(KobraXEntity, SensorEntity):
entity_description: KobraXSensorDescription
@property
def native_value(self) -> Any:
value = self.state_data.get(self.entity_description.value_key)
if self.entity_description.value_key == "progress" and value is not None:
return round(float(value) * 100, 1)
return value
async def async_setup_entry(hass, entry, async_add_entities):
coordinator = hass.data[DOMAIN][entry.entry_id]["coordinator"]
async_add_entities(
[
KobraXSensor(coordinator, entry, description.key, description.name)
for description in SENSORS
]
)

View File

@@ -0,0 +1,20 @@
{
"config": {
"step": {
"user": {
"title": "Kobra X",
"description": "Connect to your KX-Bridge instance",
"data": {
"host": "KX-Bridge URL",
"printer_name": "Printer name"
}
}
},
"error": {
"cannot_connect": "Could not connect to KX-Bridge"
},
"abort": {
"already_configured": "This KX-Bridge URL is already configured"
}
}
}

View File

@@ -0,0 +1,20 @@
{
"config": {
"step": {
"user": {
"title": "Kobra X",
"description": "Connect to your KX-Bridge instance",
"data": {
"host": "KX-Bridge URL",
"printer_name": "Printer name"
}
}
},
"error": {
"cannot_connect": "Could not connect to KX-Bridge"
},
"abort": {
"already_configured": "This KX-Bridge URL is already configured"
}
}
}

10
hacs.json Normal file
View File

@@ -0,0 +1,10 @@
{
"name": "Kobra X LAN",
"content_in_root": false,
"domains": [
"kobrax"
],
"homeassistant": "2024.6.0",
"iot_class": "local_polling",
"render_readme": true
}