python-foomatic 0.7.9.5build1 / usr / share / pyshared / foomatic / foomatic.py

#!/usr/bin/python
#
# foomatic.py - Pythonic wrappers for foomatic-configure
#
#     foomatic-gui - GNOME2 interface to the foomatic printing system
#     Copyright (C) 2002-04 Chris Lawrence <lawrencc@debian.org>
#
#     This program is free software; you can redistribute it and/or modify
#     it under the terms of the GNU General Public License as published by
#     the Free Software Foundation; either version 2 of the License, or
#     (at your option) any later version.
#
#     This program is distributed in the hope that it will be useful,
#     but WITHOUT ANY WARRANTY; without even the implied warranty of
#     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#     GNU General Public License for more details.
#
#     You should have received a copy of the GNU General Public License
#     along with this program; if not, write to the Free Software
#     Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
#
# $Id: foomatic.py,v 1.47 2010/05/23 00:32:34 lordsutch Exp $

import commands
from commands import mkarg
import glob
import os
import re
import sys
import xml.sax
import pprint    
import urlutils
import urlparse
import gzip

# Only run in debug mode on my machines... cheesy, yes.
DEBUG=os.path.exists('/home/cnlawren') or os.path.exists('/home/quango')

class FoomaticXMLHandler(xml.sax.ContentHandler):
    "Generic parser handler used by the other classes with some sensible defaults."
    def __init__(self):
        xml.sax.ContentHandler.__init__(self)
        self._data = ''
        self._curinfo = self._block = {}
        self._blocks = []

    def characters(self, characters):
        self._data += characters

    def startElement(self, name, attrs):
        self._data = ''

    def endElement(self, name):
        self._block[name] = self._data.strip()

    def endDocument(self):
        del self._block

class FoomaticQueueHandler(FoomaticXMLHandler):
    def __init__(self):
        FoomaticXMLHandler.__init__(self)
        self.queues = []
        self.defqueue = self.spooler = ''
        self._isfoomatic = False

    def startElement(self, name, attrs):
        FoomaticXMLHandler.startElement(self, name, attrs)
        if name == 'queue':
            self._curinfo = self._block = {}
            for (attr, val) in attrs.items():
                if attr == 'foomatic':
                    self._isfoomatic = bool(int(val))
                elif attr == 'spooler':
                    self.spooler = val

    def endElement(self, name):
        if name == 'defaultqueue':
            self.defqueue = self._data.strip()
        elif name == 'queue':
            # Removed the URI remapping code, since Foomatic 3.x does
            # it already
            if 'name' in self._curinfo:
                self._curinfo['_foomatic'] = self._isfoomatic
                self.queues.append(self._curinfo)
            self._curinfo = self._block = {}
        else:
            FoomaticXMLHandler.endElement(self, name)

AUTODETECT_TYPES = ('general', 'parallel', 'usb', 'serial', 'snmp')
class FoomaticOverviewHandler(FoomaticXMLHandler):
    def __init__(self):
        FoomaticXMLHandler.__init__(self)
        self.printers = {}
        self.makes = {}
        self.cmakes = {} # All caps make and model
        self.autodetect_ids = {}
        self._drivers_block = self._inautoblock = False

    def startElement(self, name, attrs):
        FoomaticXMLHandler.startElement(self, name, attrs)
        if name == 'printer':
            self.drivers = []
            self._curinfo = self._block = {}
        elif name == 'drivers':
            self._drivers_block = True
            self.drivers = []
        elif name == 'autodetect':
            self.autodetect = {}
            self._inautoblock = True
        elif name in AUTODETECT_TYPES:
            self._block = {}

    def endElement(self, name):
        if name == 'printer':
            id = self._curinfo.get('id')
            make = self._curinfo.get('make')
            model = self._curinfo.get('model')
            autodetect = self._curinfo.get('autodetect')

            if make and model:
                self.makes.setdefault(make, {})[model] = self._curinfo
                # Also include uppercase make and model for IEEE 1284 ad-hoc
                # detection support
                self.cmakes.setdefault(make.upper(), {})[model.upper()] = self._curinfo

            if id:
                self.printers[id] = self._curinfo

            if autodetect:
                for x in autodetect.values():
                    if x and ('model' in x and 'manufacturer' in x):
                        self.autodetect_ids[(x['manufacturer'], x['model'])] = self._curinfo
                    if x and 'description' in x:
                        self.autodetect_ids[x['description']] = self._curinfo
            #pprint.pprint(self._curinfo)
            self._curinfo = {}
        elif name == 'drivers':
            self._curinfo[name] = self.drivers
            self._drivers_block = False
            del self.drivers
        elif name == 'driver':
            driver = self._data.strip()
            if self._drivers_block:
                self.drivers.append(driver)
            else:
                self._curinfo['driver'] = driver
        elif name == 'autodetect':
            self._curinfo['autodetect'] = self.autodetect
            self._inautoblock = False
            del self.autodetect
        elif name in AUTODETECT_TYPES:
            if not hasattr(self, 'autodetect'):
                self.autodetect = {}
            self.autodetect[name] = self._block
            self._block = self._curinfo
        else:
            FoomaticXMLHandler.endElement(self, name)

