Add automatic printer capability detection and autocomplete

CHANGELOG.md

@@ -11,6 +11,7 @@ See also [Creating new releases] for instructions on how to create a new release
   - `GUTENBERG_TRUST_X_FORWARDED_HOST`, `GUTENBERG_TRUST_X_FORWARDED_PROTO` and `GUTENBERG_TRUST_X_REAL_IP` [#175]
   - `GUTENBERG_TRUSTED_PROXY_IPS` [#190]
 - Docker Images are now published in the GitHub Container Registry [#199]
+- Added automatic CUPS printer capability configuration when selecting a printer in the Django admin panel [#200]
 
 ### Changed
 - Modified nginx Docker image config to correctly pass the `X-Forwarded-Host` header [#175]
@@ -75,7 +76,8 @@ The previous significant commit was made on [2022-08-26](https://github.com/KSIU
 [#178]: https://github.com/KSIUJ/gutenberg/pull/178
 [#190]: https://github.com/KSIUJ/gutenberg/pull/190
 [#194]: https://github.com/KSIUJ/gutenberg/pull/194
-- [#199]: https://github.com/KSIUJ/gutenberg/pull/199
+[#199]: https://github.com/KSIUJ/gutenberg/pull/199
+[#200]: https://github.com/KSIUJ/gutenberg/pull/200
 
 [keep a changelog]: https://keepachangelog.com/en/1.1.0/
 [OpenID Connect chapter]: https://ksiuj.github.io/gutenberg/admin/openid-connect.html

backend/control/admin.py

@@ -1,9 +1,17 @@
+import logging
+
+from celery.app.control import flatten_reply
 from django.contrib import admin
+from django.core.exceptions import PermissionDenied
+from django.http import HttpResponseNotAllowed, JsonResponse
+from django.urls import path
 
 from control.forms import LocalPrinterParamsForm
-# Register your models here.
 from control.models import GutenbergJob, PrintingProperties, PrinterPermissions, LocalPrinterParams, Printer, \
     JobArtefact
+from gutenberg.celery import app
+
+logger = logging.getLogger('gutenberg.control')
 
 
 class PrintingPropertiesInline(admin.TabularInline):
@@ -35,6 +43,54 @@ class PrinterAdmin(admin.ModelAdmin):
     list_display = ('name', 'display_order')
     ordering = ('display_order', 'name')
 
+    def get_urls(self):
+        urls = super().get_urls()
+        custom_urls = [
+            path(
+                'cups-printer-options/',
+                self.admin_site.admin_view(self.cups_printer_options_view),
+                name='control_printer_cups_printer_options',
+            ),
+        ]
+        return custom_urls + urls
+
+    def cups_printer_options_view(self, request):
+        """Return Gutenberg's supported CUPS settings for a selected queue."""
+        if request.method != 'GET':
+            return HttpResponseNotAllowed(['GET'])
+
+        if not (self.has_add_permission(request) or self.has_change_permission(request)):
+            raise PermissionDenied
+
+        cups_printer_name = request.GET.get('name', '').strip()
+        if not cups_printer_name:
+            return JsonResponse({'error': 'The "name" query parameter is required.'}, status=400)
+        if len(cups_printer_name) > 128:
+            return JsonResponse({'error': 'The printer name is too long.'}, status=400)
+
+        try:
+            replies = app.control.broadcast(
+                'gutenberg_get_cups_printer_options',
+                arguments={'cups_printer_name': cups_printer_name},
+                reply=True,
+                limit=1,
+                timeout=5,
+            )
+            replies = [
+                reply for reply in flatten_reply(replies).values()
+                if isinstance(reply, dict) and 'error' not in reply
+            ]
+        except Exception:
+            logger.exception('Failed to get CUPS printer options from workers')
+            return JsonResponse({'error': 'Could not contact a printing worker.'}, status=503)
+
+        if not replies:
+            return JsonResponse({'error': 'No printing worker returned printer capabilities.'}, status=503)
+        if not replies[0]:
+            return JsonResponse({'error': 'Could not discover capabilities for this printer.'}, status=502)
+
+        return JsonResponse(replies[0])
+
 
 admin.site.register(Printer, PrinterAdmin)
 admin.site.register(GutenbergJob, GutenbergJobAdmin)

backend/control/static/css/cups_printer_name_autocomplete.css

@@ -19,3 +19,7 @@
 .cups-printer-list-autocomplete li .autocomplete-button:hover, .cups-printer-list-autocomplete li .autocomplete-button:focus {
     color: var(--link-hover-color);
 }
+
+.cups-printer-options-status.error {
+    color: var(--error-fg);
+}

backend/control/static/js/cups_printer_name_autocomplete.js

@@ -2,11 +2,74 @@ document.addEventListener("DOMContentLoaded", () => {
 
   document.querySelectorAll(".cups-printer-list-autocomplete").forEach(wrapper => {
     const input = wrapper.querySelector("input");
+    const status = wrapper.querySelector(".cups-printer-options-status");
+    let latestRequest = 0;
+
+    const setFieldValue = (id, value) => {
+      const field = document.getElementById(id);
+      if (!field || value === undefined) return;
+      field.value = value ?? "";
+      field.dispatchEvent(new Event("change", { bubbles: true }));
+    };
+
+    const setCheckboxValue = (id, value) => {
+      const field = document.getElementById(id);
+      if (!field || typeof value !== "boolean") return;
+      field.checked = value;
+      field.dispatchEvent(new Event("change", { bubbles: true }));
+    };
+
+    const populatePrinterOptions = async (cupsPrinterName) => {
+      const requestNumber = ++latestRequest;
+      status.textContent = "Loading printer capabilities…";
+      status.classList.remove("error");
+
+      try {
+        const url = new URL(wrapper.dataset.optionsUrl, window.location.origin);
+        url.searchParams.set("name", cupsPrinterName);
+        const response = await fetch(url, {
+          credentials: "same-origin",
+          headers: { Accept: "application/json" },
+        });
+        const options = await response.json();
+        if (!response.ok) throw new Error(options.error || "Could not load printer capabilities.");
+        if (requestNumber !== latestRequest) return;
+
+        // Do not overwrite an administrator's custom display name while editing.
+        const printerName = document.getElementById("id_name");
+        if (printerName && !printerName.value.trim()) {
+          printerName.value = cupsPrinterName;
+          printerName.dispatchEvent(new Event("change", { bubbles: true }));
+        }
+        setFieldValue("id_printer_type", "LP");
+        setCheckboxValue("id_color_supported", options.color_supported);
+        setCheckboxValue("id_duplex_supported", options.duplex_supported);
+
+        const inlinePrefix = input.id.replace(/cups_printer_name$/, "");
+        [
+          "print_grayscale_param",
+          "print_color_param",
+          "print_one_sided_param",
+          "print_two_sided_long_edge_param",
+          "print_two_sided_short_edge_param",
+        ].forEach(fieldName => setFieldValue(`${inlinePrefix}${fieldName}`, options[fieldName]));
+
+        status.textContent = "Printer capabilities loaded.";
+      } catch (error) {
+        if (requestNumber !== latestRequest) return;
+        status.textContent = error.message || "Could not load printer capabilities.";
+        status.classList.add("error");
+      }
+    };
+
     const items = wrapper.querySelectorAll("ul li");
     items.forEach(item => {
       const onSelect = () => {
-        input.value = item.getAttribute("data-value");
+        const cupsPrinterName = item.getAttribute("data-value");
+        input.value = cupsPrinterName;
+        input.dispatchEvent(new Event("change", { bubbles: true }));
         input.focus();
+        populatePrinterOptions(cupsPrinterName);
       };
       const button = item.querySelector(".autocomplete-button");
       button.addEventListener("click", (event) => {
@@ -19,7 +82,7 @@ document.addEventListener("DOMContentLoaded", () => {
         onSelect();
       });
       button.addEventListener('keyup', (event) => {
-        if (event.key !== 'Space') return;
+        if (event.key !== ' ' && event.code !== 'Space') return;
         event.preventDefault();
         onSelect();
       });

backend/control/templates/widgets/cups_printer_name_autocomplete.html

@@ -1,4 +1,4 @@
-<div class="cups-printer-list-autocomplete">
+<div class="cups-printer-list-autocomplete" data-options-url="{{ cups_printer_options_url }}">
     <input
         class="vTextField"
         type="text"
@@ -22,4 +22,5 @@
     {% else %}
     <p class="empty-message">No CUPS printer found</p>
     {% endif %}
+    <p class="cups-printer-options-status" role="status" aria-live="polite"></p>
 </div>

backend/control/widgets.py

@@ -1,4 +1,5 @@
 from django import forms
+from django.urls import reverse
 
 
 class CupsPrinterNameAutocomplete(forms.TextInput):
@@ -12,6 +13,7 @@ def get_context(self, name, value, attrs):
         context = super().get_context(name, value, attrs)
         printer_names = self.get_printer_names()
         context["printer_names"] = printer_names
+        context["cups_printer_options_url"] = reverse('admin:control_printer_cups_printer_options')
         return context
 
     class Media:

backend/printing/backends.py

@@ -3,6 +3,7 @@
 import subprocess
 import time
 from abc import ABC, abstractmethod
+from pathlib import Path
 from typing import Any
 
 from django.conf import settings
@@ -58,6 +59,142 @@ def print(self, job: GutenbergJob, file_path: str):
 
 class LocalCupsPrinter(PrinterBackend):
     common_options = ['-h', settings.CUPS_SERVERNAME]
+    ipp_capabilities_test = Path(__file__).with_name('ipptool') / 'get-printer-capabilities.test'
+
+    @staticmethod
+    def _parse_lpoptions(output: str) -> dict[str, list[str]]:
+        """Return the available values for each option from ``lpoptions -l`` output."""
+        options = {}
+        for line in output.splitlines():
+            match = re.match(r'^(?P<name>[^/\s:]+)(?:/[^:]+)?:\s*(?P<values>.*)$', line)
+            if not match:
+                continue
+            options[match.group('name')] = [value.lstrip('*') for value in match.group('values').split()]
+        return options
+
+    @staticmethod
+    def _parse_ipptool_attributes(output: str) -> dict[str, list[str]]:
+        """Extract the requested simple IPP attributes from ``ipptool -tv`` output."""
+        attributes = {}
+        for line in output.splitlines():
+            match = re.match(r'^\s*(?P<name>[\w-]+) \([^)]*\) = (?P<values>.*)$', line)
+            if not match:
+                continue
+            attributes[match.group('name')] = [value.strip() for value in match.group('values').split(',')]
+        return attributes
+
+    @staticmethod
+    def _get_cups_printer_uri(cups_printer_name: str) -> str | None:
+        output = subprocess.check_output(
+            ['lpstat'] + LocalCupsPrinter.common_options + ['-v', cups_printer_name],
+            stderr=subprocess.STDOUT,
+            text=True,
+            timeout=TASK_TIMEOUT_S,
+        )
+        match = re.search(r'^device for [^:]+:\s*(?P<uri>\S+)$', output, re.MULTILINE)
+        return match.group('uri') if match else None
+
+    @staticmethod
+    def _option_value(options: dict[str, list[str]], option_names: tuple[str, ...],
+                      values: tuple[str, ...]) -> str | None:
+        """Find a CUPS option and value without changing CUPS' original spelling."""
+        for option_name in option_names:
+            available_values = options.get(option_name, [])
+            value_by_normalized_name = {value.casefold(): value for value in available_values}
+            for value in values:
+                if selected := value_by_normalized_name.get(value.casefold()):
+                    # IPP capability attributes end in ``-supported`` while
+                    # CUPS expects the corresponding job-template attribute.
+                    return f'{option_name.removesuffix("-supported")}={selected}'
+        return None
+
+    @staticmethod
+    def _configuration_from_options(cups_printer_name: str,
+                                    options: dict[str, list[str]]) -> dict[str, str | bool | None]:
+        """Map CUPS or IPP capabilities to the fields Gutenberg currently supports."""
+        color_option_names = ('print-color-mode-supported', 'print-color-mode', 'ColorModel', 'ColorMode',
+                              'OutputMode')
+        grayscale_param = LocalCupsPrinter._option_value(
+            options, color_option_names, ('monochrome', 'mono', 'gray', 'grey', 'black'))
+        color_param = LocalCupsPrinter._option_value(
+            options, color_option_names, ('color', 'rgb', 'cmyk'))
+        one_sided_param = LocalCupsPrinter._option_value(
+            options, ('sides-supported', 'sides'), ('one-sided',))
+        two_sided_long_edge_param = LocalCupsPrinter._option_value(
+            options, ('sides-supported', 'sides'), ('two-sided-long-edge',))
+        two_sided_short_edge_param = LocalCupsPrinter._option_value(
+            options, ('sides-supported', 'sides'), ('two-sided-short-edge',))
+        # Older PPD-based CUPS queues commonly use these names instead of
+        # the IPP-standard ``sides`` option.
+        if one_sided_param is None:
+            one_sided_param = LocalCupsPrinter._option_value(options, ('Duplex',), ('None',))
+        if two_sided_long_edge_param is None:
+            two_sided_long_edge_param = LocalCupsPrinter._option_value(
+                options, ('Duplex',), ('DuplexNoTumble',))
+        if two_sided_short_edge_param is None:
+            two_sided_short_edge_param = LocalCupsPrinter._option_value(
+                options, ('Duplex',), ('DuplexTumble',))
+
+        return {
+            'cups_printer_name': cups_printer_name,
+            'color_supported': 'true' in {value.casefold() for value in options.get('color-supported', [])} or
+                               color_param is not None,
+            # Gutenberg's current boolean model advertises both duplex modes.
+            'duplex_supported': two_sided_long_edge_param is not None and two_sided_short_edge_param is not None,
+            'print_grayscale_param': grayscale_param,
+            'print_color_param': color_param,
+            'print_one_sided_param': one_sided_param,
+            'print_two_sided_long_edge_param': two_sided_long_edge_param,
+            'print_two_sided_short_edge_param': two_sided_short_edge_param,
+        }
+
+    @staticmethod
+    def get_cups_printer_options(cups_printer_name: str) -> dict[str, str | bool | None]:
+        """Discover the subset of CUPS options Gutenberg can configure for a queue.
+
+        The result deliberately mirrors ``LocalPrinterParams`` rather than exposing
+        every driver-specific CUPS option. Driverless queues are queried through
+        IPP; PPD-based queues fall back to ``lpoptions -l``.
+        """
+        try:
+            printer_uri = LocalCupsPrinter._get_cups_printer_uri(cups_printer_name)
+            if printer_uri and printer_uri.startswith(('ipp://', 'ipps://')):
+                output = subprocess.check_output(
+                    # ``-t`` reports the test result and ``-v`` includes the
+                    # response attributes which we parse below.
+                    ['ipptool', '-tv', printer_uri, str(LocalCupsPrinter.ipp_capabilities_test)],
+                    stderr=subprocess.STDOUT,
+                    text=True,
+                    timeout=TASK_TIMEOUT_S,
+                )
+                options = LocalCupsPrinter._parse_ipptool_attributes(output)
+                if any(name in options for name in ('color-supported', 'print-color-mode-supported',
+                                                    'sides-supported')):
+                    return LocalCupsPrinter._configuration_from_options(cups_printer_name, options)
+                logger.warning("IPP capability query for CUPS printer %s returned no supported options",
+                               cups_printer_name)
+        except Exception as error:
+            logger.warning("Failed to query IPP capabilities for CUPS printer %s: %s", cups_printer_name, error,
+                           exc_info=True)
+
+        try:
+            output = subprocess.check_output(
+                ['lpoptions'] + LocalCupsPrinter.common_options + ['-p', cups_printer_name, '-l'],
+                stderr=subprocess.STDOUT,
+                text=True,
+                timeout=TASK_TIMEOUT_S,
+            )
+        except Exception as error:
+            logger.error("Failed to get options for CUPS printer %s: %s", cups_printer_name, error,
+                         exc_info=True)
+            return {}
+
+        options = LocalCupsPrinter._parse_lpoptions(output)
+        if not options:
+            logger.error("CUPS returned no options for printer %s", cups_printer_name)
+            return {}
+        return LocalCupsPrinter._configuration_from_options(cups_printer_name, options)
+
     @staticmethod
     def parse_lpstat_job_status(output: str, backend_job_id: Any) -> str | None:
         """

backend/printing/ipptool/get-printer-capabilities.test

@@ -0,0 +1,15 @@
+# Query only the standard attributes required to configure Gutenberg's printer model.
+{
+    NAME "Get printer capabilities"
+    OPERATION get-printer-attributes
+    VERSION 2.0
+
+    GROUP operation
+    ATTR charset attributes-charset utf-8
+    ATTR language attributes-natural-language en
+    ATTR uri printer-uri $uri
+    # A comma-separated value makes this a 1setOf keyword attribute.
+    ATTR keyword requested-attributes color-supported,print-color-mode-supported,sides-supported
+
+    STATUS successful-ok
+}

backend/printing/tasks.py

@@ -29,3 +29,9 @@ def get_own_supported_formats(state) -> dict:
 @control_command(name="gutenberg_list_cups_printer_names")
 def list_cups_printer_names(state) -> list[str]:
     return LocalCupsPrinter.list_cups_printer_names()
+
+
+@control_command(name="gutenberg_get_cups_printer_options")
+def get_cups_printer_options(state, cups_printer_name: str) -> dict[str, str | bool | None]:
+    """Return the configurable Gutenberg options for one CUPS printer queue."""
+    return LocalCupsPrinter.get_cups_printer_options(cups_printer_name)