"""HTTP REST API for the Fichero D11s thermal label printer. Start with: fichero-server [--host HOST] [--port PORT] or: python -m fichero.api Endpoints: GET /status – Printer status GET /info – Printer info (model, firmware, battery, …) POST /print/text – Print a text label POST /print/image – Print an uploaded image file """ from __future__ import annotations import argparse import asyncio import io import re import os from contextlib import asynccontextmanager from pathlib import Path from typing import Annotated from fastapi import FastAPI, File, Form, HTTPException, UploadFile from fastapi.middleware.cors import CORSMiddleware from fastapi.openapi.docs import get_swagger_ui_html from fastapi.responses import HTMLResponse from fastapi.staticfiles import StaticFiles from PIL import Image from bleak import BleakScanner from fichero.cli import DOTS_PER_MM, do_print from fichero.imaging import text_to_image from fichero.printer import ( PAPER_GAP, PrinterError, PrinterNotFound, PrinterTimeout, connect, ) # --------------------------------------------------------------------------- # Global connection settings (env vars or CLI flags at startup) # --------------------------------------------------------------------------- _DEFAULT_ADDRESS: str | None = os.environ.get("FICHERO_ADDR") # Default to BLE transport (most reliable for Fichero/D11s printers) # Set FICHERO_TRANSPORT=classic to force Classic Bluetooth (RFCOMM) _DEFAULT_CLASSIC: bool = os.environ.get("FICHERO_TRANSPORT", "").lower() == "classic" _DEFAULT_CHANNEL: int = int(os.environ.get("FICHERO_CHANNEL", "1")) _PAPER_MAP = {"gap": 0, "black": 1, "continuous": 2} def _parse_paper(value: str) -> int: if value in _PAPER_MAP: return _PAPER_MAP[value] try: val = int(value) if 0 <= val <= 2: return val except ValueError: pass raise HTTPException(status_code=422, detail=f"Invalid paper type '{value}'. Use gap, black, continuous or 0-2.") # --------------------------------------------------------------------------- # FastAPI app # --------------------------------------------------------------------------- @asynccontextmanager async def lifespan(app: FastAPI): # noqa: ARG001 yield app = FastAPI( title="Fichero Printer API", description="REST API for the Fichero D11s (AiYin) thermal label printer.", version = "0.1.37", lifespan=lifespan, docs_url=None, redoc_url=None, ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # Serve static files for the modern web UI (if built and present in 'web' dir) _WEB_ROOT = Path(__file__).parent / "web" if _WEB_ROOT.exists(): # Typical SPA assets folder if (_WEB_ROOT / "assets").exists(): app.mount("/assets", StaticFiles(directory=_WEB_ROOT / "assets"), name="assets") def _address(address: str | None) -> str | None: """Return the effective BLE address (request value overrides env default).""" return address or _DEFAULT_ADDRESS def _ui_html() -> str: default_address = _DEFAULT_ADDRESS or "" default_transport = "classic" if _DEFAULT_CLASSIC else "ble" try: template_path = Path(__file__).parent / "index.html" template = template_path.read_text(encoding="utf-8") except FileNotFoundError: return "

Error: index.html not found

" # Simple substitution for initial values template = ( template.replace("{default_address}", default_address) .replace("{ble_selected}", " selected" if default_transport == "ble" else "") .replace("{classic_selected}", " selected" if default_transport == "classic" else "") .replace("{default_channel}", str(_DEFAULT_CHANNEL)) ) # Inject debug scan section and script scan_html = """

Debug Scan

Scans for all nearby BLE devices to help with debugging connection issues.