class FoomaticPrinterHandler(FoomaticXMLHandler):
    def __init__(self):
        FoomaticXMLHandler.__init__(self)
        self.make = self.model = self.url = self.functionality = self.driver =\
                    self.contrib_url = self.ppd = ''
        self.mechanism = {}
        self.languages = {}
        self.autodetect = {}
        self.comments = {} # Indexed by language
        self.verified = False
        self._inmechanism = False

    def startElement(self, name, attrs):
        FoomaticXMLHandler.startElement(self, name, attrs)
        if name == 'printer':
            pass
##         elif name in ('postscript', 'pcl'):
##             self._block[name] = attrs
        elif name == 'mechanism':
            self._block = {}
            self._inmechanism = True
        elif name in ('dpi', 'consumables', 'comments', 'resolution', 'lang'):
            self._blocks.append(self._block)
            self._block = {}
        elif name == 'autodetect':
            self.autodetect = {}
            self._inautoblock = True
        elif name in AUTODETECT_TYPES:
            self._blocks.append(self._block)
            self._block = {}

    def endElement(self, name):
        if name == 'printer':
            pass
        elif name in ('make', 'model', 'url', 'functionality', 'driver',
                      'contrib_url', 'ppd'):
            setattr(self, name, self._data.strip())
        elif name == 'verified':
            self.verified = True
        elif name == 'mechanism':
            self.mechanism, self._block = self._block, self._curinfo
            self._inmechanism = False
        elif name == 'comments':
            if not self._inmechanism:
                self.comments, self._block = self._block, self._blocks.pop()
            else:
                popped = self._blocks.pop()
                self._block, popped[name] = popped, self._block
        elif name in ('inkjet', 'laser'):
            self._block['type'] = name
        elif name == 'transfer':
            self._block['type'] = 'thermal transfer'
        elif name == 'led':
            self._block['type'] = 'LED'
        elif name == 'dotmatrix':
            self._block['type'] = 'dot-matrix'
        
        elif name == 'color':
            self._block['color'] = True
        elif name in ('resolution', 'dpi', 'consumables'):
            popped = self._blocks.pop()
            self._block, popped[name] = popped, self._block
        elif name == 'autodetect':
            self._curinfo['autodetect'] = self.autodetect
            self._inautoblock = False
            del self.autodetect
        elif name in AUTODETECT_TYPES:
            self._block, self.autodetect[name] = self._blocks.pop(), self._block
        elif name == 'lang':
            self._block, self.languages = self._blocks.pop(), self._block
        elif name in ('x', 'y'):
            try:
                self._block[name] = int(self._data.strip())
            except:
                pass
        else:
            FoomaticXMLHandler.endElement(self, name)

def parse_foomatic(commandline, handler):
    pipe = os.popen(commandline, 'r')
    data = pipe.read()
    pipe.close()
    if data:
        xml.sax.parseString(data, handler)

def get_printer_queues():
    handler = FoomaticQueueHandler()
    try:
        parse_foomatic('foomatic-configure -q -Q -r 2>/dev/null', handler)
    except:
        return None
    return handler

def get_printer_db():
    handler = FoomaticOverviewHandler()
    try:
        parse_foomatic('foomatic-configure -q -O 2>/dev/null', handler)
    except:
        return None
    return handler

# Parse the xml printer and driver information - not done yet
# A wrapper for "foomatic-configure -X -p %s"
def get_printer_info(printer):
    handler = FoomaticPrinterHandler()
    try:
        parse_foomatic('foomatic-configure -X -p' + mkarg(printer) + " 2>/dev/null", handler)
    except:
        return None
    return handler

def get_driver_info(driver):
    handler = FoomaticDriverHandler()
    try:
        parse_foomatic('foomatic-configure -X -d' + mkarg(driver) + ' 2>/dev/null', handler)
    except:
        return None
    return handler

