blacklight/platypush
1
0
Fork 0
forked from platypush/platypush
Code Issues Pull requests Projects Releases Packages Wiki Activity

[#398] cups plugin refactor.

Browse source 
1. Renamed plugin: `printer.cups` ➡️  `cups`.
2. Replaced `Response` objects with schemas.
This commit is contained in:
Fabio Manganiello 2024-05-13 02:21:04 +02:00
parent 2efffb8ebe
commit 7ae99b4325
Signed by: blacklight
GPG key ID: D90FBA7F76362774
11 changed files with 257 additions and 134 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
docs/source/platypush/plugins/cups.rst Normal file
View file

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

5
docs/source/platypush/plugins/printer.cups.rst
View file

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

5
docs/source/platypush/responses/printer.cups.rst
View file

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

2
docs/source/plugins.rst
View file

@ -26,6 +26,7 @@ Plugins
platypush/plugins/clipboard.rst
platypush/plugins/config.rst
platypush/plugins/csv.rst
platypush/plugins/cups.rst
platypush/plugins/db.rst
platypush/plugins/dbus.rst
platypush/plugins/dropbox.rst
@ -99,7 +100,6 @@ Plugins
platypush/plugins/otp.rst
platypush/plugins/pihole.rst
platypush/plugins/ping.rst
platypush/plugins/printer.cups.rst
platypush/plugins/pushbullet.rst
platypush/plugins/pwm.pca9685.rst
platypush/plugins/qrcode.rst

1
docs/source/responses.rst
View file

@ -6,5 +6,4 @@ Responses
:maxdepth: 1
:caption: Responses:
platypush/responses/printer.cups.rst
platypush/responses/tensorflow.rst

0
platypush/message/response/printer/__init__.py
View file

56
platypush/message/response/printer/cups.py
View file

@ -1,56 +0,0 @@
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:

175
platypush/plugins/printer/cups/__init__.py → platypush/plugins/cups/__init__.py
View file

@ -2,35 +2,36 @@ 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
from platypush.schemas.cups import JobAddedSchema, PrinterSchema
class PrinterCupsPlugin(Plugin):
class CupsPlugin(Plugin):
"""
A plugin to interact with a CUPS printer server.
A plugin to interact with local and remote printers over a CUPS server.
"""
def __init__(
self, host: str = 'localhost', printer: Optional[str] = None, **kwargs
self,
host: str = 'localhost',
port: int = 631,
printer: Optional[str] = None,
**kwargs
):
"""
:param host: CUPS host IP/name (default: localhost).
:param port: CUPS port (default: 631).
:param printer: Default printer name that should be used.
"""
super().__init__(**kwargs)
self.host = host
self.port = port
self.printer = printer
def _get_connection(self, host: Optional[str] = None):
# noinspection PyPackageRequirements
def _get_connection(self, host: Optional[str] = None, port: Optional[int] = None):
import cups
connection = cups.Connection(host=host or self.host)
connection = cups.Connection(host=host or self.host, port=port or self.port)
return connection
def _get_printer(self, printer: Optional[str] = None):
@ -39,46 +40,38 @@ class PrinterCupsPlugin(Plugin):
return printer
@action
def get_printers(self, host: Optional[str] = None) -> PrintersResponse:
def get_printers(
self, host: Optional[str] = None, port: Optional[int] = None
) -> List[dict]:
"""
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.
:param port: CUPS server port (default: default configured ``port``).
:return: .. schema:: cups.PrinterSchema(many=True)
"""
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()
]
)
conn = self._get_connection(host, port)
return PrinterSchema().dump(conn.getPrinters())
@action
def print_test_page(
self, printer: Optional[str] = None, host: Optional[str] = None
) -> PrinterJobAddedResponse:
self,
printer: Optional[str] = None,
host: Optional[str] = None,
port: Optional[int] = None,
) -> dict:
"""
Print the CUPS test page.
:param printer: Printer name (default: default configured ``printer``).
:param host: CUPS server IP/name (default: default configured ``host``).
:param port: CUPS server port (default: default configured ``port``).
:return: .. schema:: cups.JobAddedSchema
"""
conn = self._get_connection(host)
conn = self._get_connection(host, port)
printer = self._get_printer(printer)
job_id = conn.printTestPage(printer)
return PrinterJobAddedResponse(printer=printer, job_id=job_id)
return dict(JobAddedSchema().dump({'printer': printer, 'job_id': job_id}))
@action
def print_file(
@ -86,25 +79,28 @@ class PrinterCupsPlugin(Plugin):
filename: str,
printer: Optional[str] = None,
host: Optional[str] = None,
port: Optional[int] = None,
title: Optional[str] = None,
options: Optional[Dict[str, Any]] = None,
) -> PrinterJobAddedResponse:
) -> dict:
"""
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 port: CUPS server port (default: default configured ``port``).
:param title: Print title.
:param options: Extra CUPS name->value options.
:return: .. schema:: cups.JobAddedSchema
"""
filename = os.path.abspath(os.path.expanduser(filename))
conn = self._get_connection(host)
conn = self._get_connection(host, port)
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)
return dict(JobAddedSchema().dump({'printer': printer, 'job_id': job_id}))
@action
def print_files(
@ -112,25 +108,28 @@ class PrinterCupsPlugin(Plugin):
filenames: List[str],
printer: Optional[str] = None,
host: Optional[str] = None,
port: Optional[int] = None,
title: Optional[str] = None,
options: Optional[Dict[str, Any]] = None,
) -> PrinterJobAddedResponse:
) -> dict:
"""
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 port: CUPS server port (default: default configured ``port``).
:param title: Print title.
:param options: Extra CUPS name->value options.
:return: .. schema:: cups.JobAddedSchema
"""
filenames = [os.path.abspath(os.path.expanduser(f)) for f in filenames]
conn = self._get_connection(host)
conn = self._get_connection(host, port)
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)
return dict(JobAddedSchema().dump({'printer': printer, 'job_id': job_id}))
@action
def add_printer(
@ -140,6 +139,7 @@ class PrinterCupsPlugin(Plugin):
info: str,
location: Optional[str] = None,
host: Optional[str] = None,
port: Optional[int] = None,
):
"""
Add a printer.
@ -147,89 +147,120 @@ class PrinterCupsPlugin(Plugin):
: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 port: CUPS server port (default: default configured ``port``).
:param info: Human-readable information about the printer.
:param location: Human-readable printer location info.
"""
conn = self._get_connection(host)
conn = self._get_connection(host, port)
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):
def delete_printer(
self, printer: str, host: Optional[str] = None, port: Optional[int] = 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 = self._get_connection(host, port)
conn.deletePrinter(printer)
@action
def enable_printer(self, printer: Optional[str], host: Optional[str] = None):
def enable_printer(
self,
printer: Optional[str],
host: Optional[str] = None,
port: Optional[int] = None,
):
"""
Enable a printer on a CUPS server.
:param printer: Printer name.
:param host: CUPS server IP/name (default: default configured ``host``).
:param port: CUPS server port (default: default configured ``port``).
"""
conn = self._get_connection(host)
conn = self._get_connection(host, port)
printer = self._get_printer(printer)
conn.enablePrinter(printer)
@action
def disable_printer(
self, printer: Optional[str] = None, host: Optional[str] = None
self,
printer: Optional[str] = None,
host: Optional[str] = None,
port: Optional[int] = None,
):
"""
Disable a printer on a CUPS server.
:param printer: Printer name.
:param host: CUPS server IP/name (default: default configured ``host``).
:param port: CUPS server port (default: default configured ``port``).
"""
conn = self._get_connection(host)
conn = self._get_connection(host, port)
printer = self._get_printer(printer)
conn.disablePrinter(printer)
@action
def get_jobs(self, host: Optional[str] = None) -> Dict[int, Dict[str, Any]]:
def get_jobs(
self, host: Optional[str] = None, port: Optional[int] = None
) -> Dict[int, Dict[str, Any]]:
"""
Get the list of active jobs.
:param host: CUPS server IP/name (default: default configured ``host``).
:param port: CUPS server port (default: default configured ``port``).
:return: A job_id -> job_info dict.
"""
conn = self._get_connection(host)
conn = self._get_connection(host, port)
return conn.getJobs()
@action
def accept_jobs(self, printer: Optional[str] = None, host: Optional[str] = None):
def accept_jobs(
self,
printer: Optional[str] = None,
host: Optional[str] = None,
port: Optional[int] = None,
):
"""
Start accepting jobs on a printer.
:param printer: Printer name.
:param host: CUPS server IP/name (default: default configured ``host``).
:param port: CUPS server port (default: default configured ``port``).
"""
conn = self._get_connection(host)
conn = self._get_connection(host, port)
printer = self._get_printer(printer)
conn.acceptJobs(printer)
@action
def reject_jobs(self, printer: Optional[str] = None, host: Optional[str] = None):
def reject_jobs(
self,
printer: Optional[str] = None,
host: Optional[str] = None,
port: Optional[int] = None,
):
"""
Start rejecting jobs on a printer.
:param printer: Printer name.
:param host: CUPS server IP/name (default: default configured ``host``).
:param port: CUPS server port (default: default configured ``port``).
"""
conn = self._get_connection(host)
conn = self._get_connection(host, port)
printer = self._get_printer(printer)
conn.rejectJobs(printer)
@action
def cancel_job(
self, job_id: int, purge_job: bool = False, host: Optional[str] = None
self,
job_id: int,
purge_job: bool = False,
host: Optional[str] = None,
port: Optional[int] = None,
):
"""
Cancel a printer job.
@ -237,8 +268,9 @@ class PrinterCupsPlugin(Plugin):
: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``).
:param port: CUPS server port (default: default configured ``port``).
"""
conn = self._get_connection(host)
conn = self._get_connection(host, port)
conn.cancelJob(job_id, purge_job=purge_job)
@action
@ -248,6 +280,7 @@ class PrinterCupsPlugin(Plugin):
source_printer_uri: str,
target_printer_uri: str,
host: Optional[str] = None,
port: Optional[int] = None,
):
"""
Move a job to another printer/URI.
@ -256,8 +289,9 @@ class PrinterCupsPlugin(Plugin):
:param source_printer_uri: Source printer URI.
:param target_printer_uri: Target printer URI.
:param host: CUPS server IP/name (default: default configured ``host``).
:param port: CUPS server port (default: default configured ``port``).
"""
conn = self._get_connection(host)
conn = self._get_connection(host, port)
conn.moveJob(
printer_uri=source_printer_uri,
job_id=job_id,
@ -266,15 +300,19 @@ class PrinterCupsPlugin(Plugin):
@action
def finish_document(
self, printer: Optional[str] = None, host: Optional[str] = None
self,
printer: Optional[str] = None,
host: Optional[str] = None,
port: Optional[int] = 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``).
:param port: CUPS server port (default: default configured ``port``).
"""
conn = self._get_connection(host)
conn = self._get_connection(host, port)
printer = self._get_printer(printer)
conn.finishDocument(printer)
@ -284,6 +322,7 @@ class PrinterCupsPlugin(Plugin):
printer_class: str,
printer: Optional[str] = None,
host: Optional[str] = None,
port: Optional[int] = None,
):
"""
Add a printer to a class.
@ -291,8 +330,9 @@ class PrinterCupsPlugin(Plugin):
:param printer_class: Class name.
:param printer: Printer name.
:param host: CUPS server IP/name (default: default configured ``host``).
:param port: CUPS server port (default: default configured ``port``).
"""
conn = self._get_connection(host)
conn = self._get_connection(host, port)
printer = self._get_printer(printer)
conn.addPrinterToClass(printer, printer_class)
@ -302,6 +342,7 @@ class PrinterCupsPlugin(Plugin):
printer_class: str,
printer: Optional[str] = None,
host: Optional[str] = None,
port: Optional[int] = None,
):
"""
Delete a printer from a class.
@ -309,20 +350,24 @@ class PrinterCupsPlugin(Plugin):
:param printer_class: Class name.
:param printer: Printer name.
:param host: CUPS server IP/name (default: default configured ``host``).
:param port: CUPS server port (default: default configured ``port``).
"""
conn = self._get_connection(host)
conn = self._get_connection(host, port)
printer = self._get_printer(printer)
conn.deletePrinterFromClass(printer, printer_class)
@action
def get_classes(self, host: Optional[str] = None) -> Dict[str, Dict[str, Any]]:
def get_classes(
self, host: Optional[str] = None, port: Optional[int] = 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``).
:param port: CUPS server port (default: default configured ``port``).
:return: dict - class_name -> class_info.
"""
conn = self._get_connection(host)
conn = self._get_connection(host, port)
return conn.getClasses()

2
platypush/plugins/printer/cups/manifest.yaml → platypush/plugins/cups/manifest.yaml
View file

@ -12,5 +12,5 @@ manifest:
- python-pycups
pip:
- pycups
package: platypush.plugins.printer.cups
package: platypush.plugins.cups
type: plugin

0
platypush/plugins/printer/__init__.py
View file

140
platypush/schemas/cups.py Normal file
View file

@ -0,0 +1,140 @@
from marshmallow import EXCLUDE, fields
from marshmallow.schema import Schema
class PrinterSchema(Schema):
"""
Schema for the printers returned by the CUPS API.
"""
class Meta: # type: ignore
"""
Exclude unknown fields.
"""
unknown = EXCLUDE
name = fields.String(
required=True,
metadata={
'description': 'The name of the printer.',
'example': 'HP_DeskJet_5820_series_585C26',
},
)
printer_type = fields.String(
required=True,
data_key='printer-type',
metadata={
'description': 'Unique type ID of the printer.',
'example': 2101260,
},
)
info = fields.String(
required=True,
data_key='printer-info',
metadata={
'description': 'Human-readable description of the printer.',
'example': 'HP DeskJet 5820 series',
},
)
uri = fields.String(
required=True,
data_key='device-uri',
metadata={
'description': 'The URI of the printer.',
'example': (
'dnssd://HP%20DeskJet%205820%20series%20%5B585C26%5D._ipp._tcp.local/'
'?uuid=1c852a4d-b800-1f08-abcd-705a0f585c26'
),
},
)
state = fields.String(
required=True,
data_key='printer-state',
metadata={
'description': 'The state of the printer.',
'example': 3,
},
)
is_shared = fields.Boolean(
required=True,
data_key='printer-is-shared',
metadata={
'description': 'Whether the printer is shared.',
'example': False,
},
)
state_message = fields.String(
required=True,
data_key='printer-state-message',
metadata={
'description': 'The state message of the printer.',
'example': 'Idle',
},
)
state_reasons = fields.List(
fields.String(),
required=True,
data_key='printer-state-reasons',
metadata={
'description': 'The reasons for the printer state.',
'example': ['none'],
},
)
location = fields.String(
required=True,
data_key='printer-location',
metadata={
'description': 'Human-readable location of the printer.',
'example': 'Living Room',
},
)
uri_supported = fields.String(
required=True,
data_key='printer-uri-supported',
metadata={
'description': 'The URI of the printer exposed by the CUPS server.',
'example': 'ipp://localhost:631/printers/HP_OfficeJet_5230',
},
)
make_and_model = fields.String(
required=True,
data_key='printer-make-and-model',
metadata={
'description': 'The make and model of the printer.',
'example': 'HP Officejet 5200 Series, hpcups 3.19.1',
},
)
class JobAddedSchema(Schema):
"""
Schema for a printer job added event.
"""
printer = fields.String(
required=True,
metadata={
'description': 'The name of the printer.',
'example': 'HP_DeskJet_5820_series_585C26',
},
)
job_id = fields.Integer(
required=True,
data_key='job-id',
metadata={
'description': 'The ID of the job.',
'example': 1,
},
)