Scripts to aid in power management of CUPS printers

_README.md

CUPS Printer Power Helper Scripts

There are two scripts here which can aid in managing the physical power state of a CUPS printer in Home Assistant.

See here for usage: https://github.com/tigattack/docker-cups-canon-airprint

printer_idle.py

This script POST's the idle status of your printer(s) to the defined webhook URL. It is designed to be run by a scheduler (such as cron or a basic loop & sleep in a bash script).

This POSTed data can be used by e.g. Home Assistant to determine when the printer is idle and power off a smart plug.

Here's a sample of the webhook body: {"printer": "MyPrinter", "idle": true, "idle_time": 43680, "last_job_time": 1742350946}

Variables:

• PRINTER_IDLE_PRINTERS (default: null): Comma-seperated list of printer names as set in CUPS. Only required if multiple printers are available.
• PRINTER_IDLE_THRESHOLD (default: 3600): Seconds since last job to consider printer idle.
• PRINTER_IDLE_WEBHOOK_URL (default: null): Webhook URL to send idle printer information to.
• PRINTER_IDLE_ALWAYS_SEND (default: false): Whether to always send the webhook even if the state hasn't changed.
• PRINTER_IDLE_LOGLEVEL (default: INFO): Can be any valid Python logging level. Likely DEBUG is the only other useful option here, though.

printer_power_on.py

This script POST's to a webhook URL to request the printer be powered on (e.g. by Home Assistant).

This runs as a print job pre-hook via tea4cups and will wait (up to the configured timeout) for the printer to become available before allowing the job to continue by means of attempting a connection to port 631 (IPP) on the printer.

If the printer does not become available within the wait timeout, the script will give up waiting and the job will be cancelled.

Here's a sample of the webhook body: {"power_on": "MyPrinter"}

Variables:

• PRINTER_POWERON_NAME (default: null): Printer name, as set in CUPS, for which power on requests should be sent.
• PRINTER_POWERON_HOST (default: null): Printer IP or hostname. This is only used as a fallback when it can't be discovered from the tea4cups environment, where the script assumes it will find an IPP or HTTP printer. The printer host is used to check for printer availability.
• PRINTER_POWERON_WAIT_TIMEOUT (default: 120): How long to wait for the printer to become available.
• PRINTER_POWERON_WEBHOOK_URL (default: null): Webhook URL to send printer power on request to.
• PRINTER_POWERON_LOGLEVEL (default: INFO): Can be any valid Python logging level. Likely DEBUG is the only other useful option here, though.
• TEA4CUPS_DEBUG (default: no): tea4cups will produce debug messages via CUPS' error logs if this variable is yes. Must be yes or no.

Troubleshooting:

1. Set CUPS_LOGLEVEL=debug
2. Set TEA4CUPS_DEBUG=yes
3. Set PRINTER_POWERON_LOGLEVEL=debug
4. Look for log output between the following two lines in the CUPS error stream:
   1. Begin forked Prehooks
   2. End forked Prehooks
5. Good luck.

printer_idle.py

#!/usr/bin/python3
# https://github.com/tigattack/docker-cups-canon-airprint
# https://gist.github.com/tigattack/3d54b842d6c842b6fa40618ff6279a1a

import http.client
import json
import logging
import os
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

import cups

log_level = os.getenv("PRINTER_IDLE_LOGLEVEL", "INFO")
try:
    log_level = getattr(logging, log_level.upper())
except AttributeError:
    print(f"Invalid log level: {log_level} - Defaulting to INFO")
    log_level = logging.INFO

logging.basicConfig(level=log_level)
log = logging.getLogger("printer_idle")


def strtobool(value: str) -> bool:
    value = value.lower()
    if value in ("y", "yes", "on", "1", "true", "t"):
        return True
    return False


class PrinterIdle:
    def __init__(self, printer_name: str, idle_threshold: int):
        self.conn = cups.Connection()
        self.printer_name = printer_name
        self.idle_threshold = idle_threshold

        if printer_name == "":
            printers = self.get_printers().keys()
            if len(printers) == 0:
                raise ValueError("Printer name not defined and no printers found.")
            if len(printers) > 1:
                raise ValueError(
                    f"Printer name not defined and multiple printers were found: {list(printers)}"
                )
            self.printer_name = list(printers)[0]

    def get_printers(self) -> dict[str, Any]:
        printers: dict[str, Any] = self.conn.getPrinters()
        return printers

    def check_printer(self):
        printers = self.get_printers()
        if self.printer_name not in printers:
            raise ValueError(f"Printer {self.printer_name} not found")
        return True

    def get_last_job_time(self):
        jobs: dict[int, Any] = self.conn.getJobs(which_jobs="completed")
        for jid, _ in jobs.items():
            job_attrs = self.conn.getJobAttributes(jid)

            printer_uri: str = job_attrs.get("job-printer-uri")
            job_time: str = job_attrs.get("time-at-completed")

            if printer_uri.endswith(self.printer_name) and job_time is not None:
                return datetime.fromtimestamp(job_time)
        return None

    def check_idle(self):
        if self.last_job_time is None:
            log.debug(
                "Idle time undefined for printer %s. Printer must be idle.",
                self.printer_name,
            )
            return True
        idle_time = datetime.now() - self.last_job_time
        return idle_time > timedelta(seconds=self.idle_threshold)

    @property
    def last_job_time(self):
        return self.get_last_job_time()

    @property
    def is_idle(self):
        return self.check_idle()