def set_default_queue(queuename):
    mask = os.umask(0755)
    os.system('foomatic-configure -q -D -n' + mkarg(queuename) + ' 2>/dev/null')
    os.umask(mask)

def remove_queue(queuename):
    os.system('foomatic-configure -q -R -n' + mkarg(queuename) + ' 2>/dev/null')

# Should be a dictionary, like those returned in get_printer_queues().queues
def setup_queue(queueinfo):
    if not queueinfo.get('location'):
        queueinfo['location'] = ''
    if not queueinfo.get('description'):
        queueinfo['description'] = ''

    args = [('-n', 'name'),
            ('-N', 'description'),
            ('-L', 'location'),
            ('-c', 'connect')]

    if queueinfo.get('ppdfile'):
        args += [('--ppd', 'ppdfile')]
    else:
        args += [('-p', 'printer'), ('-d', 'driver')]

    if queueinfo.get('spooler'):
	args += [('-s', 'spooler')]   
 
    argstr = ' '.join([ a + mkarg(queueinfo[b] or '') for (a,b) in args])
    command = 'foomatic-configure -f -q -w '+argstr #+' 2>/dev/null'
    if DEBUG:
        print command
    mask = os.umask(0755)
    os.system(command)
    os.umask(mask)

def send_test_page(template, queueinfo):
    SUBSTITUTIONS = {'QUEUENAME' : 'name', 'PRINTER' : 'printer',
                     'DRIVER' : 'driver', 'DESCRIPTION' : 'description',
                     'LOCATION' : 'location', 'CONNECTION' : 'connect'}
    
    tf = file(template, 'r')
    data = tf.read()
    tf.close()

    for (macro, field) in SUBSTITUTIONS.iteritems():
        sub = queueinfo.get(field, 'N/A')
        # Escape any parentheses or backslashes
        sub = sub.replace('\\', '\\\\').replace('(', '\(').replace(')', '\)')
        data = data.replace(macro, sub)

    #command = 'cat > /tmp/testpage.ps'
    command = 'foomatic-printjob -P' + mkarg(queueinfo['name'])
    if DEBUG:
        print command

    try:
        pipe = os.popen(command, 'w')
        pipe.write(data)
        pipe.close()
    except IOError, x:
        return False

_mappingre = re.compile(r'(\'[.0-9A-Za-z_-]*?\') => ', re.DOTALL)
_refre = re.compile(r'{(\'[.0-9A-Za-z_-]*?\')}', re.DOTALL)
_quotere = re.compile(r"'(([^\\]|\\.)*?)'", re.DOTALL)
_xre = re.compile(r"\\x{([a-f0-9]+)}")

def convert_perl_data(content):
    "Convert the output of foomatic-configure -P into Python.  May be ugly."
    content = content.replace('->', '')
    content = content.replace('undef', 'None')
    content = content.replace('$QUEUES[0]', 'QUEUES[0]')
    content = _mappingre.sub(r'\1 : ', content)
    content = _refre.sub(r'[\1]', content)
    content = _xre.sub(r'\\x\1', content)
    # Triple-quote all strings, since Adobe PPDs can contain multiline literals
    content = _quotere.sub(r"'''\1'''", content)
    #content = content.replace("'", "'''")
    return content

def get_printer_options(queueinfo=None, queuename='', debug=False):
    "Return the data for a particular printer queue, converted from Perl."
    if queueinfo:
        queuename = queueinfo['name']
    if not queuename:
        return
    command = 'LANG=C LC_ALL=C foomatic-configure -q -P -n' + mkarg(queuename)+ ' 2>/dev/null'
    try:
        data = commands.getoutput(command)
    except IOError:
        return {}
    pydata = convert_perl_data(data)
    QUEUES = [None]
    if debug:
        open('/tmp/queue.py', 'w').write(pydata)
    exec pydata
    return QUEUES[0]