📱 Click "Scan for BLE Devices" to search for nearby Bluetooth devices...
""" scan_script = r''' // Scan for BLE Devices function async function scanForDevices() { console.log('Scan function called - checking elements...'); const resultsEl = document.getElementById('scan-results'); const scanButton = document.getElementById('scan-button'); const loadingEl = document.getElementById('scan-loading'); const textEl = document.getElementById('scan-text'); console.log('Elements found:', { resultsEl, scanButton, loadingEl, textEl }); // Show loading state scanButton.disabled = true; loadingEl.style.display = 'inline-block'; textEl.textContent = 'Scanning...'; resultsEl.textContent = '🔍 Searching for BLE devices (this may take up to 10 seconds)...'; console.log('Starting scan request...'); try { const response = await fetch('scan'); let responseData; try { responseData = await response.json(); } catch (e) { const text = await response.text(); throw new Error(`Invalid JSON response: ${text}`); } if (!response.ok) { const errorDetail = responseData.detail || `HTTP error! status: ${response.status}`; throw new Error(errorDetail); } const devices = responseData; if (devices.length === 0) { resultsEl.textContent = '📡 No BLE devices found.\n\nTroubleshooting tips:\n- Make sure your printer is powered on\n- Ensure Bluetooth is enabled on this device\n- Bring the printer closer (within 5 meters)\n- Try restarting the printer'; } else { let resultText = '🎉 Found ' + devices.length + ' device(s):\n\n'; devices.forEach((d, index) => { resultText += `${index + 1}. ${d.name || 'Unknown Device'}\n`; resultText += ` Address: ${d.address}\n`; // Handle case where RSSI might not be available if (d.rssi !== undefined) { resultText += ` Signal: ${d.rssi} dBm (${Math.abs(d.rssi) < 60 ? 'Strong' : Math.abs(d.rssi) < 80 ? 'Good' : 'Weak'})\n`; } else { resultText += ` Signal: Not available\n`; } if (d.metadata) { resultText += ` Metadata: ${d.metadata}\n`; } resultText += ` ${'='.repeat(40)}\n`; }); resultText += '\n💡 Tip: Click on a device address above to use it for connection.'; resultsEl.textContent = resultText; } } catch (e) { resultsEl.textContent = '❌ Error during scan: ' + e.message + '\n\nPossible causes:\n- Bluetooth adapter not available\n- Missing permissions\n- Bluetooth service not running'; console.error('Scan error:', e); } finally { // Reset button state console.log('Resetting button state...'); scanButton.disabled = false; loadingEl.style.display = 'none'; textEl.textContent = 'Scan for BLE Devices (10s)'; console.log('Scan completed. Final result:', resultsEl.textContent); } } ''' # Inject scan HTML after main content if "" in template: parts = template.split("", 1) template = parts[0] + "" + scan_html + parts[1] elif "" in template: parts = template.split("", 1) template = parts[0] + scan_html + "" + parts[1] else: # Fallback if no main or body tag template += scan_html # Inject scan script before the closing tag of the main script if "" in template: parts = template.rsplit("", 1) template = parts[0] + scan_script + "" + parts[1] else: # Fallback if no script tag found template += f"" return template # --------------------------------------------------------------------------- # Endpoints # --------------------------------------------------------------------------- @app.get("/", include_in_schema=False, response_class=HTMLResponse) async def root(legacy: bool = False): """Serve a compact printer UI for Home Assistant.""" # Prefer the modern SPA if available, unless ?legacy=true is used if not legacy and (_WEB_ROOT / "index.html").exists(): return HTMLResponse((_WEB_ROOT / "index.html").read_text(encoding="utf-8")) return HTMLResponse(_ui_html()) @app.get("/docs", include_in_schema=False) async def docs(): """Serve Swagger UI with ingress-safe relative OpenAPI URL.""" return get_swagger_ui_html( openapi_url="openapi.json", title=f"{app.title} - Swagger UI", ) @app.get( "/status", summary="Get printer status", response_description="Current printer status flags", ) async def get_status( address: str | None = None, classic: bool = _DEFAULT_CLASSIC, channel: int = _DEFAULT_CHANNEL, ): """Return the real-time status of the printer (paper, battery, heat, …).""" try: async with connect(_address(address), classic=classic, channel=channel) as pc: status = await pc.get_status() except PrinterNotFound as exc: detail = str(exc) if "BLE" in detail or "BLE" in str(classic): detail += "\n\nBLE Troubleshooting:\n" detail += "- Ensure Home Assistant has Bluetooth permissions (host_dbus: true)\n" detail += "- Make sure the printer is powered on and discoverable\n" detail += "- Try restarting the printer\n" detail += "- Check that no other device is connected to the printer" raise HTTPException(status_code=404, detail=detail) from exc except PrinterTimeout as exc: detail = str(exc) if not classic: # BLE timeout detail += "\n\nBLE Connection Tips:\n" detail += "- Bring the printer closer to the Home Assistant host\n" detail += "- Ensure no Bluetooth interference (WiFi, USB 3.0, microwaves)\n" detail += "- Try restarting the Bluetooth service on your host" raise HTTPException(status_code=504, detail=detail) from exc except PrinterError as exc: detail = str(exc) error_str = str(exc).lower() if not classic and "br-connection-not-supported" in error_str: detail += "\n\n🔧 HOME ASSISTANT BLE PERMISSION FIX:\n" detail += "This error occurs when Home Assistant doesn't have proper Bluetooth permissions.\n\n" detail += "📋 STEP-BY-STEP SOLUTION:\n" detail += "1. Edit your add-on configuration in Home Assistant\n" detail += "2. Add this line to the configuration:\n" detail += " host_dbus: true\n" detail += "3. Save the configuration\n" detail += "4. Restart the Fichero add-on\n" detail += "5. Try connecting again\n\n" detail += "💡 If you're using the Home Assistant OS:\n" detail += "- Go to Settings > Add-ons > Fichero Printer > Configuration\n" detail += "- Add 'host_dbus: true' under the 'host_dbus' section\n" detail += "- This gives the add-on access to the system Bluetooth stack\n\n" detail += "🔄 ALTERNATIVE SOLUTION:\n" detail += "If BLE still doesn't work after adding host_dbus, try Classic Bluetooth:\n" detail += "1. Set 'classic=true' in your API calls\n" detail += "2. Use channel=1 (most common for Fichero printers)\n" detail += "3. Use the 'Pair Device' button in the web UI first" elif not classic and "dbus" in error_str: detail += "\n\nBLE Permission Fix:\n" detail += "1. Add 'host_dbus: true' to your add-on configuration\n" detail += "2. Restart the add-on\n" detail += "3. If using Classic Bluetooth, try: classic=true with channel=1" elif not classic and "connection" in error_str: detail += "\n\nBLE Connection Help:\n" detail += "- Verify the BLE address is correct (not Classic Bluetooth address)\n" detail += "- Ensure no other device is paired/connected to the printer\n" detail += "- Try power cycling the printer" raise HTTPException(status_code=502, detail=detail) from exc return { "ok": status.ok, "printing": status.printing, "cover_open": status.cover_open, "no_paper": status.no_paper, "low_battery": status.low_battery, "overheated": status.overheated, "charging": status.charging, "raw": status.raw, } @app.get( "/info", summary="Get printer info", response_description="Model, firmware, serial number and battery level", ) async def get_info( address: str | None = None, classic: bool = _DEFAULT_CLASSIC, channel: int = _DEFAULT_CHANNEL, ): """Return static and dynamic printer information.""" try: async with connect(_address(address), classic=classic, channel=channel) as pc: info = await pc.get_info() info.update(await pc.get_all_info()) except PrinterNotFound as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except PrinterTimeout as exc: raise HTTPException(status_code=504, detail=str(exc)) from exc except PrinterError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc return info @app.get( "/scan", summary="Scan for BLE devices", response_description="List of discovered BLE devices", ) async def scan_devices(): """Scan for nearby BLE devices for 10 seconds for debugging.""" try: devices = await BleakScanner.discover(timeout=10.0) result = [] for d in devices: device_info = { "address": d.address, "name": d.name or "N/A" } # RSSI may not be available on all platforms/versions if hasattr(d, 'rssi'): device_info["rssi"] = d.rssi if hasattr(d, 'metadata'): device_info["metadata"] = str(d.metadata) result.append(device_info) return result except Exception as exc: # This provides more debug info to the user if scanning fails raise HTTPException( status_code=500, detail=f"An error occurred during BLE scanning: {exc}" ) @app.post( "/pair", summary="Pair and trust a Classic Bluetooth device", status_code=200, description="⚠️ ONLY for Classic Bluetooth (RFCOMM). BLE does not require pairing!", ) async def pair_device( address: Annotated[str | None, Form(description="Device address (optional, overrides FICHERO_ADDR)")] = None, ): """ Attempt to pair and trust the device using `bluetoothctl`. ⚠️ IMPORTANT: This is ONLY for Classic Bluetooth (RFCOMM) connections. BLE connections do NOT require pairing and will NOT work with this endpoint. For BLE issues, ensure: - The printer is powered on and discoverable - Home Assistant has proper Bluetooth permissions (host_dbus: true) - You're using the correct BLE address (not Classic Bluetooth address) """ addr = _address(address) if not addr: raise HTTPException(status_code=422, detail="Address is required to pair.") # Basic validation for MAC address format to mitigate injection risk. if not re.match(r"^[0-9A-F]{2}(:[0-9A-F]{2}){5}$", addr, re.IGNORECASE): raise HTTPException(status_code=422, detail=f"Invalid address format: {addr}") cmd = f'echo -e "pair {addr}\\ntrust {addr}\\nquit" | bluetoothctl' try: proc = await asyncio.create_subprocess_shell( cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=15.0) except FileNotFoundError: raise HTTPException(status_code=500, detail="`bluetoothctl` command not found. Is BlueZ installed and in PATH?") except asyncio.TimeoutError: raise HTTPException(status_code=504, detail="`bluetoothctl` command timed out after 15 seconds.") output = stdout.decode(errors="ignore") error = stderr.decode(errors="ignore") if "Failed to pair" in output or "not available" in output.lower(): raise HTTPException(status_code=502, detail=f"Pairing failed. Output: {output}. Error: {error}") return {"ok": True, "message": "Pair/trust command sent. Check output for details.", "output": output, "error": error} @app.post( "/unpair", summary="Unpair a Bluetooth device", status_code=200, ) async def unpair_device( address: Annotated[str | None, Form(description="Device address (optional, overrides FICHERO_ADDR)")] = None, ): """ Attempt to unpair the device using `bluetoothctl`. """ addr = _address(address) if not addr: raise HTTPException(status_code=422, detail="Address is required to unpair.") # Basic validation for MAC address format to mitigate injection risk. if not re.match(r"^[0-9A-F]{2}(:[0-9A-F]{2}){5}$", addr, re.IGNORECASE): raise HTTPException(status_code=422, detail=f"Invalid address format: {addr}") cmd = f'echo -e "remove {addr}\\nquit" | bluetoothctl' try: proc = await asyncio.create_subprocess_shell( cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=15.0) except FileNotFoundError: raise HTTPException(status_code=500, detail="`bluetoothctl` command not found. Is BlueZ installed and in PATH?") except asyncio.TimeoutError: raise HTTPException(status_code=504, detail="`bluetoothctl` command timed out after 15 seconds.") output = stdout.decode(errors="ignore") error = stderr.decode(errors="ignore") if "Failed to remove" in output or "not available" in output.lower(): raise HTTPException(status_code=502, detail=f"Unpairing failed. Output: {output}. Error: {error}") return {"ok": True, "message": "Unpair command sent. Check output for details.", "output": output, "error": error} @app.post( "/print/text", summary="Print a text label", status_code=200, ) async def print_text( text: Annotated[str, Form(description="Text to print on the label")], density: Annotated[int, Form(description="Print density: 0=light, 1=medium, 2=dark", ge=0, le=2)] = 2, paper: Annotated[str, Form(description="Paper type: gap, black, or continuous")] = "gap", copies: Annotated[int, Form(description="Number of copies", ge=1, le=99)] = 1, font_size: Annotated[int, Form(description="Font size in points", ge=6, le=200)] = 30, label_length: Annotated[int | None, Form(description="Label length in mm (overrides label_height)", ge=5, le=500)] = None, label_height: Annotated[int, Form(description="Label height in pixels", ge=40, le=4000)] = 240, address: Annotated[str | None, Form(description="BLE address (optional, overrides FICHERO_ADDR)")] = None, classic: Annotated[bool, Form(description="Use Classic Bluetooth RFCOMM")] = _DEFAULT_CLASSIC, channel: Annotated[int, Form(description="RFCOMM channel")] = _DEFAULT_CHANNEL, ): """Print a plain-text label. The text is rendered as a 96 px wide, 1-bit image and sent to the printer. """ paper_val = _parse_paper(paper) max_rows = (label_length * DOTS_PER_MM) if label_length is not None else label_height img = text_to_image(text, font_size=font_size, label_height=max_rows) try: async with connect(_address(address), classic=classic, channel=channel) as pc: ok = await do_print(pc, img, density=density, paper=paper_val, copies=copies, dither=False, max_rows=max_rows) except PrinterNotFound as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except PrinterTimeout as exc: raise HTTPException(status_code=504, detail=str(exc)) from exc except PrinterError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc if not ok: raise HTTPException(status_code=502, detail="Printer did not confirm completion.") return {"ok": True, "copies": copies, "text": text} @app.post( "/print/image", summary="Print an image", status_code=200, ) async def print_image( file: Annotated[UploadFile, File(description="Image file to print (PNG, JPEG, BMP, …)")], density: Annotated[int, Form(description="Print density: 0=light, 1=medium, 2=dark", ge=0, le=2)] = 2, paper: Annotated[str, Form(description="Paper type: gap, black, or continuous")] = "gap", copies: Annotated[int, Form(description="Number of copies", ge=1, le=99)] = 1, dither: Annotated[bool, Form(description="Apply Floyd-Steinberg dithering")] = True, label_length: Annotated[int | None, Form(description="Max label length in mm (overrides label_height)", ge=5, le=500)] = None, label_height: Annotated[int, Form(description="Max label height in pixels", ge=40, le=4000)] = 240, address: Annotated[str | None, Form(description="BLE address (optional, overrides FICHERO_ADDR)")] = None, classic: Annotated[bool, Form(description="Use Classic Bluetooth RFCOMM")] = _DEFAULT_CLASSIC, channel: Annotated[int, Form(description="RFCOMM channel")] = _DEFAULT_CHANNEL, ): """Print an image file. The image is resized to 96 px wide, optionally dithered to 1-bit, and sent to the printer. Supported formats: PNG, JPEG, BMP, GIF, TIFF, WEBP. """ # Validate content type loosely — Pillow will raise on unsupported data data = await file.read() if not data: raise HTTPException(status_code=422, detail="Uploaded file is empty.") try: img = Image.open(io.BytesIO(data)) img.load() # ensure the image is fully decoded except Exception as exc: raise HTTPException(status_code=422, detail=f"Cannot decode image: {exc}") from exc paper_val = _parse_paper(paper) max_rows = (label_length * DOTS_PER_MM) if label_length is not None else label_height try: async with connect(_address(address), classic=classic, channel=channel) as pc: ok = await do_print(pc, img, density=density, paper=paper_val, copies=copies, dither=dither, max_rows=max_rows) except PrinterNotFound as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except PrinterTimeout as exc: raise HTTPException(status_code=504, detail=str(exc)) from exc except PrinterError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc if not ok: raise HTTPException(status_code=502, detail="Printer did not confirm completion.") return {"ok": True, "copies": copies, "filename": file.filename} # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- def main() -> None: """Start the Fichero HTTP API server.""" global _DEFAULT_ADDRESS, _DEFAULT_CLASSIC, _DEFAULT_CHANNEL try: import uvicorn # noqa: PLC0415 except ImportError: print("ERROR: uvicorn is required to run the API server.") print("Install it with: pip install 'fichero-printer[api]'") raise SystemExit(1) from None parser = argparse.ArgumentParser(description="Fichero Printer API Server") parser.add_argument("--host", default="127.0.0.1", help="Bind host (default: 127.0.0.1)") parser.add_argument("--port", type=int, default=8765, help="Bind port (default: 8765)") parser.add_argument("--address", default=_DEFAULT_ADDRESS, metavar="BLE_ADDR", help="Default BLE address (or set FICHERO_ADDR env var)") parser.add_argument("--classic", action="store_true", default=_DEFAULT_CLASSIC, help="Default to Classic Bluetooth RFCOMM") parser.add_argument("--channel", type=int, default=_DEFAULT_CHANNEL, help="Default RFCOMM channel (default: 1)") parser.add_argument("--reload", action="store_true", help="Enable auto-reload (development)") args = parser.parse_args() # Push CLI overrides into module-level defaults so all handlers pick them up _DEFAULT_ADDRESS = args.address _DEFAULT_CLASSIC = args.classic _DEFAULT_CHANNEL = args.channel # Pass the app object directly when not reloading so that the module-level # globals (_DEFAULT_ADDRESS etc.) set above are visible to the handlers. # The string form "fichero.api:app" is required for --reload only, because # uvicorn's reloader needs to re-import the module in a worker process. uvicorn.run( "fichero.api:app" if args.reload else app, host=args.host, port=args.port, reload=args.reload, ) if __name__ == "__main__": main()