def send_webhook(
    webhook_url: str,
    printer_name: str,
    is_idle: bool,
    idle_time: int,
    last_job_time: datetime | None,
):
    parsed_url = urlparse(webhook_url)
    webhook_scheme = parsed_url.scheme
    webhook_host = parsed_url.hostname
    webhook_port = parsed_url.port or (443 if parsed_url.scheme == "https" else 80)

    if webhook_scheme == "https":
        conn = http.client.HTTPSConnection(webhook_host, webhook_port)
    else:
        conn = http.client.HTTPConnection(webhook_host, webhook_port)

    last_job_timestamp = 0 if last_job_time is None else int(last_job_time.timestamp())
    webhook_body = json.dumps(
        {
            "printer": printer_name,
            "idle": is_idle,
            "idle_time": idle_time,
            "last_job_time": last_job_timestamp,
            "source": "cups",
        }
    )
    log.debug("Sending info to webhook: %s", webhook_body)
    try:
        conn.request(
            "POST",
            webhook_url,
            webhook_body,
            {"Content-Type": "application/json"},
        )
        response = conn.getresponse()
        log.debug("Webhook responded %s %s", response.status, response.reason)
    except Exception as e:
        log.error("Error sending webhook: %s", e)
        response = None
    finally:
        conn.close()
    return response


def main():
    # Comma-seperated list of printer names as set in CUPS. Only required if multiple printers are available.
    printers = os.getenv("PRINTER_IDLE_PRINTERS", "")
    # Seconds since last job to consider printer idle.
    idle_threshold = os.getenv("PRINTER_IDLE_THRESHOLD", 3600)
    # Webhook URL to send idle printer information to.
    webhook_url = os.getenv("PRINTER_IDLE_WEBHOOK_URL")
    # Whether to always send the webhook even if the state hasn't changed
    always_post_state = strtobool(os.getenv("PRINTER_IDLE_ALWAYS_SEND", "false"))

    printers = printers.split(",")

    for printer_name in printers:
        try:
            printer = PrinterIdle(printer_name, int(idle_threshold))
        except ValueError as exc:
            log.error("An error occured setting up the idle check: %s", exc)
            continue
        except RuntimeError:
            log.warning("Failed to connect to CUPS. The service may not be running.")
            continue

        try:
            printer.check_printer()
        except ValueError as exc:
            log.error("Skipping printer: %s", exc)
            continue

        if not printer_name:
            printer_name = printer.printer_name

        state_path = Path(f"/run/printer_idle_{printer_name.lower()}.state")
        last_state = state_path.read_text().strip() if state_path.exists() else None
        state_path.write_text("idle" if printer.is_idle else "active")

        idle_time = 0
        last_job_time = printer.last_job_time
        if last_job_time is None:
            idle_time_human = "Unknown (no jobs found, must be idle)"
        else:
            idle_time = datetime.now() - last_job_time
            idle_time_human = f"{idle_time.days}d {idle_time.seconds // 3600}h {idle_time.seconds % 3600 // 60}m"

        if printer.is_idle:
            if last_state != "idle":
                log.info(f"Printer {printer_name} has changed to idle state.")
            log.debug(f"Printer {printer_name} has been idle for {idle_time_human}.")
        else:
            idle_time = 0
            if last_state != "active":
                log.info(f"Printer {printer_name} has changed to active state.")
            log.debug(
                f"Printer {printer_name} is not idle. Last job completed {idle_time_human} ago."
            )

        state_changed = (printer.is_idle and last_state == "idle") or (
            not printer.is_idle and last_state == "active"
        )

        if not state_changed and not always_post_state:
            log.debug(
                "Skipping webhook - Printer state has not changed from %s", last_state
            )
        else:
            if webhook_url is None:
                log.warning("Skipping webhook - PRINTER_IDLE_WEBHOOK_URL unset.")
                return

            if not state_changed and always_post_state:
                log.debug(
                    "State has not changed from %s, but PRINTER_IDLE_ALWAYS_SEND is true, so sending anyway.",
                    last_state,
                )

            idle_seconds = (
                int(idle_time.total_seconds())
                if isinstance(idle_time, timedelta)
                else 0
            )
            log.debug("Sending webhook for idle printer")
            webhook_response = send_webhook(
                webhook_url,
                printer_name,
                printer.is_idle,
                idle_seconds,
                last_job_time,
            )
            if webhook_response is None:
                log.error("Webhook request failed.")
                return
            elif webhook_response.status != 200:
                log.error("Webhook responded with status %s", webhook_response.status)
                return
            log.debug("Webhook sent successfully")


if __name__ == "__main__":
    main()

printer_power_on.py

