Alex CubeandClaude Opus 5.5
9e06f3ebd0
- printer.py: protocol with flow control (pause / resend / done) - lxd02.py: print text or pictures from the command line - windows/: POS-58 driver + local ESC/POS server = a regular Windows printer, Explorer menu for pictures - custom_components/lxd02: Home Assistant integration with a persistent queue Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
108 lines
4.1 KiB
Python
108 lines
4.1 KiB
Python
"""LX-D02 thermal printer: service lxd02.print_text.
|
|
|
|
Jobs go to a persistent queue (.storage/lxd02). If the printer is off, they wait
|
|
and print when it shows up in Bluetooth again.
|
|
|
|
configuration.yaml:
|
|
lxd02:
|
|
mac: AA:BB:CC:DD:EE:FF
|
|
"""
|
|
import asyncio
|
|
from datetime import timedelta
|
|
import logging
|
|
|
|
from bleak.exc import BleakError
|
|
import voluptuous as vol
|
|
from bleak_retry_connector import BleakClientWithServiceCache, establish_connection
|
|
|
|
from homeassistant.components import bluetooth
|
|
from homeassistant.core import HomeAssistant, ServiceCall, callback
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
import homeassistant.helpers.config_validation as cv
|
|
from homeassistant.helpers.event import async_track_time_interval
|
|
from homeassistant.helpers.storage import Store
|
|
|
|
from .printer import print_payloads, render, to_payloads
|
|
|
|
DOMAIN = "lxd02"
|
|
_LOGGER = logging.getLogger(__name__)
|
|
RETRY = timedelta(seconds=60)
|
|
|
|
CONFIG_SCHEMA = vol.Schema(
|
|
{DOMAIN: vol.Schema({vol.Required("mac"): cv.string})}, extra=vol.ALLOW_EXTRA
|
|
)
|
|
|
|
PRINT_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Required("text"): cv.string,
|
|
vol.Optional("size", default=28): vol.All(vol.Coerce(int), vol.Range(min=10, max=120)),
|
|
vol.Optional("density", default=3): vol.All(vol.Coerce(int), vol.Range(min=0, max=7)),
|
|
vol.Optional("feed", default=40): vol.All(vol.Coerce(int), vol.Range(min=0, max=500)),
|
|
vol.Optional("separator", default=False): cv.boolean,
|
|
}
|
|
)
|
|
|
|
|
|
async def async_setup(hass: HomeAssistant, config) -> bool:
|
|
mac = config[DOMAIN]["mac"].upper()
|
|
lock = asyncio.Lock()
|
|
store = Store(hass, 1, DOMAIN)
|
|
queue = await store.async_load() or []
|
|
|
|
async def print_job(device, job):
|
|
img = await hass.async_add_executor_job(
|
|
render, job["text"], job["size"], job["separator"]
|
|
)
|
|
payloads = to_payloads(img)
|
|
try:
|
|
client = await establish_connection(BleakClientWithServiceCache, device, "LX-D02")
|
|
except BleakError as err:
|
|
raise HomeAssistantError(f"LX-D02 connect failed: {err!r}") from err
|
|
try:
|
|
await print_payloads(client, mac, payloads, job["density"], job["feed"])
|
|
except (asyncio.TimeoutError, RuntimeError, BleakError) as err:
|
|
raise HomeAssistantError(f"LX-D02 print failed: {err!r}") from err
|
|
finally:
|
|
await client.disconnect()
|
|
|
|
async def flush():
|
|
async with lock:
|
|
while queue:
|
|
device = bluetooth.async_ble_device_from_address(hass, mac, connectable=True)
|
|
if device is None:
|
|
_LOGGER.info("LX-D02 not in range, %d job(s) queued", len(queue))
|
|
return
|
|
try:
|
|
await print_job(device, queue[0])
|
|
except HomeAssistantError as err:
|
|
_LOGGER.warning("%s; %d job(s) stay queued", err, len(queue))
|
|
return
|
|
except Exception: # bad job (render etc.): drop it, don't block the queue
|
|
_LOGGER.exception("LX-D02 job dropped: %r", queue[0])
|
|
queue.pop(0)
|
|
await store.async_save(queue)
|
|
|
|
async def print_text(call: ServiceCall) -> None:
|
|
queue.append(dict(call.data))
|
|
await store.async_save(queue)
|
|
await flush()
|
|
|
|
@callback
|
|
def seen(service_info, change) -> None:
|
|
if queue and not lock.locked():
|
|
hass.async_create_task(flush())
|
|
|
|
# HA calls `seen` only when the advert changes or the printer comes back after
|
|
# being marked unavailable, not on every packet. So a failed attempt (or a replay
|
|
# of stale history right after HA restart) is never retried by adverts alone:
|
|
# the timer retries while the queue is not empty.
|
|
async_track_time_interval(hass, callback(lambda _: seen(None, None)), RETRY)
|
|
bluetooth.async_register_callback(
|
|
hass,
|
|
seen,
|
|
bluetooth.BluetoothCallbackMatcher(address=mac, connectable=True),
|
|
bluetooth.BluetoothScanningMode.ACTIVE,
|
|
)
|
|
hass.services.async_register(DOMAIN, "print_text", print_text, schema=PRINT_SCHEMA)
|
|
return True
|