595 lines
19 KiB
Python
595 lines
19 KiB
Python
"""The Riedel RRCS integration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import secrets
|
|
from typing import Any
|
|
|
|
import voluptuous as vol
|
|
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
|
|
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_SCAN_INTERVAL, Platform
|
|
from homeassistant.core import HomeAssistant, ServiceCall, ServiceResponse, SupportsResponse
|
|
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError, ServiceValidationError
|
|
from homeassistant.helpers import config_validation as cv
|
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
|
|
|
from .const import (
|
|
ATTR_ENTRY_ID,
|
|
CONF_CALLBACK_PORT,
|
|
CONF_CALLBACK_TOKEN,
|
|
CONF_DISCOVER_GPIO,
|
|
CONF_GPIO_ENTITIES,
|
|
CONF_NOTIFICATIONS,
|
|
CONF_POLL_GPIO,
|
|
CONF_RPC_PATH,
|
|
CONF_TIMEOUT,
|
|
CONF_TRANSKEY_PREFIX,
|
|
DEFAULT_DISCOVER_GPIO,
|
|
DEFAULT_NOTIFICATIONS,
|
|
DEFAULT_PATH,
|
|
DEFAULT_POLL_GPIO,
|
|
DEFAULT_PORT,
|
|
DEFAULT_SCAN_INTERVAL,
|
|
DEFAULT_TIMEOUT,
|
|
DEFAULT_TRANSKEY_PREFIX,
|
|
DOMAIN,
|
|
NOTIFICATION_URL_TEMPLATE,
|
|
SERVICE_CALL_METHOD,
|
|
SERVICE_CLEAR_KEY_LABEL,
|
|
SERVICE_KILL_XP,
|
|
SERVICE_PRESS_KEY,
|
|
SERVICE_SET_GP_OUTPUT,
|
|
SERVICE_SET_INPUT_GAIN,
|
|
SERVICE_SET_KEY_LABEL,
|
|
SERVICE_SET_LOGIC_SOURCE,
|
|
SERVICE_SET_OUTPUT_GAIN,
|
|
SERVICE_SET_PORT_ALIAS,
|
|
SERVICE_SET_XP,
|
|
SERVICE_SET_XP_VOLUME,
|
|
)
|
|
from .coordinator import RRCSCoordinator
|
|
from .models import RRCSConfigEntry, RRCSRuntimeData
|
|
from .notification import (
|
|
RRCSNotificationDispatcher,
|
|
async_register_listener,
|
|
async_register_view,
|
|
async_remove_listener,
|
|
)
|
|
from .rrcs import (
|
|
GpioAddress,
|
|
RRCSClient,
|
|
RRCSConnectionError,
|
|
RRCSError,
|
|
parse_gpio_config,
|
|
)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
PLATFORMS: list[Platform] = [
|
|
Platform.BINARY_SENSOR,
|
|
Platform.EVENT,
|
|
Platform.SENSOR,
|
|
Platform.SWITCH,
|
|
]
|
|
|
|
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
|
|
|
|
|
|
async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool:
|
|
"""Register the integration's services."""
|
|
_async_register_services(hass)
|
|
return True
|
|
|
|
|
|
async def async_setup_entry(hass: HomeAssistant, entry: RRCSConfigEntry) -> bool:
|
|
"""Set up Riedel RRCS from a config entry."""
|
|
options = {**entry.data, **entry.options}
|
|
|
|
client = RRCSClient(
|
|
session=async_get_clientsession(hass),
|
|
host=entry.data[CONF_HOST],
|
|
port=entry.data.get(CONF_PORT, DEFAULT_PORT),
|
|
path=options.get(CONF_RPC_PATH, DEFAULT_PATH),
|
|
timeout=options.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
|
|
transkey_prefix=options.get(CONF_TRANSKEY_PREFIX, DEFAULT_TRANSKEY_PREFIX),
|
|
)
|
|
|
|
# Fail fast and let HA retry rather than creating half a device.
|
|
try:
|
|
await client.get_version()
|
|
except RRCSError as err:
|
|
raise ConfigEntryNotReady(f"Cannot reach RRCS at {client.url}: {err}") from err
|
|
|
|
coordinator = RRCSCoordinator(
|
|
hass,
|
|
entry,
|
|
client,
|
|
scan_interval=options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL),
|
|
poll_gpio=options.get(CONF_POLL_GPIO, DEFAULT_POLL_GPIO),
|
|
)
|
|
|
|
gpio_inputs, gpio_outputs, gpio_names = await _async_collect_gpios(client, options)
|
|
coordinator.set_gpios(gpio_inputs, gpio_outputs, gpio_names)
|
|
|
|
runtime = RRCSRuntimeData(
|
|
client=client,
|
|
coordinator=coordinator,
|
|
gpio_inputs=gpio_inputs,
|
|
gpio_outputs=gpio_outputs,
|
|
gpio_names=gpio_names,
|
|
)
|
|
entry.runtime_data = runtime
|
|
|
|
if options.get(CONF_NOTIFICATIONS, DEFAULT_NOTIFICATIONS):
|
|
await _async_start_notifications(hass, entry, runtime, options)
|
|
|
|
await coordinator.async_config_entry_first_refresh()
|
|
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
|
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
|
|
return True
|
|
|
|
|
|
async def async_unload_entry(hass: HomeAssistant, entry: RRCSConfigEntry) -> bool:
|
|
"""Unload a config entry."""
|
|
runtime = entry.runtime_data
|
|
|
|
if runtime.callback_token:
|
|
async_remove_listener(hass, runtime.callback_token)
|
|
if runtime.callback_port and runtime.callback_path:
|
|
try:
|
|
await runtime.client.unregister_for_all_events(
|
|
runtime.callback_port, runtime.callback_path
|
|
)
|
|
except RRCSError as err:
|
|
_LOGGER.debug("Could not unregister notifications cleanly: %s", err)
|
|
|
|
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
|
|
|
|
|
async def _async_update_listener(hass: HomeAssistant, entry: RRCSConfigEntry) -> None:
|
|
"""Reload the entry when its options change."""
|
|
await hass.config_entries.async_reload(entry.entry_id)
|
|
|
|
|
|
async def _async_start_notifications(
|
|
hass: HomeAssistant,
|
|
entry: RRCSConfigEntry,
|
|
runtime: RRCSRuntimeData,
|
|
options: dict[str, Any],
|
|
) -> None:
|
|
"""Stand up the callback endpoint and register it with the gateway."""
|
|
token = entry.data.get(CONF_CALLBACK_TOKEN)
|
|
if not token:
|
|
token = secrets.token_hex(8)
|
|
hass.config_entries.async_update_entry(
|
|
entry, data={**entry.data, CONF_CALLBACK_TOKEN: token}
|
|
)
|
|
|
|
port = options.get(CONF_CALLBACK_PORT) or hass.http.server_port
|
|
path = NOTIFICATION_URL_TEMPLATE.format(token=token)
|
|
|
|
if getattr(hass.config.api, "use_ssl", False) and not options.get(CONF_CALLBACK_PORT):
|
|
_LOGGER.warning(
|
|
"Home Assistant is serving HTTPS on port %s but RRCS pushes plain HTTP. "
|
|
"Set a plain-HTTP callback port in the integration options or "
|
|
"notifications will not arrive",
|
|
port,
|
|
)
|
|
|
|
async_register_view(hass)
|
|
dispatcher = RRCSNotificationDispatcher(hass, entry.entry_id, runtime.coordinator)
|
|
async_register_listener(hass, token, dispatcher.handle)
|
|
|
|
try:
|
|
await runtime.client.register_for_all_events(port, path)
|
|
except RRCSError as err:
|
|
# Not fatal: polling still works, and the coordinator retries the
|
|
# registration on every refresh.
|
|
_LOGGER.warning("Could not register for RRCS notifications: %s", err)
|
|
|
|
runtime.callback_token = token
|
|
runtime.callback_port = port
|
|
runtime.callback_path = path
|
|
runtime.coordinator.set_registration(port, path)
|
|
|
|
|
|
async def _async_collect_gpios(
|
|
client: RRCSClient, options: dict[str, Any]
|
|
) -> tuple[list[GpioAddress], list[GpioAddress], dict[str, str]]:
|
|
"""Build the GPIO entity list from discovery plus manual configuration."""
|
|
addresses: dict[str, GpioAddress] = {}
|
|
names: dict[str, str] = {}
|
|
|
|
if options.get(CONF_DISCOVER_GPIO, DEFAULT_DISCOVER_GPIO):
|
|
try:
|
|
for address in await client.discover_gpios():
|
|
addresses[address.key] = address
|
|
except RRCSError as err:
|
|
_LOGGER.debug("GPIO discovery failed: %s", err)
|
|
|
|
for address, name in parse_gpio_config(options.get(CONF_GPIO_ENTITIES)):
|
|
addresses[address.key] = address
|
|
if name:
|
|
names[address.key] = name
|
|
|
|
inputs = [address for address in addresses.values() if address.is_input]
|
|
outputs = [address for address in addresses.values() if not address.is_input]
|
|
return inputs, outputs, names
|
|
|
|
|
|
# --- Services ----------------------------------------------------------------------
|
|
|
|
_ENTRY_FIELD = {vol.Optional(ATTR_ENTRY_ID): cv.string}
|
|
|
|
_XP_FIELDS = {
|
|
vol.Required("source_net", default=1): vol.Coerce(int),
|
|
vol.Required("source_node"): vol.Coerce(int),
|
|
vol.Required("source_port"): vol.Coerce(int),
|
|
vol.Required("dest_net", default=1): vol.Coerce(int),
|
|
vol.Required("dest_node"): vol.Coerce(int),
|
|
vol.Required("dest_port"): vol.Coerce(int),
|
|
}
|
|
|
|
_KEY_FIELDS = {
|
|
vol.Required("node"): vol.Coerce(int),
|
|
vol.Required("port"): vol.Coerce(int),
|
|
vol.Optional("is_input", default=False): cv.boolean,
|
|
vol.Optional("page", default=1): vol.Coerce(int),
|
|
vol.Optional("expansion_panel", default=0): vol.Coerce(int),
|
|
vol.Required("key_number"): vol.Coerce(int),
|
|
vol.Optional("is_virt_key", default=False): cv.boolean,
|
|
}
|
|
|
|
SET_XP_SCHEMA = vol.Schema(
|
|
{
|
|
**_ENTRY_FIELD,
|
|
**_XP_FIELDS,
|
|
vol.Optional("priority"): vol.All(vol.Coerce(int), vol.Range(min=0, max=4)),
|
|
vol.Optional("destructive", default=False): cv.boolean,
|
|
}
|
|
)
|
|
|
|
KILL_XP_SCHEMA = vol.Schema({**_ENTRY_FIELD, **_XP_FIELDS})
|
|
|
|
SET_XP_VOLUME_SCHEMA = vol.Schema(
|
|
{
|
|
**_ENTRY_FIELD,
|
|
**_XP_FIELDS,
|
|
vol.Optional("single", default=True): cv.boolean,
|
|
vol.Optional("conference", default=False): cv.boolean,
|
|
vol.Required("volume"): vol.All(vol.Coerce(int), vol.Range(min=0, max=256)),
|
|
}
|
|
)
|
|
|
|
SET_GP_OUTPUT_SCHEMA = vol.Schema(
|
|
{
|
|
**_ENTRY_FIELD,
|
|
vol.Optional("net", default=1): vol.Coerce(int),
|
|
vol.Required("node"): vol.Coerce(int),
|
|
vol.Optional("port", default=128): vol.Coerce(int),
|
|
vol.Optional("slot", default=0): vol.Coerce(int),
|
|
vol.Required("index"): vol.Coerce(int),
|
|
vol.Required("state"): cv.boolean,
|
|
}
|
|
)
|
|
|
|
SET_LOGIC_SOURCE_SCHEMA = vol.Schema(
|
|
{
|
|
**_ENTRY_FIELD,
|
|
vol.Required("object_id"): vol.Coerce(int),
|
|
vol.Required("state"): cv.boolean,
|
|
}
|
|
)
|
|
|
|
PRESS_KEY_SCHEMA = vol.Schema(
|
|
{
|
|
**_ENTRY_FIELD,
|
|
**_KEY_FIELDS,
|
|
vol.Optional("press", default=True): cv.boolean,
|
|
vol.Optional("trigger"): vol.All(vol.Coerce(int), vol.Range(min=1, max=2)),
|
|
vol.Optional("pool_port", default=0): vol.Coerce(int),
|
|
}
|
|
)
|
|
|
|
SET_KEY_LABEL_SCHEMA = vol.Schema(
|
|
{
|
|
**_ENTRY_FIELD,
|
|
**_KEY_FIELDS,
|
|
vol.Required("label"): vol.All(cv.string, vol.Length(max=8)),
|
|
vol.Optional("marker"): vol.Coerce(int),
|
|
}
|
|
)
|
|
|
|
CLEAR_KEY_LABEL_SCHEMA = vol.Schema(
|
|
{**_ENTRY_FIELD, **_KEY_FIELDS, vol.Optional("clear_marker", default=False): cv.boolean}
|
|
)
|
|
|
|
SET_PORT_ALIAS_SCHEMA = vol.Schema(
|
|
{
|
|
**_ENTRY_FIELD,
|
|
vol.Optional("net", default=1): vol.Coerce(int),
|
|
vol.Required("node"): vol.Coerce(int),
|
|
vol.Required("port"): vol.Coerce(int),
|
|
vol.Required("alias"): vol.All(cv.string, vol.Length(max=8)),
|
|
vol.Optional("is_input", default=False): cv.boolean,
|
|
}
|
|
)
|
|
|
|
_GAIN_SCHEMA = vol.Schema(
|
|
{
|
|
**_ENTRY_FIELD,
|
|
vol.Optional("net", default=1): vol.Coerce(int),
|
|
vol.Required("node"): vol.Coerce(int),
|
|
vol.Required("port"): vol.Coerce(int),
|
|
# Half-decibel steps; -128 mutes.
|
|
vol.Required("gain"): vol.All(vol.Coerce(int), vol.Range(min=-128, max=36)),
|
|
}
|
|
)
|
|
|
|
CALL_METHOD_SCHEMA = vol.Schema(
|
|
{
|
|
**_ENTRY_FIELD,
|
|
vol.Required("method"): cv.string,
|
|
vol.Optional("params", default=list): vol.Any(list, dict),
|
|
vol.Optional("include_transkey", default=True): cv.boolean,
|
|
}
|
|
)
|
|
|
|
|
|
def _resolve_client(hass: HomeAssistant, call: ServiceCall) -> RRCSClient:
|
|
"""Find the client a service call is aimed at."""
|
|
entries = [
|
|
entry
|
|
for entry in hass.config_entries.async_entries(DOMAIN)
|
|
if entry.state is ConfigEntryState.LOADED
|
|
]
|
|
entry_id = call.data.get(ATTR_ENTRY_ID)
|
|
if entry_id:
|
|
for entry in entries:
|
|
if entry.entry_id == entry_id:
|
|
return entry.runtime_data.client
|
|
raise ServiceValidationError(f"No loaded RRCS config entry with id {entry_id}")
|
|
if not entries:
|
|
raise ServiceValidationError("No RRCS gateway is currently loaded")
|
|
if len(entries) > 1:
|
|
raise ServiceValidationError(
|
|
"Several RRCS gateways are configured; pass entry_id to pick one"
|
|
)
|
|
return entries[0].runtime_data.client
|
|
|
|
|
|
def _key_args(data: dict[str, Any]) -> list[Any]:
|
|
"""Build the shared key addressing arguments."""
|
|
return [
|
|
data["node"],
|
|
data["port"],
|
|
data["is_input"],
|
|
data["page"],
|
|
data["expansion_panel"],
|
|
data["key_number"],
|
|
data["is_virt_key"],
|
|
]
|
|
|
|
|
|
def _async_register_services(hass: HomeAssistant) -> None:
|
|
"""Register every RRCS service, once."""
|
|
if hass.services.has_service(DOMAIN, SERVICE_SET_XP):
|
|
return
|
|
|
|
async def _guard(coro) -> Any:
|
|
try:
|
|
return await coro
|
|
except RRCSConnectionError as err:
|
|
raise HomeAssistantError(f"RRCS unreachable: {err}") from err
|
|
except RRCSError as err:
|
|
raise HomeAssistantError(str(err)) from err
|
|
|
|
async def set_xp(call: ServiceCall) -> None:
|
|
client = _resolve_client(hass, call)
|
|
data = call.data
|
|
args = [
|
|
data["source_net"],
|
|
data["source_node"],
|
|
data["source_port"],
|
|
data["dest_net"],
|
|
data["dest_node"],
|
|
data["dest_port"],
|
|
]
|
|
priority = data.get("priority")
|
|
if data.get("destructive"):
|
|
await _guard(
|
|
client.call("SetXpDestructive", *args, priority or 1, check=True)
|
|
)
|
|
elif priority is not None:
|
|
await _guard(client.call("SetXpPrio", *args, priority, check=True))
|
|
else:
|
|
await _guard(client.call("SetXp", *args, check=True))
|
|
|
|
async def kill_xp(call: ServiceCall) -> None:
|
|
client = _resolve_client(hass, call)
|
|
data = call.data
|
|
await _guard(
|
|
client.call(
|
|
"KillXp",
|
|
data["source_net"],
|
|
data["source_node"],
|
|
data["source_port"],
|
|
data["dest_net"],
|
|
data["dest_node"],
|
|
data["dest_port"],
|
|
check=True,
|
|
)
|
|
)
|
|
|
|
async def set_xp_volume(call: ServiceCall) -> None:
|
|
client = _resolve_client(hass, call)
|
|
data = call.data
|
|
await _guard(
|
|
client.call(
|
|
"SetXpVolume",
|
|
data["source_net"],
|
|
data["source_node"],
|
|
data["source_port"],
|
|
data["dest_net"],
|
|
data["dest_node"],
|
|
data["dest_port"],
|
|
data["single"],
|
|
data["conference"],
|
|
data["volume"],
|
|
check=True,
|
|
)
|
|
)
|
|
|
|
async def set_gp_output(call: ServiceCall) -> None:
|
|
client = _resolve_client(hass, call)
|
|
data = call.data
|
|
address = GpioAddress(
|
|
net=data["net"],
|
|
node=data["node"],
|
|
port=data["port"],
|
|
slot=data["slot"],
|
|
index=data["index"],
|
|
is_input=False,
|
|
)
|
|
await _guard(client.set_gp_output(address, data["state"]))
|
|
|
|
async def set_logic_source(call: ServiceCall) -> None:
|
|
client = _resolve_client(hass, call)
|
|
await _guard(
|
|
client.set_logic_source(call.data["object_id"], call.data["state"])
|
|
)
|
|
|
|
async def press_key(call: ServiceCall) -> None:
|
|
client = _resolve_client(hass, call)
|
|
data = call.data
|
|
args = _key_args(data)
|
|
if data.get("trigger") is not None:
|
|
await _guard(
|
|
client.call(
|
|
"PressKeyEx",
|
|
*args,
|
|
data["press"],
|
|
data["trigger"],
|
|
data["pool_port"],
|
|
)
|
|
)
|
|
else:
|
|
await _guard(
|
|
client.call(
|
|
"PressKey", *args, data["press"], data["pool_port"], check=True
|
|
)
|
|
)
|
|
|
|
async def set_key_label(call: ServiceCall) -> None:
|
|
client = _resolve_client(hass, call)
|
|
data = call.data
|
|
args = _key_args(data)
|
|
if data.get("marker") is not None:
|
|
await _guard(
|
|
client.call(
|
|
"SetKeyLabelAndMarker",
|
|
*args,
|
|
data["label"],
|
|
data["marker"],
|
|
check=True,
|
|
)
|
|
)
|
|
else:
|
|
await _guard(client.call("SetKeyLabel", *args, data["label"], check=True))
|
|
|
|
async def clear_key_label(call: ServiceCall) -> None:
|
|
client = _resolve_client(hass, call)
|
|
method = (
|
|
"ClearKeyLabelAndMarker" if call.data["clear_marker"] else "ClearKeyLabel"
|
|
)
|
|
await _guard(client.call(method, *_key_args(call.data), check=True))
|
|
|
|
async def set_port_alias(call: ServiceCall) -> None:
|
|
client = _resolve_client(hass, call)
|
|
data = call.data
|
|
await _guard(
|
|
client.call(
|
|
"SetPortAlias",
|
|
data["net"],
|
|
data["node"],
|
|
data["port"],
|
|
data["alias"],
|
|
data["is_input"],
|
|
check=True,
|
|
)
|
|
)
|
|
|
|
def _gain_handler(method: str):
|
|
async def handler(call: ServiceCall) -> None:
|
|
client = _resolve_client(hass, call)
|
|
data = call.data
|
|
await _guard(
|
|
client.call(
|
|
method, data["net"], data["node"], data["port"], data["gain"], check=True
|
|
)
|
|
)
|
|
|
|
return handler
|
|
|
|
async def call_method(call: ServiceCall) -> ServiceResponse:
|
|
client = _resolve_client(hass, call)
|
|
params = call.data["params"]
|
|
if isinstance(params, dict):
|
|
params = list(params.values())
|
|
if call.data["include_transkey"]:
|
|
result = await _guard(client.call(call.data["method"], *params))
|
|
else:
|
|
result = await _guard(
|
|
client.call_raw(call.data["method"], tuple(params))
|
|
)
|
|
return {"result": _jsonable(result)}
|
|
|
|
hass.services.async_register(DOMAIN, SERVICE_SET_XP, set_xp, SET_XP_SCHEMA)
|
|
hass.services.async_register(DOMAIN, SERVICE_KILL_XP, kill_xp, KILL_XP_SCHEMA)
|
|
hass.services.async_register(
|
|
DOMAIN, SERVICE_SET_XP_VOLUME, set_xp_volume, SET_XP_VOLUME_SCHEMA
|
|
)
|
|
hass.services.async_register(
|
|
DOMAIN, SERVICE_SET_GP_OUTPUT, set_gp_output, SET_GP_OUTPUT_SCHEMA
|
|
)
|
|
hass.services.async_register(
|
|
DOMAIN, SERVICE_SET_LOGIC_SOURCE, set_logic_source, SET_LOGIC_SOURCE_SCHEMA
|
|
)
|
|
hass.services.async_register(DOMAIN, SERVICE_PRESS_KEY, press_key, PRESS_KEY_SCHEMA)
|
|
hass.services.async_register(
|
|
DOMAIN, SERVICE_SET_KEY_LABEL, set_key_label, SET_KEY_LABEL_SCHEMA
|
|
)
|
|
hass.services.async_register(
|
|
DOMAIN, SERVICE_CLEAR_KEY_LABEL, clear_key_label, CLEAR_KEY_LABEL_SCHEMA
|
|
)
|
|
hass.services.async_register(
|
|
DOMAIN, SERVICE_SET_PORT_ALIAS, set_port_alias, SET_PORT_ALIAS_SCHEMA
|
|
)
|
|
hass.services.async_register(
|
|
DOMAIN, SERVICE_SET_INPUT_GAIN, _gain_handler("SetInputGain"), _GAIN_SCHEMA
|
|
)
|
|
hass.services.async_register(
|
|
DOMAIN, SERVICE_SET_OUTPUT_GAIN, _gain_handler("SetOutputGain"), _GAIN_SCHEMA
|
|
)
|
|
hass.services.async_register(
|
|
DOMAIN,
|
|
SERVICE_CALL_METHOD,
|
|
call_method,
|
|
CALL_METHOD_SCHEMA,
|
|
supports_response=SupportsResponse.OPTIONAL,
|
|
)
|
|
|
|
|
|
def _jsonable(value: Any) -> Any:
|
|
"""Make a decoded XML-RPC value safe to hand back as a service response."""
|
|
if isinstance(value, dict):
|
|
return {str(key): _jsonable(item) for key, item in value.items()}
|
|
if isinstance(value, (list, tuple)):
|
|
return [_jsonable(item) for item in value]
|
|
if isinstance(value, (str, int, float, bool)) or value is None:
|
|
return value
|
|
if isinstance(value, bytes):
|
|
return value.decode("utf-8", errors="replace")
|
|
return str(value)
|