#!/usr/bin/python3
# https://github.com/tigattack/docker-cups-canon-airprint
# https://gist.github.com/tigattack/3d54b842d6c842b6fa40618ff6279a1a

import http.client
import json
import logging
import os
import socket
import sys
import time
from pathlib import Path
from urllib.parse import urlparse

WEBHOOK_URL_SECRETS_FILE = Path("/run/secrets/tea4cups-poweron-webhook-url")
PRINTER_HOST_SECRETS_FILE = Path("/run/secrets/tea4cups-poweron-host")

log_level = os.getenv("PRINTER_POWERON_LOGLEVEL", "INFO")
try:
    log_level = getattr(logging, log_level.upper())
except AttributeError:
    print(f"Invalid log level: {log_level} - Defaulting to INFO")
    log_level = logging.INFO

logging.basicConfig(level=log_level)
log = logging.getLogger("printer_webhook_monitor")


def send_webhook(webhook_url: str, printer_name: str):
    parsed_url = urlparse(webhook_url)
    webhook_scheme = parsed_url.scheme
    webhook_host = parsed_url.hostname
    webhook_port = parsed_url.port or (443 if parsed_url.scheme == "https" else 80)

    webhook_path = parsed_url.path

    if webhook_scheme == "https":
        conn = http.client.HTTPSConnection(webhook_host, webhook_port)
    else:
        conn = http.client.HTTPConnection(webhook_host, webhook_port)

    try:
        log.info("Triggering webhook...")
        conn.request(
            "POST",
            webhook_path,
            json.dumps(
                {
                    "power_on": printer_name,
                    "source": "cups",
                }
            ),
            {"Content-Type": "application/json"},
        )
        response = conn.getresponse()
        log.debug("Webhook responded: %s %s", response.status, response.reason)
    except Exception as e:
        log.error("Error sending webhook: %s", e)
    finally:
        conn.close()


def is_printer_available(printer_host: str, port: int = 631) -> bool:
    try:
        with socket.create_connection((printer_host, port), timeout=5):
            return True
    except (socket.timeout, ConnectionRefusedError):
        return False


def wait_for_printer(
    printer_host: str, printer_name: str, port: int = 631, timeout: int = 60
):
    log.info("Waiting for %s to become available for printing...", printer_name)
    start_time = time.time()
    while time.time() - start_time < timeout:
        if is_printer_available(printer_host, port):
            log.info("%s is online", printer_name)
            return True
        log.debug("Printer not available yet, retrying...")
        time.sleep(1)
    log.error(
        "Timeout reached: %s did not become available within %d seconds",
        printer_name,
        timeout,
    )
    return False


def get_printer_host():
    """
    Attempt to read & parse the printer URI from the tea4cups environment.
    If that fails, read the printer host from the secrets file.
    """
    printer_uri = os.getenv("DEVICE_URI")
    if not printer_uri:
        log.warning("Failed to discover printer URI from tea4cups environment.")
        return None

    printer_host = urlparse(printer_uri).hostname
    if printer_host:
        return printer_host
    log.warning(
        "Failed to parse hostname from printer URI: %s. Attempting to read from environment via secrets file",
        printer_uri,
    )

    try:
        printer_host = PRINTER_HOST_SECRETS_FILE.read_text().strip()
    except FileNotFoundError:
        log.error("Secrets file /run/secrets/tea4cups-poweron-host not found.")
        return None

    if not printer_host:
        log.error("The printer host secrets file seems to be empty.")
        return None
    if printer_host == "undef":
        log.error("PRINTER_POWERON_HOST environment variable is not set.")
        return None

    return printer_host


def main():
    printer_name = os.getenv("TEAPRINTERNAME")
    wait_timeout = int(os.getenv("PRINTER_POWERON_WAIT_TIMEOUT", 120))

    try:
        webhook_url = WEBHOOK_URL_SECRETS_FILE.read_text().strip()
        if not webhook_url:
            log.error("The webhook URL secrets file seems to be empty.")
            sys.exit(-1)
        if webhook_url == "undef":
            log.error("PRINTER_POWERON_WEBHOOK_URL environment variable is not set.")
            sys.exit(-1)
    except FileNotFoundError:
        log.error("Secrets file /run/secrets/tea4cups-poweron-webhook-url not found.")
        sys.exit(-1)

    if not printer_name:
        log.error("Failed to get printer name from tea4cups environment.")
        sys.exit(-1)

    printer_uri = get_printer_host()
    if not printer_uri:
        log.error("Failed to get printer host.")
        sys.exit(-1)

    if is_printer_available(printer_uri):
        log.info("Printer %s is already online", printer_name)
        return

    send_webhook(webhook_url, printer_name)
    printer_available = wait_for_printer(
        printer_uri, printer_name, timeout=wait_timeout
    )

    if not printer_available:
        sys.exit(-1)


if __name__ == "__main__":
    main()

tea4cups.conf

[global]
directory : /var/spool/cups/

[PRINTER_NAME_PLACEHOLDER]
prehook_poweron_printer : /opt/power_scripts/printer_power_on.py