Added CUPS plugin - closes #93

This commit is contained in:
Fabio Manganiello 2020-01-05 19:25:20 +01:00
parent a9fdbe7812
commit bcd05ecf4a
9 changed files with 366 additions and 0 deletions

View file

@ -234,6 +234,7 @@ autodoc_mock_imports = ['googlesamples.assistant.grpc.audio_helpers',
'telegram', 'telegram',
'telegram.ext', 'telegram.ext',
'pyfirmata2' 'pyfirmata2'
'cups'
] ]
sys.path.insert(0, os.path.abspath('../..')) sys.path.insert(0, os.path.abspath('../..'))

View file

@ -0,0 +1,5 @@
``platypush.plugins.printer.cups``
==================================
.. automodule:: platypush.plugins.printer.cups
:members:

View file

@ -76,6 +76,7 @@ Plugins
platypush/plugins/music.mpd.rst platypush/plugins/music.mpd.rst
platypush/plugins/music.snapcast.rst platypush/plugins/music.snapcast.rst
platypush/plugins/ping.rst platypush/plugins/ping.rst
platypush/plugins/printer.cups.rst
platypush/plugins/pushbullet.rst platypush/plugins/pushbullet.rst
platypush/plugins/redis.rst platypush/plugins/redis.rst
platypush/plugins/serial.rst platypush/plugins/serial.rst

View file

@ -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:

View file

View file

@ -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:

View file

@ -210,3 +210,7 @@ croniter
# Support for Arduino integration # Support for Arduino integration
# pyserial # pyserial
# pyfirmata2 # pyfirmata2
# Support for CUPS printers management
pycups

View file

@ -269,5 +269,7 @@ setup(
'telegram': ['python-telegram-bot'], 'telegram': ['python-telegram-bot'],
# Support for Arduino integration # Support for Arduino integration
'arduino': ['pyserial', 'pyfirmata2'], 'arduino': ['pyserial', 'pyfirmata2'],
# Support for CUPS printers management
'cups': ['pycups'],
}, },
) )