Split brother_ql_print -> brother_ql.{printing,discovering}

This commit is contained in:
Philipp Klaus
2018-07-31 21:59:32 +02:00
parent 5267986fd0
commit c42773b613
3 changed files with 125 additions and 47 deletions
+21 -47
View File
@@ -4,22 +4,23 @@
Testing the packaged version of the Linux Kernel backend Testing the packaged version of the Linux Kernel backend
""" """
import argparse, logging, sys, time import argparse, logging, sys
from pprint import pprint
from brother_ql.backends import backend_factory, guess_backend, available_backends from brother_ql.backends import backend_factory, guess_backend, available_backends
from brother_ql.reader import interpret_response from brother_ql.printing import send
from brother_ql.discovering import discover, pretty_print_discovered_devices
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def main(): def main():
# Command line parsing...
parser = argparse.ArgumentParser() parser = argparse.ArgumentParser()
parser.add_argument('--backend', choices=available_backends, help='Forces the use of a specific backend') parser.add_argument('--backend', choices=available_backends, help='Forces the use of a specific backend')
parser.add_argument('--list-printers', action='store_true', help='List the devices available with the selected --backend') parser.add_argument('--list-printers', action='store_true', help='List the devices available with the selected --backend')
parser.add_argument('--debug', action='store_true', help='Enable debugging output') parser.add_argument('--debug', action='store_true', help='Enable debugging output')
parser.add_argument('instruction_file', nargs='?', help='file containing the instructions to be sent to the printer') parser.add_argument('instruction_file', nargs='?', help='file containing the instructions to be sent to the printer')
parser.add_argument('device', metavar='DEVICE_IDENTIFIER', nargs='?', help='Identifier string specifying the device. If not specified, select first detected device') parser.add_argument('printer', metavar='PRINTER_IDENTIFIER', nargs='?', help='Identifier string specifying the printer. If not specified, selects the first detected device.')
args = parser.parse_args() args = parser.parse_args()
if args.list_printers and not args.backend: if args.list_printers and not args.backend:
@@ -28,6 +29,7 @@ def main():
if not args.list_printers and not args.instruction_file: if not args.list_printers and not args.instruction_file:
parser.error("the following arguments are required: instruction_file") parser.error("the following arguments are required: instruction_file")
# Reading the instruction input file into content variable
if args.instruction_file == '-': if args.instruction_file == '-':
try: try:
content = sys.stdin.buffer.read() content = sys.stdin.buffer.read()
@@ -37,76 +39,48 @@ def main():
with open(args.instruction_file, 'rb') as f: with open(args.instruction_file, 'rb') as f:
content = f.read() content = f.read()
# Setting up the requested level of logging.
level = logging.DEBUG if args.debug else logging.INFO level = logging.DEBUG if args.debug else logging.INFO
logging.basicConfig(level=level) logging.basicConfig(level=level)
# State any shortcomings of this software early on.
if args.backend == 'network': if args.backend == 'network':
logger.warning("The network backend doesn't supply any 'readback' functionality. No status reports will be received.") logger.warning("The network backend doesn't supply any 'readback' functionality. No status reports will be received.")
# Select the backend based: Either explicitly stated or derived from identifier. Otherwise: Default.
selected_backend = None selected_backend = None
if args.backend: if args.backend:
selected_backend = args.backend selected_backend = args.backend
else: else:
try: try:
selected_backend = guess_backend(args.device) selected_backend = guess_backend(args.printer)
except: except:
logger.info("No backend stated. Selecting the default linux_kernel backend.") logger.info("No backend stated. Selecting the default linux_kernel backend.")
selected_backend = 'linux_kernel' selected_backend = 'linux_kernel'
be = backend_factory(selected_backend) # List any printers found, if explicitly asked to do so or if no identifier has been provided.
list_available_devices = be['list_available_devices'] if args.list_printers or not args.printer:
BrotherQLBackend = be['backend_class'] available_devices = discover(backend=selected_backend)
pretty_print_discovered_devices(available_devices)
if args.list_printers or not args.device:
available_devices = list_available_devices()
for ad in available_devices:
result = {'model': 'unknown'}
result.update(ad)
logger.info(" Found a label printer: {identifier} (model: {model})".format(**result))
if args.list_printers: if args.list_printers:
for printer in available_devices: for printer in available_devices:
print(printer['identifier']) print(printer['identifier'])
sys.exit(0) sys.exit(0)
# Determine the identifier. Either selecting the explicitly stated one or using the first found device.
identifier = None identifier = None
if not args.device: if not args.printer:
"We need to search for available devices and select the first." "We need to search for available devices and select the first."
if not available_devices: if not available_devices:
sys.exit("No printer found") sys.exit("No printer found")
identifier = available_devices[0]['identifier'] identifier = available_devices[0]['identifier']
print("Selecting first device %s" % identifier) print("Selecting first device %s" % identifier)
else: else:
"An identifier for the device was given, let's use it." "A string identifier for the device was given, let's use it."
identifier = args.device identifier = args.printer
printer = BrotherQLBackend(identifier) # Finally, do the actual printing.
send(instructions=content, printer_identifier=identifier, backend_name=selected_backend, block=True)
start = time.time()
logger.info('Sending instructions to the printer. Total: %d bytes.', len(content))
printer.write(content)
if selected_backend == 'network':
""" No need to wait for completion. The network backend doesn't support readback. """
return
printing_completed = False
waiting_to_receive = False
while time.time() - start < 10:
data = printer.read()
if not data:
time.sleep(0.005)
continue
try:
result = interpret_response(data)
except ValueError:
logger.error("TIME %.3f - Couln't understand response: %s", time.time()-start, data)
continue
logger.debug('TIME %.3f - result: %s', time.time()-start, result)
if result['errors']:
logger.error('Errors occured: %s', result['errors'])
if result['status_type'] == 'Printing completed': printing_completed = True
if result['status_type'] == 'Phase change' and result['phase_type'] == 'Waiting to receive': waiting_to_receive = True
if printing_completed and waiting_to_receive:
break
if not (printing_completed and waiting_to_receive):
logger.warning('Printing potentially not successful?')
if __name__ == "__main__": main() if __name__ == "__main__": main()
+30
View File
@@ -0,0 +1,30 @@
#!/usr/bin/env python
"""
Printer device discovery using the different brother_ql.backends
"""
import logging
from brother_ql.backends import backend_factory, guess_backend, available_backends
logger = logging.getLogger(__name__)
def discover(backend_name='linux_kernel'):
be = backend_factory(selected_backend)
list_available_devices = be['list_available_devices']
BrotherQLBackend = be['backend_class']
available_devices = list_available_devices()
return available_devices
def pretty_print_discovered_devices(available_devices):
for ad in available_devices:
result = {'model': 'unknown'}
result.update(ad)
logger.info(" Found a label printer: {identifier} (model: {model})".format(**result))
for printer in available_devices:
print(printer['identifier'])
sys.exit(0)
+74
View File
@@ -0,0 +1,74 @@
#!/usr/bin/env python
"""
The implementation of printing with the brother_ql package.
"""
import logging, time
from brother_ql.backends import backend_factory, guess_backend
from brother_ql.reader import interpret_response
logger = logging.getLogger(__name__)
def send(instructions, printer_identifier=None, backend_name=None, blocking=True):
"""
instructions: Bytes containing the instructions to be sent to the printer.
printer_identifier: String descriptor for the printer.
backend_name: Can enforce the use of a specific backend.
blocking: Boolean indicating whether the print() call should wait for completion.
"""
selected_backend = None
if backend_name:
selected_backend = backend_name
else:
try:
selected_backend = guess_backend(printer_identifier)
except:
logger.info("No backend stated. Selecting the default linux_kernel backend.")
selected_backend = 'linux_kernel'
be = backend_factory(selected_backend)
list_available_devices = be['list_available_devices']
BrotherQLBackend = be['backend_class']
printer = BrotherQLBackend(printer_identifier)
start = time.time()
logger.info('Sending instructions to the printer. Total: %d bytes.', len(instructions))
printer.write(instructions)
if not blocking:
return
if selected_backend == 'network':
""" No need to wait for completion. The network backend doesn't support readback. """
return
printing_completed = False
waiting_to_receive = False
while time.time() - start < 10:
data = printer.read()
if not data:
time.sleep(0.005)
continue
try:
result = interpret_response(data)
except ValueError:
logger.error("TIME %.3f - Couln't understand response: %s", time.time()-start, data)
continue
logger.debug('TIME %.3f - result: %s', time.time()-start, result)
if result['errors']:
logger.error('Errors occured: %s', result['errors'])
if result['status_type'] == 'Printing completed':
printing_completed = True
if result['status_type'] == 'Phase change' and result['phase_type'] == 'Waiting to receive':
waiting_to_receive = True
if printing_completed and waiting_to_receive:
break
if not printing_completed:
logger.warning("'printing completed' status not received.")
if not waiting_to_receive:
logger.warning("'waiting to receive' status not received.")
if (not printing_completed) or (not waiting_to_receive):
logger.warning('Printing potentially not successful?')
if printing_completed and waiting_to_receive:
logger.info("Printing was successful. Waiting for the next job.")