Added Network and Serial support

This commit is contained in:
Manuel F Martinez 2012-09-23 21:24:45 -07:00
parent e9effc23da
commit 3760788cfe
3 changed files with 142 additions and 67 deletions

View File

@ -1 +1 @@
__all__ = ["constants","escpos","exceptions"] __all__ = ["constants","escpos","exceptions","printer"]

View File

@ -2,63 +2,20 @@
''' '''
@author: Manuel F Martinez <manpaz@bashlinux.com> @author: Manuel F Martinez <manpaz@bashlinux.com>
@organization: Bashlinux @organization: Bashlinux
@copyright: Copyright (c) 2010 Bashlinux @copyright: Copyright (c) 2012 Bashlinux
@license: GPL @license: GPL
''' '''
import usb.core
import usb.util
import Image import Image
import time import time
from constants import * from constants import *
from exceptions import * from exceptions import *
class DeviceDescriptor:
""" Search device on USB tree and return it if found """
def __init__(self, idVendor, idProduct, interface) :
self.idVendor = idVendor
self.idProduct = idProduct
self.interface = interface
def get_device(self) :
device = usb.core.find(idVendor=self.idVendor, idProduct=self.idProduct)
return device
class Escpos: class Escpos:
""" ESC/POS Printer object """ """ ESC/POS Printer object """
handle = None
device = None device = None
def __init__(self, idVendor, idProduct, interface=0, in_ep=0x82, out_ep=0x01) :
self.idVendor = idVendor
self.idProduct = idProduct
self.interface = interface
self.in_ep = in_ep
self.out_ep = out_ep
device_descriptor = DeviceDescriptor(self.idVendor, self.idProduct, self.interface)
self.device = device_descriptor.get_device()
if self.device is None:
print "Cable isn't plugged in"
if self.device.is_kernel_driver_active(0):
try:
self.device.detach_kernel_driver(0)
except usb.core.USBError as e:
print "Could not detatch kernel driver: %s" % str(e)
try:
self.device.set_configuration()
self.device.reset()
except usb.core.USBError as e:
print "Could not set configuration: %s" % str(e)
def _raw(self, msg):
""" Print any of the commands above, or clear text """
self.device.write(self.out_ep, msg, self.interface)
def _check_image_size(self, size): def _check_image_size(self, size):
""" Check and fix the size of the image to 32 bits """ """ Check and fix the size of the image to 32 bits """
@ -73,6 +30,7 @@ class Escpos:
def _print_image(self, line, size): def _print_image(self, line, size):
""" Print formatted image """
i = 0 i = 0
cont = 0 cont = 0
buffer = "" buffer = ""
@ -94,7 +52,7 @@ class Escpos:
def image(self, img): def image(self, img):
"""Parse image and then print it""" """ Parse image and prepare it to a printable format """
pixels = [] pixels = []
pix_line = "" pix_line = ""
im_left = "" im_left = ""
@ -294,23 +252,4 @@ class Escpos:
elif ctl.upper() == "HT": elif ctl.upper() == "HT":
self._raw(CTL_HT) self._raw(CTL_HT)
elif ctl.upper() == "VT": elif ctl.upper() == "VT":
self._raw(CTL_VT) delf._raw(CTL_VT)
def __del__(self):
""" Release device interface """
if self.handle:
try:
self.handle.releaseInterface()
self.handle.resetEndpoint(self.out_ep)
self.handle.reset()
except Exception, err:
print err
self.handle, self.device = None, None
# Give a chance to return the interface to the system
# The following message could appear if the application is executed
# too fast twice or more times.
#
# >> could not detach kernel driver from interface 0: No data available
# >> No interface claimed
time.sleep(1)

136
printer.py Normal file
View File

@ -0,0 +1,136 @@
#!/usr/bin/python
'''
@author: Manuel F Martinez <manpaz@bashlinux.com>
@organization: Bashlinux
@copyright: Copyright (c) 2012 Bashlinux
@license: GPL
'''
import usb.core
import usb.util
import serial
import socket
from escpos import *
from constants import *
from exceptions import *
class Usb(Escpos):
""" Define USB printer """
def __init__(self, idVendor, idProduct, interface=0, in_ep=0x82, out_ep=0x01):
"""
@param idVendor : Vendor ID
@param idProduct : Product ID
@param interface : USB device interface
@param in_ep : Input end point
@param out_ep : Output end point
"""
self.idVendor = idVendor
self.idProduct = idProduct
self.interface = interface
self.in_ep = in_ep
self.out_ep = out_ep
self.open()
def open(self):
""" Search device on USB tree and set is as escpos device """
self.device = usb.core.find(idVendor=self.idVendor, idProduct=self.idProduct)
if self.device is None:
print "Cable isn't plugged in"
if self.device.is_kernel_driver_active(0):
try:
self.device.detach_kernel_driver(0)
except usb.core.USBError as e:
print "Could not detatch kernel driver: %s" % str(e)
try:
self.device.set_configuration()
self.device.reset()
except usb.core.USBError as e:
print "Could not set configuration: %s" % str(e)
def _raw(self, msg):
""" Print any command sent in raw format """
self.device.write(self.out_ep, msg, self.interface)
def __del__(self):
""" Release USB interface """
if self.device:
usb.util.dispose_resources(self.device)
self.device = None
class Serial(Escpos):
""" Define Serial printer """
def __init__(self, devfile="/dev/ttyUSB0", baudrate=9600, bytesize=8, timeout=5):
"""
@param devfile : Device file under dev filesystem
@param baudrate : Baud rate for serial transmission
@param timeout : Read/Write timeout
@param bytesize : Serial buffer size
"""
self.devfile = devfile
self.baudrate = baudrate
self.bytesize = bytesize
self.timeout = timeout
self.open()
def open(self):
""" Setup serial port and set is as escpos device """
self.device = serial.Serial(port=self.devfile, baudrate=self.baudrate, bytesize=self.bytesize, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, timeout=self.timeout, dsrdtr=True)
if self.device is not None:
print "Serial printer enabled"
else:
print "Unable to open serial printer on: %s" % self.devfile
def _raw(self, msg):
""" Print any command sent in raw format """
self.device.write(msg)
def __del__(self):
""" Close Serial interface """
if self.device is not None:
self.device.close()
class Network(Escpos):
""" Define Network printer """
def __init__(self,host,port=9100):
"""
@param host : Printer's hostname or IP address
@param port : Port to write to
"""
self.host = host
self.port = port
self.open()
def open(self):
""" Open TCP socket and set it as escpos device """
self.device = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.device.connect((self.host, self.port))
if self.device is None:
print "Could not open socket for %s" % self.host
def _raw(self, msg):
self.device.send(msg)
def __del__(self):
""" Close TCP connection """
self.device.close()