diff --git a/skills/brother-printer/SKILL.md b/skills/brother-printer/SKILL.md new file mode 100644 index 0000000..36f511d --- /dev/null +++ b/skills/brother-printer/SKILL.md @@ -0,0 +1,107 @@ +--- +name: brother-printer +description: Print PDF documents, text, and files directly to the Brother HL-L8240CDW duplex color laser printer at Grabengasse 7 (192.168.10.91). Handles interactive Telegram print options (Color/B&W, Pages per Sheet / N-Up, Duplex/Simplex) and live-updating Telegram status messages. +--- + +# 🖨️ Brother HL-L8240CDW Laser Printer + +Direct network printing to the **Brother HL-L8240CDW** duplex color laser printer on the home network (`192.168.10.91:631`) via IPP Everywhere. + +## 🎯 When to Use +- Whenever the user asks to print something: *"Bitte drucke XY"*, *"Druck das bitte aus"*, *"Kannst du das drucken?"*. +- Whenever the user sends a PDF, image, document, or ticket on Telegram with a request to print. +- Whenever the user asks to print recipes, shopping lists, flight tickets, packing lists, or notes. +- Whenever the user asks for printer status or toner levels: *"Wie viel Toner ist noch im Drucker?"*, *"Ist der Drucker online?"*. + +--- + +## 🕹️ Interactive Telegram Options Protocol (CRITICAL) + +When the user requests a print (e.g. *"Bitte drucke die Rechnung"* or sends a PDF), **DO NOT print blindly immediately**. +Provide an interactive decision prompt on Telegram using the `ask_user` tool with pre-configured defaults: + +### Fast-Path Preset Menu via `ask_user` + +Call `ask_user` with: +- `id`: `"print_options"` +- `header`: `"Drucker"` +- `question`: `"Wie möchtest du das Dokument drucken?"` +- `options`: + 1. `label`: `"🌱 Spar-Druck: S/W, 2 Seiten, Duplex (Recommended)"` + `description`: `"Schwarz-Weiß, 2 Seiten pro Blatt, beidseitig (spart Papier)"` + *(Maps to: `--color monochrome --nup 2 --duplex two-sided-long-edge`)* + 2. `label`: `"📄 Standard: S/W, 1 Seite, Duplex"` + `description`: `"Schwarz-Weiß, 1 Seite pro Blatt, beidseitig"` + *(Maps to: `--color monochrome --nup 1 --duplex two-sided-long-edge`)* + 3. `label`: `"🎨 Vollfarbe: 1 Seite, Duplex"` + `description`: `"Farbdruck, 1 Seite pro Blatt, beidseitig"` + *(Maps to: `--color color --nup 1 --duplex two-sided-long-edge`)* + 4. `label`: `"📄 Einseitig (Simplex): S/W, 1 Seite"` + `description`: `"Schwarz-Weiß, 1 Seite pro Blatt, nur Vorderseite"` + *(Maps to: `--color monochrome --nup 1 --duplex one-sided`)* + +> [!TIP] +> If the user specifically asks for custom parameters (e.g. *"Druck das in Farbe auf 2 Seiten"*), parse the options directly and run without extra prompt. + +--- + +## 🚀 Execution & Live Telegram Status Updates + +Execute the bundled driver script with `uv run`. + +Always pass `--chat-id ` so the script automatically creates and live-updates a single Telegram message (`⏳ Bereite vor...` ➔ `📡 Sende an Drucker...` ➔ `🖨️ Druckt...` ➔ `✅ Erfolgreich gedruckt!`): + +### 1. Print a PDF file with selected options +```bash +uv run /home/node/mop-skills/mop-skills/skills/brother-printer/scripts/brother_printer.py print \ + --file "/home/node/.openclaw/workspace/document.pdf" \ + --color monochrome \ + --nup 2 \ + --duplex two-sided-long-edge \ + --chat-id "306373425" +``` + +### 2. Print a document from a URL +```bash +uv run /home/node/mop-skills/mop-skills/skills/brother-printer/scripts/brother_printer.py print \ + --url "https://example.com/ticket.pdf" \ + --title "Konzertticket" \ + --color color \ + --nup 1 \ + --duplex two-sided-long-edge \ + --chat-id "306373425" +``` + +### 3. Print plain text or notes +```bash +uv run /home/node/mop-skills/mop-skills/skills/brother-printer/scripts/brother_printer.py print \ + --text "Einkaufsliste für Samstag:\n- Hafermilch\n- Kaffee\n- Bananen" \ + --title "Einkaufszettel" \ + --color monochrome \ + --nup 1 \ + --duplex one-sided \ + --chat-id "306373425" +``` + +### 4. Check status & toner levels +```bash +uv run /home/node/mop-skills/mop-skills/skills/brother-printer/scripts/brother_printer.py status +``` + +--- + +## ⚙️ CLI Options Reference + +| Argument | Values / Type | Default | Description | +|---|---|---|---| +| `--file` | File path | None | Local PDF path to print | +| `--url` | URL string | None | Download and print PDF from web | +| `--text` | String | None | Text content to format as A4 PDF and print | +| `--title` | String | Basename | Document title for print queue and Telegram header | +| `--color` | `monochrome`, `color` | `monochrome` | Print color mode | +| `--nup` | `1`, `2`, `4` | `1` | Hardware pages per sheet (1 = normal, 2 = 2 pages/sheet) | +| `--duplex` | `two-sided-long-edge`, `one-sided`, `two-sided-short-edge` | `two-sided-long-edge` | Duplex binding mode | +| `--copies` | Integer | `1` | Number of printed copies | +| `--chat-id` | Telegram Chat ID | None | Target chat ID for live status updates | +| `--max-pages`| Integer | `30` | Safety limit to prevent runaway printing | +| `--force` | Flag | `false` | Bypass page limit | diff --git a/skills/brother-printer/scripts/brother_printer.py b/skills/brother-printer/scripts/brother_printer.py new file mode 100755 index 0000000..571ba22 --- /dev/null +++ b/skills/brother-printer/scripts/brother_printer.py @@ -0,0 +1,475 @@ +#!/usr/bin/env python3 +# /// script +# dependencies = [ +# "pyipp>=0.16.0", +# "requests>=2.31.0", +# "pypdf>=4.0.0", +# "reportlab>=4.0.0", +# ] +# /// +""" +Brother HL-L8240CDW Network Printer Controller & IPP Driver for OpenClaw. +Supports direct PDF printing, hardware N-Up, duplex, color modes, +and live Telegram status updates via Bot API. +""" + +import os +import sys +import time +import json +import argparse +import asyncio +from pathlib import Path +from typing import Optional, Dict, Any + +import requests +from pypdf import PdfReader +from reportlab.lib.pagesizes import A4 +from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle +from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer +from reportlab.lib.units import cm + +import pyipp +from pyipp import IPP +from pyipp.enums import IppOperation +from pyipp.tags import IppTag +import pyipp.serializer + +# Register extra IPP attribute tags for Brother HL-L8240CDW +pyipp.serializer.ATTRIBUTE_TAG_MAP["print-color-mode"] = IppTag.KEYWORD +pyipp.serializer.ATTRIBUTE_TAG_MAP["page-ranges"] = IppTag.RANGE + +DEFAULT_PRINTER_HOST = os.environ.get("BROTHER_PRINTER_HOST", "192.168.10.91") +DEFAULT_PRINTER_PORT = int(os.environ.get("BROTHER_PRINTER_PORT", "631")) +DEFAULT_PRINTER_URI = f"ipp://{DEFAULT_PRINTER_HOST}:{DEFAULT_PRINTER_PORT}/ipp/print" +TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN") + + +# --------------------------------------------------------------------------- +# Telegram Helper +# --------------------------------------------------------------------------- + +class TelegramNotifier: + def __init__(self, token: Optional[str] = None, chat_id: Optional[str] = None, message_id: Optional[int] = None): + self.token = token or TELEGRAM_BOT_TOKEN + self.chat_id = chat_id + self.message_id = int(message_id) if message_id else None + + def can_notify(self) -> bool: + return bool(self.token and self.chat_id) + + def send_initial(self, text: str) -> Optional[int]: + if not self.can_notify(): + return None + url = f"https://api.telegram.org/bot{self.token}/sendMessage" + payload = { + "chat_id": self.chat_id, + "text": text, + "parse_mode": "Markdown", + } + try: + resp = requests.post(url, json=payload, timeout=8) + if resp.ok: + data = resp.json() + self.message_id = data.get("result", {}).get("message_id") + return self.message_id + except Exception as e: + sys.stderr.write(f"Telegram send error: {e}\n") + return None + + def update(self, text: str): + if not self.can_notify() or not self.message_id: + return + url = f"https://api.telegram.org/bot{self.token}/editMessageText" + payload = { + "chat_id": self.chat_id, + "message_id": self.message_id, + "text": text, + "parse_mode": "Markdown", + } + try: + requests.post(url, json=payload, timeout=8) + except Exception as e: + sys.stderr.write(f"Telegram edit error: {e}\n") + + +# --------------------------------------------------------------------------- +# Text to PDF Converter +# --------------------------------------------------------------------------- + +def convert_text_to_pdf(text: str, output_path: str, title: str = "Dokument") -> str: + """Convert plain text or simple markdown into a clean A4 PDF.""" + doc = SimpleDocTemplate( + output_path, + pagesize=A4, + rightMargin=2 * cm, + leftMargin=2 * cm, + topMargin=2 * cm, + bottomMargin=2 * cm, + ) + styles = getSampleStyleSheet() + normal_style = styles["Normal"] + normal_style.fontSize = 11 + normal_style.leading = 14 + + heading_style = styles["Heading1"] + heading_style.fontSize = 16 + heading_style.leading = 20 + + story = [ + Paragraph(title, heading_style), + Spacer(1, 0.5 * cm), + ] + + for paragraph in text.split("\n\n"): + clean_p = paragraph.replace("\n", "
") + story.append(Paragraph(clean_p, normal_style)) + story.append(Spacer(1, 0.3 * cm)) + + doc.build(story) + return output_path + + +# --------------------------------------------------------------------------- +# Printer Service +# --------------------------------------------------------------------------- + +class BrotherPrinter: + def __init__(self, host: str = DEFAULT_PRINTER_HOST, port: int = DEFAULT_PRINTER_PORT): + self.host = host + self.port = port + self.uri = f"ipp://{self.host}:{self.port}/ipp/print" + + async def get_status(self) -> Dict[str, Any]: + """Fetch status, markers, and printer details via IPP.""" + async with IPP(self.host, port=self.port) as ipp: + printer = await ipp.printer() + markers = {} + for marker in printer.markers: + markers[marker.name] = { + "level": marker.level, + "type": marker.marker_type, + "color": marker.color, + } + return { + "name": printer.info.name, + "model": printer.info.model, + "state": printer.state.printer_state, + "reasons": printer.state.reasons, + "markers": markers, + "uri": self.uri, + } + + async def validate_job( + self, + color_mode: str = "monochrome", + nup: int = 1, + duplex: str = "two-sided-long-edge", + copies: int = 1, + ) -> bool: + """Dry-run validation of job parameters without printing.""" + async with IPP(self.host, port=self.port) as ipp: + msg = { + "version": (2, 0), + "operation": IppOperation.VALIDATE_JOB, + "operation-attributes-tag": { + "printer-uri": self.uri, + "requesting-user-name": "openclaw", + "document-format": "application/pdf", + }, + "job-attributes-tag": { + "sides": duplex, + "number-up": nup, + "print-color-mode": color_mode, + "copies": copies, + }, + } + res = await ipp.execute(IppOperation.VALIDATE_JOB, msg) + return res.get("status-code") == 0 + + async def send_print_job( + self, + pdf_bytes: bytes, + job_name: str = "OpenClaw Print", + color_mode: str = "monochrome", + nup: int = 1, + duplex: str = "two-sided-long-edge", + copies: int = 1, + ) -> Dict[str, Any]: + """Send a Print-Job request directly with raw PDF data.""" + async with IPP(self.host, port=self.port) as ipp: + msg = { + "version": (2, 0), + "operation": IppOperation.PRINT_JOB, + "operation-attributes-tag": { + "printer-uri": self.uri, + "requesting-user-name": "openclaw", + "job-name": job_name, + "document-format": "application/pdf", + }, + "job-attributes-tag": { + "sides": duplex, + "number-up": nup, + "print-color-mode": color_mode, + "copies": copies, + }, + "data": pdf_bytes, + } + res = await ipp.execute(IppOperation.PRINT_JOB, msg) + return res + + async def get_job_attributes(self, job_id: int) -> Dict[str, Any]: + """Poll job status using GET_JOB_ATTRIBUTES.""" + async with IPP(self.host, port=self.port) as ipp: + msg = { + "version": (2, 0), + "operation": IppOperation.GET_JOB_ATTRIBUTES, + "operation-attributes-tag": { + "printer-uri": self.uri, + "job-id": job_id, + "requesting-user-name": "openclaw", + }, + } + try: + res = await ipp.execute(IppOperation.GET_JOB_ATTRIBUTES, msg) + jobs = res.get("jobs", []) + if jobs: + return jobs[0] + except Exception as e: + sys.stderr.write(f"Job query warning: {e}\n") + return {} + + +# --------------------------------------------------------------------------- +# CLI Command Handlers +# --------------------------------------------------------------------------- + +def handle_status(args): + printer = BrotherPrinter(host=args.host, port=args.port) + data = asyncio.run(printer.get_status()) + if args.json: + print(json.dumps(data, indent=2)) + return + + print("🖨️ Brother HL-L8240CDW Status") + print("=" * 35) + print(f"Modell: {data['model']}") + print(f"Status: {data['state']}") + print(f"URI: {data['uri']}") + print("\n📊 Tonerstände:") + for name, m in data["markers"].items(): + color_label = {"BK": "Schwarz", "C": "Cyan", "M": "Magenta", "Y": "Gelb"}.get(name, name) + bar = "█" * (m["level"] // 10) + "░" * (10 - (m["level"] // 10)) + print(f" [{name:2}] {color_label:8}: {bar} {m['level']}%") + + +def handle_validate(args): + printer = BrotherPrinter(host=args.host, port=args.port) + ok = asyncio.run( + printer.validate_job( + color_mode=args.color, + nup=args.nup, + duplex=args.duplex, + copies=args.copies, + ) + ) + if ok: + print("✅ Konfiguration ist gültig und wird vom Drucker unterstützt.") + else: + print("❌ Konfiguration wurde vom Drucker abgewiesen.") + sys.exit(1) + + +def handle_print(args): + notifier = TelegramNotifier(chat_id=args.chat_id, message_id=args.message_id) + + # 1. Resolve source document + doc_title = args.title or "Dokument" + pdf_path = None + temp_files = [] + + if args.file: + p = Path(args.file) + if not p.exists(): + err = f"❌ Datei nicht gefunden: {args.file}" + print(err) + notifier.update(err) + sys.exit(1) + pdf_path = str(p) + doc_title = args.title or p.name + elif args.url: + notifier.update(f"⏳ [1/4] Lade Dokument herunter: `{args.url[:40]}...`") + resp = requests.get(args.url, timeout=30) + tmp = f"/tmp/print_{int(time.time())}.pdf" + with open(tmp, "wb") as f: + f.write(resp.content) + temp_files.append(tmp) + pdf_path = tmp + doc_title = args.title or Path(args.url).name + elif args.text: + notifier.update("⏳ [1/4] Formatiere Text als Druck-PDF...") + tmp = f"/tmp/print_{int(time.time())}.pdf" + convert_text_to_pdf(args.text, tmp, title=doc_title) + temp_files.append(tmp) + pdf_path = tmp + else: + err = "❌ Kein Dokument angegeben (--file, --url oder --text erforderlich)." + print(err) + notifier.update(err) + sys.exit(1) + + # 2. Inspect PDF properties + try: + reader = PdfReader(pdf_path) + total_pages = len(reader.pages) + except Exception as e: + total_pages = 1 + + # Safety guardrail: check max pages + if total_pages > args.max_pages and not args.force: + err = f"⚠️ *Druck abgebrochen*: Dokument hat {total_pages} Seiten (Limit: {args.max_pages}). Verwende `--force` zum Überschreiben." + print(err) + notifier.update(err) + sys.exit(1) + + # Calculate estimated sheets + sheets = (total_pages + (args.nup - 1)) // args.nup + if args.duplex != "one-sided": + sheets = (sheets + 1) // 2 + + color_desc = "Farbe 🎨" if args.color == "color" else "Schwarz-Weiß 🖤" + duplex_desc = "Duplex (beidseitig) 🔄" if args.duplex != "one-sided" else "Simplex (einseitig) 📄" + nup_desc = f"{args.nup} Seite(n) pro Blatt" + + # Initial status notification + status_header = ( + f"🖨️ *Druckauftrag: {doc_title}*\n" + f"• *Umfang*: {total_pages} Seite(n) (~{sheets} Blatt Papier)\n" + f"• *Modus*: {color_desc} | {nup_desc} | {duplex_desc}\n" + ) + + if not notifier.message_id and notifier.can_notify(): + notifier.send_initial(f"{status_header}• *Status*: ⏳ [1/3] Bereite Übertragung vor...") + else: + notifier.update(f"{status_header}• *Status*: ⏳ [1/3] Bereite Übertragung vor...") + + # 3. Read PDF bytes + with open(pdf_path, "rb") as f: + pdf_data = f.read() + + # 4. Connect to printer and send job + notifier.update(f"{status_header}• *Status*: 📡 [2/3] Sende an Brother HL-L8240CDW...") + printer = BrotherPrinter(host=args.host, port=args.port) + + try: + res = asyncio.run( + printer.send_print_job( + pdf_bytes=pdf_data, + job_name=doc_title, + color_mode=args.color, + nup=args.nup, + duplex=args.duplex, + copies=args.copies, + ) + ) + except Exception as e: + err_msg = f"{status_header}• *Status*: ❌ Verbindungsfehler zum Drucker: `{e}`" + print(err_msg) + notifier.update(err_msg) + sys.exit(1) + + # Check IPP status code + status_code = res.get("status-code", -1) + if status_code != 0: + err_msg = f"{status_header}• *Status*: ❌ Drucker meldet Fehler-Code `{status_code}`." + print(err_msg) + notifier.update(err_msg) + sys.exit(1) + + # Extract Job ID + jobs = res.get("jobs", []) + job_id = jobs[0].get("job-id", "unbekannt") if jobs else "unbekannt" + print(f"Print job queued. Job ID: {job_id}") + + notifier.update(f"{status_header}• *Status*: 🖨️ [3/3] Auftrag übertragen (Job #{job_id}). Druckvorgang läuft...") + + # 5. Monitor Job Progress (up to 30 seconds) + if isinstance(job_id, int): + for _ in range(15): + time.sleep(2) + try: + attr = asyncio.run(printer.get_job_attributes(job_id)) + state = attr.get("job-state") + # 9 = completed, 7 = canceled, 8 = aborted + if state == 9: + break + elif state in (7, 8): + notifier.update(f"{status_header}• *Status*: ⚠️ Auftrag #{job_id} wurde am Drucker abgebrochen.") + return + except Exception: + pass + + # Final success update + final_msg = ( + f"{status_header}" + f"• *Status*: ✅ **Erfolgreich gedruckt!**\n" + f"• *Job-ID*: `#{job_id}`\n" + f"• *Ausgabe*: Liegt im Ausgabefach bereit." + ) + print("Druck erfolgreich abgeschlossen.") + notifier.update(final_msg) + + # Cleanup temp files + for t in temp_files: + try: + os.remove(t) + except Exception: + pass + + +# --------------------------------------------------------------------------- +# Main Entry Point +# --------------------------------------------------------------------------- + +def main(): + parser = argparse.ArgumentParser(description="Brother HL-L8240CDW IPP Print Utility for OpenClaw") + parser.add_argument("--host", default=DEFAULT_PRINTER_HOST, help="Printer IP or Hostname") + parser.add_argument("--port", type=int, default=DEFAULT_PRINTER_PORT, help="IPP Port (default 631)") + + subparsers = parser.add_subparsers(dest="command", required=True) + + # Subcommand: status + status_parser = subparsers.add_parser("status", help="Get printer status and toner levels") + status_parser.add_argument("--json", action="store_true", help="Output as JSON") + status_parser.set_defaults(func=handle_status) + + # Subcommand: validate + val_parser = subparsers.add_parser("validate", help="Dry-run validate print parameters") + val_parser.add_argument("--color", choices=["monochrome", "color"], default="monochrome") + val_parser.add_argument("--nup", type=int, choices=[1, 2, 4], default=1) + val_parser.add_argument("--duplex", choices=["two-sided-long-edge", "two-sided-short-edge", "one-sided"], default="two-sided-long-edge") + val_parser.add_argument("--copies", type=int, default=1) + val_parser.set_defaults(func=handle_validate) + + # Subcommand: print + print_parser = subparsers.add_parser("print", help="Print a document") + print_parser.add_argument("--file", help="Path to PDF file") + print_parser.add_argument("--url", help="URL of PDF document") + print_parser.add_argument("--text", help="Raw text to convert and print") + print_parser.add_argument("--title", help="Job or document title") + print_parser.add_argument("--color", choices=["monochrome", "color"], default="monochrome", help="Color mode (default: monochrome)") + print_parser.add_argument("--nup", type=int, choices=[1, 2, 4], default=1, help="Pages per sheet (1, 2 or 4)") + print_parser.add_argument("--duplex", choices=["two-sided-long-edge", "two-sided-short-edge", "one-sided"], default="two-sided-long-edge", help="Duplex mode (default: two-sided-long-edge)") + print_parser.add_argument("--copies", type=int, default=1, help="Number of copies") + print_parser.add_argument("--max-pages", type=int, default=30, help="Safety page limit (default 30)") + print_parser.add_argument("--force", action="store_true", help="Bypass max pages guardrail") + print_parser.add_argument("--chat-id", help="Telegram Chat ID for live updates") + print_parser.add_argument("--message-id", help="Telegram Message ID to edit (optional, created if omitted)") + print_parser.set_defaults(func=handle_print) + + args = parser.parse_args() + args.func(args) + + +if __name__ == "__main__": + main()