def organize_printer_options(options):
    "Convert the output of get_printer_options into something a bit more "
    "useful for the GUI."
    newopts = {}
    defaultargs = {}
    labels = {}
    for arg in options['args']:
        # Skip hidden options; they do us little good.
        if arg.get('hidden'): continue

        group = arg.get('group', '')
        grouptrans = arg.get('grouptrans')
        if isinstance(grouptrans, (tuple, list)) and len(grouptrans):
            grouplabel = grouptrans[0]
        else:
            grouplabel = group
        argname = arg['name']
        # Ignore the PageRegion setting
        if argname == 'PageRegion':
            continue
        default = arg['default']
        comment = arg['comment']
        argtype = arg['type']
        if argtype == 'bool':
            if default == '0':
                default = 'False'
            elif default == '1':
                default = 'True'
            
            argvals = [('True', arg['comment_true'] or 'On'),
                       ('False', arg['comment_false'] or 'Off'),]
        elif argtype == 'enum': # or arg['vals']:
            argvals = []
            for val in arg['vals']:
                argvals.append( (val['value'], val['comment']) )
        elif argtype in ('float', 'int'):
            argvals = (float(arg['min']), float(arg['max']))
            if 'fdefault' in arg:
                default = float(arg['fdefault'])
        else:
            # I think 'string' is also a possibility; not sure how
            # to handle that here.
            print >> sys.stderr, 'Unhandled option type:', arg['type']
            continue

        newopts.setdefault(grouplabel, []).append( (argname, argvals) )
        defaultargs[argname] = default
        labels[argname] = comment

    return newopts, defaultargs, labels

def set_printer_options(queueinfo=None, options=None, queuename='',
                        ppdfile=None):
    "Set printer options for the selected queue; options is a dictionary."
    if queueinfo:
        queuename = queueinfo['name']
    if not queuename or not options:
        return

    settings = []
    for (name, val) in options.iteritems():
        settings.append( '-o' + mkarg("%s=%s" % (name, val)) )

    command = 'foomatic-configure -q -n' + mkarg(queuename) + ' ' + (' '.join(settings)) # + ' 2>/dev/null'
    if DEBUG:
        print command
    os.system(command)

def guess_spooler(pqdb=None):
    if not pqdb:
        pqdb = get_printer_queues()

    if pqdb.spooler: return pqdb.spooler

    if os.path.exists('/etc/foomatic/defaultspooler'):
        try:
            return open('/etc/foomatic/defaultspooler', 'r').readline().strip()
        except:
            pass

    # Implement more detection here eventually...
    if os.path.exists('/etc/cups'):
        return 'cups'

    return None

def get_ppd_content(uri, network=True):
    if not uri.startswith('http:'):
        # Newer XML omits the full URI
        if not uri.startswith('PPD/'):
            # No idea what to do...
            print >> sys.stderr, 'Nonstandard PPD location', uri
            return None

        filename = uri[4:]
        filename = os.path.join('/usr/share/ppd/postscript/', uri[4:])
        if os.path.exists(filename):
            try:
                fp = open(filename)
                content = fp.read()
                fp.close()
                return content
            except IOError:
                pass
        if os.path.exists(filename+'.gz'):
            try:
                fp = gzip.open(filename+'.gz')
                content = fp.read()
                fp.close()
                return content
            except IOError:
                pass
        # Try the website now
        uri = urlparse.urljoin('http://openprinting.org/foomatic-db/db/source/', uri)
    
    if not network:
        print >> sys.stderr, 'PPD not available offline:', uri
        return None

    print >> sys.stderr, 'Fetching PPD file from', uri
    fp = urlutils.open_url(uri)
    if fp:
        try:
            content = fp.read()
        except:
            fp.close()
            return None
        fp.close()
        return content
    return None

def test():
    # Assemble queue list
    pqdb = get_printer_queues()
    queues = pqdb.queues
    print 'Spooler:', repr(pqdb.spooler)
    print 'Available printer queues:'
    for queue in queues:
        if 'description' in queue:
            print '  %(name)s: %(description)s' % queue
        else:
            print '  %(name)s' % queue
    print
    print 'Available options for queue %(name)s:' % queue
    x = get_printer_options(queue)
    print organize_printer_options(x)[2].values()

def test3():
    import pprint
    
    pdb = get_printer_db()
    printers, makes, autodetect = pdb.printers, pdb.makes, pdb.autodetect_ids
    print 'HP DeskJet 990C information:'
    info = makes['HP']['DeskJet 990C']
    pid = info['id']
    pprint.pprint(info)
    print pid
    data = get_printer_info(pid)
    pprint.pprint(data.__dict__)
    print 'Canon BJC-2110 information:'
    pprint.pprint(makes['Canon']['BJC-2110'])
    print 'HP LaserJet 2200 information:'
    pprint.pprint(makes['HP']['LaserJet 2200'])
    print 'Autodetect information:'
    pprint.pprint(autodetect['Hewlett-Packard PhotoSmart P1100'])
    #print get_printer_info('HP-LaserJet_2200').__dict__
    #print get_printer_info('Canon-BJC-2110').__dict__

def test2():
    import pprint
    
    pprint.pprint(get_printer_options(None, 'lw8500', True))

if __name__ == '__main__':
    #print guess_spooler()
    test()