diff --git a/docs/source/conf.py b/docs/source/conf.py index 2bd472bc..16836ea3 100644 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -234,6 +234,7 @@ autodoc_mock_imports = ['googlesamples.assistant.grpc.audio_helpers', 'telegram', 'telegram.ext', 'pyfirmata2' + 'cups' ] sys.path.insert(0, os.path.abspath('../..')) diff --git a/docs/source/platypush/plugins/printer.cups.rst b/docs/source/platypush/plugins/printer.cups.rst new file mode 100644 index 00000000..88a56092 --- /dev/null +++ b/docs/source/platypush/plugins/printer.cups.rst @@ -0,0 +1,5 @@ +``platypush.plugins.printer.cups`` +================================== + +.. automodule:: platypush.plugins.printer.cups + :members: diff --git a/docs/source/plugins.rst b/docs/source/plugins.rst index 372f3907..d24237ab 100644 --- a/docs/source/plugins.rst +++ b/docs/source/plugins.rst @@ -76,6 +76,7 @@ Plugins platypush/plugins/music.mpd.rst platypush/plugins/music.snapcast.rst platypush/plugins/ping.rst + platypush/plugins/printer.cups.rst platypush/plugins/pushbullet.rst platypush/plugins/redis.rst platypush/plugins/serial.rst diff --git a/platypush/message/response/printer/__init__.py b/platypush/message/response/printer/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/platypush/message/response/printer/cups.py b/platypush/message/response/printer/cups.py new file mode 100644 index 00000000..6f626d27 --- /dev/null +++ b/platypush/message/response/printer/cups.py @@ -0,0 +1,56 @@ +from typing import Optional, List + +from platypush.message.response import Response + + +class PrinterResponse(Response): + def __init__(self, + *args, + name: str, + printer_type: int, + info: str, + uri: str, + state: int, + is_shared: bool, + state_message: Optional[str] = None, + state_reasons: Optional[List[str]] = None, + location: Optional[str] = None, + uri_supported: Optional[str] = None, + make_and_model: Optional[str] = None, + **kwargs): + super().__init__(*args, output={ + 'name': name, + 'printer_type': printer_type, + 'info': info, + 'uri': uri, + 'state': state, + 'is_shared': is_shared, + 'state_message': state_message, + 'state_reasons': state_reasons, + 'location': location, + 'uri_supported': uri_supported, + 'make_and_model': make_and_model, + }, **kwargs) + + +class PrintersResponse(Response): + def __init__(self, + *args, + printers: List[PrinterResponse], + **kwargs): + super().__init__(*args, output={p.output['name']: p.output for p in printers}, **kwargs) + + +class PrinterJobAddedResponse(Response): + def __init__(self, + *args, + printer: str, + job_id: int, + **kwargs): + super().__init__(*args, output={ + 'printer': printer, + 'job_id': job_id, + }, **kwargs) + + +# vim:sw=4:ts=4:et: diff --git a/platypush/plugins/printer/__init__.py b/platypush/plugins/printer/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/platypush/plugins/printer/cups.py b/platypush/plugins/printer/cups.py new file mode 100644 index 00000000..e7f7f403 --- /dev/null +++ b/platypush/plugins/printer/cups.py @@ -0,0 +1,297 @@ +import os + +from typing import Optional, Dict, Any, List + +from platypush.message.response.printer.cups import PrinterResponse, PrintersResponse, PrinterJobAddedResponse +from platypush.plugins import Plugin, action + + +class PrinterCupsPlugin(Plugin): + """ + A plugin to interact with a CUPS printer server. + + Requires: + + - **pycups** (``pip install pycups``) + + """ + + def __init__(self, host: str = 'localhost', printer: Optional[str] = None, **kwargs): + """ + :param host: CUPS host IP/name (default: localhost). + :param printer: Default printer name that should be used. + """ + super().__init__(**kwargs) + self.host = host + self.printer = printer + + def _get_connection(self, host: Optional[str] = None): + # noinspection PyPackageRequirements + import cups + connection = cups.Connection(host=host or self.host) + return connection + + def _get_printer(self, printer: Optional[str] = None): + printer = printer or self.printer + assert printer, 'No printer specified nor default printer configured' + return printer + + @action + def get_printers(self, host: Optional[str] = None) -> PrintersResponse: + """ + Get the list of printers registered on a CUPS server. + :param host: CUPS server host IP/name (default: default configured ``host``). + :return: :class:`platypush.message.response.printer.cups.PrintersResponse`, as a name -> attributes dict. + """ + conn = self._get_connection(host) + return PrintersResponse(printers=[ + PrinterResponse( + name=name, + printer_type=printer.get('printer-type'), + info=printer.get('printer-info'), + uri=printer.get('device-uri'), + state=printer.get('printer-state'), + is_shared=printer.get('printer-is-shared'), + state_message=printer.get('printer-state-message'), + state_reasons=printer.get('printer-state-reasons', []), + location=printer.get('printer-location'), + uri_supported=printer.get('printer-uri-supported'), + make_and_model=printer.get('printer-make-and-model'), + ) + for name, printer in conn.getPrinters().items() + ]) + + @action + def print_test_page(self, printer: Optional[str] = None, host: Optional[str] = None) -> PrinterJobAddedResponse: + """ + Print the CUPS test page. + + :param printer: Printer name (default: default configured ``printer``). + :param host: CUPS server IP/name (default: default configured ``host``). + """ + conn = self._get_connection(host) + printer = self._get_printer(printer) + job_id = conn.printTestPage(printer) + return PrinterJobAddedResponse(printer=printer, job_id=job_id) + + @action + def print_file(self, + filename: str, + printer: Optional[str] = None, + host: Optional[str] = None, + title: Optional[str] = None, + options: Optional[Dict[str, Any]] = None) -> PrinterJobAddedResponse: + """ + Print a file. + + :param filename: Path to the file to print. + :param printer: Printer name (default: default configured ``printer``). + :param host: CUPS server IP/name (default: default configured ``host``). + :param title: Print title. + :param options: Extra CUPS name->value options. + """ + filename = os.path.abspath(os.path.expanduser(filename)) + conn = self._get_connection(host) + printer = self._get_printer(printer) + job_id = conn.printFile(printer, filename=filename, title=title or '', options=options or {}) + return PrinterJobAddedResponse(printer=printer, job_id=job_id) + + @action + def print_files(self, + filenames: List[str], + printer: Optional[str] = None, + host: Optional[str] = None, + title: Optional[str] = None, + options: Optional[Dict[str, Any]] = None) -> PrinterJobAddedResponse: + """ + Print a list of files. + + :param filenames: Paths to the files to print. + :param printer: Printer name (default: default configured ``printer``). + :param host: CUPS server IP/name (default: default configured ``host``). + :param title: Print title. + :param options: Extra CUPS name->value options. + """ + filenames = [os.path.abspath(os.path.expanduser(f)) for f in filenames] + conn = self._get_connection(host) + printer = self._get_printer(printer) + job_id = conn.printFiles(printer, filenames=filenames, title=title or '', options=options or {}) + return PrinterJobAddedResponse(printer=printer, job_id=job_id) + + @action + def add_printer(self, + name: str, + ppd_file: str, + info: str, + location: Optional[str] = None, + host: Optional[str] = None): + """ + Add a printer. + + :param name: Printer name - alphanumeric + underscore characters only. + :param ppd_file: Path to the PPD file with the printer information and configuration. + :param host: CUPS server IP/name (default: default configured ``host``). + :param info: Human-readable information about the printer. + :param location: Human-readable printer location info. + """ + conn = self._get_connection(host) + ppd_file = os.path.abspath(os.path.expanduser(ppd_file)) + # noinspection PyArgumentList + conn.addPrinter(name=name, filename=ppd_file, info=info, location=location) + + @action + def delete_printer(self, printer: str, host: Optional[str] = None): + """ + Delete a printer from a CUPS server. + + :param printer: Printer name. + :param host: CUPS server IP/name (default: default configured ``host``). + """ + conn = self._get_connection(host) + conn.deletePrinter(printer) + + @action + def enable_printer(self, printer: Optional[str], host: Optional[str] = None): + """ + Enable a printer on a CUPS server. + + :param printer: Printer name. + :param host: CUPS server IP/name (default: default configured ``host``). + """ + conn = self._get_connection(host) + printer = self._get_printer(printer) + conn.enablePrinter(printer) + + @action + def disable_printer(self, printer: Optional[str] = None, host: Optional[str] = None): + """ + Disable a printer on a CUPS server. + + :param printer: Printer name. + :param host: CUPS server IP/name (default: default configured ``host``). + """ + conn = self._get_connection(host) + printer = self._get_printer(printer) + conn.disablePrinter(printer) + + @action + def get_jobs(self, host: Optional[str] = None) -> Dict[int, Dict[str, Any]]: + """ + Get the list of active jobs. + + :param host: CUPS server IP/name (default: default configured ``host``). + :return: A job_id -> job_info dict. + """ + conn = self._get_connection(host) + return conn.getJobs() + + @action + def accept_jobs(self, printer: Optional[str] = None, host: Optional[str] = None): + """ + Start accepting jobs on a printer. + + :param printer: Printer name. + :param host: CUPS server IP/name (default: default configured ``host``). + """ + conn = self._get_connection(host) + printer = self._get_printer(printer) + conn.acceptJobs(printer) + + @action + def reject_jobs(self, printer: Optional[str] = None, host: Optional[str] = None): + """ + Start rejecting jobs on a printer. + + :param printer: Printer name. + :param host: CUPS server IP/name (default: default configured ``host``). + """ + conn = self._get_connection(host) + printer = self._get_printer(printer) + conn.rejectJobs(printer) + + @action + def cancel_job(self, job_id: int, purge_job: bool = False, host: Optional[str] = None): + """ + Cancel a printer job. + + :param job_id: Job ID to cancel. + :param purge_job: Also remove the job from the server (default: False). + :param host: CUPS server IP/name (default: default configured ``host``). + """ + conn = self._get_connection(host) + conn.cancelJob(job_id, purge_job=purge_job) + + @action + def move_job(self, + job_id: int, + source_printer_uri: str, + target_printer_uri: str, + host: Optional[str] = None): + """ + Move a job to another printer/URI. + + :param job_id: Job ID to cancel. + :param source_printer_uri: Source printer URI. + :param target_printer_uri: Target printer URI. + :param host: CUPS server IP/name (default: default configured ``host``). + """ + conn = self._get_connection(host) + conn.moveJob(printer_uri=source_printer_uri, job_id=job_id, job_printer_uri=target_printer_uri) + + @action + def finish_document(self, printer: Optional[str] = None, host: Optional[str] = None): + """ + Finish sending a document to a printer. + + :param printer: Printer name (default: default configured ``printer``). + :param host: CUPS server IP/name (default: default configured ``host``). + """ + conn = self._get_connection(host) + printer = self._get_printer(printer) + conn.finishDocument(printer) + + @action + def add_printer_to_class(self, + printer_class: str, + printer: Optional[str] = None, + host: Optional[str] = None): + """ + Add a printer to a class. + + :param printer_class: Class name. + :param printer: Printer name. + :param host: CUPS server IP/name (default: default configured ``host``). + """ + conn = self._get_connection(host) + printer = self._get_printer(printer) + conn.addPrinterToClass(printer, printer_class) + + @action + def delete_printer_from_class(self, + printer_class: str, + printer: Optional[str] = None, + host: Optional[str] = None): + """ + Delete a printer from a class. + + :param printer_class: Class name. + :param printer: Printer name. + :param host: CUPS server IP/name (default: default configured ``host``). + """ + conn = self._get_connection(host) + printer = self._get_printer(printer) + conn.deletePrinterFromClass(printer, printer_class) + + @action + def get_classes(self, host: Optional[str] = None) -> Dict[str, Dict[str, Any]]: + """ + Get the list of classes on a CUPS server. + + :param host: CUPS server IP/name (default: default configured ``host``). + :return: dict - class_name -> class_info. + """ + conn = self._get_connection(host) + return conn.getClasses() + + +# vim:sw=4:ts=4:et: diff --git a/requirements.txt b/requirements.txt index b5d3631c..acb2adb2 100644 --- a/requirements.txt +++ b/requirements.txt @@ -210,3 +210,7 @@ croniter # Support for Arduino integration # pyserial # pyfirmata2 + +# Support for CUPS printers management +pycups + diff --git a/setup.py b/setup.py index 277e7a54..24139359 100755 --- a/setup.py +++ b/setup.py @@ -269,5 +269,7 @@ setup( 'telegram': ['python-telegram-bot'], # Support for Arduino integration 'arduino': ['pyserial', 'pyfirmata2'], + # Support for CUPS printers management + 'cups': ['pycups'], }, )