File Manager

X7ROOT File Manager

Current Path: /usr/share/system-config-printer

📁 ..
📄 HIG.py(57 B)
📄 HIG.pyc(231 B)
📄 HIG.pyo(231 B)
📄 PhysicalDevice.py(10.69 KB)
📄 PhysicalDevice.pyc(7.72 KB)
📄 PhysicalDevice.pyo(7.72 KB)
📄 SearchCriterion.py(1.32 KB)
📄 SearchCriterion.pyc(982 B)
📄 SearchCriterion.pyo(982 B)
📄 ToolbarSearchEntry.py(6.91 KB)
📄 ToolbarSearchEntry.pyc(6.43 KB)
📄 ToolbarSearchEntry.pyo(6.43 KB)
📄 applet.py(15.71 KB)
📄 applet.pyc(14.03 KB)
📄 applet.pyo(14.03 KB)
📄 asyncconn.py(9.63 KB)
📄 asyncconn.pyc(9.8 KB)
📄 asyncconn.pyo(9.8 KB)
📄 asyncipp.py(26.02 KB)
📄 asyncipp.pyc(20.5 KB)
📄 asyncipp.pyo(20.5 KB)
📄 asyncpk1.py(25.68 KB)
📄 asyncpk1.pyc(20.4 KB)
📄 asyncpk1.pyo(20.4 KB)
📄 authconn.py(18.34 KB)
📄 authconn.pyc(15.74 KB)
📄 authconn.pyo(15.74 KB)
📄 check-device-ids.py(10.9 KB)
📄 check-device-ids.pyc(8.13 KB)
📄 check-device-ids.pyo(8.13 KB)
📄 config.py(1.05 KB)
📄 config.pyc(370 B)
📄 config.pyo(370 B)
📄 cupspk.py(27.35 KB)
📄 cupspk.pyc(18.01 KB)
📄 cupspk.pyo(18.01 KB)
📄 debug.py(1.62 KB)
📄 debug.pyc(1.55 KB)
📄 debug.pyo(1.55 KB)
📄 dnssdresolve.py(4.91 KB)
📄 dnssdresolve.pyc(5.17 KB)
📄 dnssdresolve.pyo(5.17 KB)
📄 errordialogs.py(3.13 KB)
📄 errordialogs.pyc(2.57 KB)
📄 errordialogs.pyo(2.57 KB)
📄 firewallsettings.py(9.25 KB)
📄 firewallsettings.pyc(8.69 KB)
📄 firewallsettings.pyo(8.69 KB)
📄 gtkinklevel.py(4.71 KB)
📄 gtkinklevel.pyc(4.18 KB)
📄 gtkinklevel.pyo(4.18 KB)
📄 gui.py(2.12 KB)
📄 gui.pyc(1.76 KB)
📄 gui.pyo(1.76 KB)
📁 icons
📄 install-printerdriver.py(2.82 KB)
📄 install-printerdriver.pyc(2.42 KB)
📄 install-printerdriver.pyo(2.42 KB)
📄 installpackage.py(1.34 KB)
📄 installpackage.pyc(1.1 KB)
📄 installpackage.pyo(1.1 KB)
📄 jobviewer.py(91.15 KB)
📄 jobviewer.pyc(64.36 KB)
📄 jobviewer.pyo(64.36 KB)
📄 monitor.py(31.96 KB)
📄 monitor.pyc(23 KB)
📄 monitor.pyo(23 KB)
📄 newprinter.py(164.81 KB)
📄 newprinter.pyc(102.42 KB)
📄 newprinter.pyo(102.42 KB)
📄 options.py(18.73 KB)
📄 options.pyc(16.57 KB)
📄 options.pyo(16.57 KB)
📄 optionwidgets.py(8.78 KB)
📄 optionwidgets.pyc(7.94 KB)
📄 optionwidgets.pyo(7.94 KB)
📄 ppdcache.py(7.41 KB)
📄 ppdcache.pyc(6.1 KB)
📄 ppdcache.pyo(6.1 KB)
📄 ppdippstr.py(7.15 KB)
📄 ppdippstr.pyc(5.64 KB)
📄 ppdippstr.pyo(5.64 KB)
📄 ppdsloader.py(10.78 KB)
📄 ppdsloader.pyc(10.22 KB)
📄 ppdsloader.pyo(10.22 KB)
📄 printerproperties.py(78.52 KB)
📄 printerproperties.pyc(51.48 KB)
📄 printerproperties.pyo(51.48 KB)
📄 probe_printer.py(15.24 KB)
📄 probe_printer.pyc(13.82 KB)
📄 probe_printer.pyo(13.82 KB)
📄 pysmb.py(6.8 KB)
📄 pysmb.pyc(5.47 KB)
📄 pysmb.pyo(5.47 KB)
📄 scp-dbus-service.py(19.04 KB)
📄 scp-dbus-service.pyc(21.34 KB)
📄 scp-dbus-service.pyo(21.34 KB)
📄 serversettings.py(21.4 KB)
📄 serversettings.pyc(16.15 KB)
📄 serversettings.pyo(16.15 KB)
📄 smburi.py(3.22 KB)
📄 smburi.pyc(2.61 KB)
📄 smburi.pyo(2.61 KB)
📄 statereason.py(6.09 KB)
📄 statereason.pyc(5.02 KB)
📄 statereason.pyo(5.02 KB)
📄 system-config-printer.py(80.8 KB)
📄 system-config-printer.pyc(57.95 KB)
📄 system-config-printer.pyo(57.95 KB)
📄 timedops.py(8 KB)
📄 timedops.pyc(8 KB)
📄 timedops.pyo(8 KB)
📁 troubleshoot
📁 ui
📄 userdefault.py(6.12 KB)
📄 userdefault.pyc(5.34 KB)
📄 userdefault.pyo(5.34 KB)
📁 xml

Editing: asyncconn.py

#!/usr/bin/python

## 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., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.

import cups
import os

import config
from debug import *
import debug

######
###### A class to keep track of what we're trying to achieve in order
###### to display that to the user if authentication is required.
######
class SemanticOperations(object):
    def __init__ (self):
        self._operation_stack = []

def _begin_operation (self, operation):
        self._operation_stack.append (operation)

def _end_operation (self):
        self._operation_stack.pop ()

def current_operation (self):
        try:
            return self._operation_stack[0]
        except IndexError:
            return None

######
###### Destructible method call.  Required so that all references can
###### be dropped when an asynchronous method call is destroyed.
######
class _AsyncMethodCall:
    def __init__ (self, fn, reply_handler, error_handler, auth_handler):
        self._fn = fn
        self._reply_handler = reply_handler
        self._error_handler = error_handler
        self._auth_handler = auth_handler
        self._destroyed = False
        debugprint ("+%s" % self)

def __del__ (self):
        debugprint ("-%s" % self)

def destroy (self):
        if self._destroyed:
            return

debugprint ("DESTROY: %s" % self)
        self._destroyed = True
        self._reply_handler = None
        self._error_handler = None
        self._auth_handler = None
        self._reply_data = None
        self._error_data = None
        self._auth_data = None

def run (self, *args, **kwds):
        self._reply_data = kwds.get ('reply_handler')
        self._error_data = kwds.get ('error_handler')
        self._auth_data = kwds.get ('auth_handler')
        kwds['reply_handler'] = self.reply_handler
        kwds['error_handler'] = self.error_handler
        kwds['auth_handler'] = self.auth_handler
        debugprint ("%s: calling %s" % (self, self._fn))
        self._fn (*args, **kwds)

def reply_handler (self, *args):
        if not self._destroyed:
            debugprint ("%s: to reply_handler at %s" % (self,
                                                        self._reply_handler))
            self._reply_handler (self, self._reply_data, *args)

def error_handler (self, *args):
        if not self._destroyed:
            debugprint ("%s: to error_handler at %s" % (self,
                                                        self._error_handler))
            self._error_handler (self, self._error_data, *args)

def auth_handler (self, *args):
        if not self._destroyed:
            debugprint ("%s: to auth_handler at %s" % (self,
                                                       self._auth_handler))
            self._auth_handler (self, self.auth_data, *args)

######
###### An asynchronous libcups API using IPP or PolicyKit as
###### appropriate.
######

class Connection(SemanticOperations):
    def __init__ (self, reply_handler=None, error_handler=None,
                  auth_handler=None, host=None, port=None, encryption=None,
                  parent=None, try_as_root=True, prompt_allowed=True):
        super (Connection, self).__init__ ()

self._destroyed = False

# Decide whether to use direct IPP or PolicyKit.
        if host == None:
            host = cups.getServer()
        use_pk = ((host.startswith ('/') or host == 'localhost') and
                  os.getuid () != 0)

def subst_reply_handler (conn, reply):
            self._subst_reply_handler (None, reply_handler, reply)

def subst_error_handler (conn, exc):
            self._subst_error_handler (None, error_handler, exc)

def subst_auth_handler (prompt, conn, method, resource):
            self._subst_auth_handler (None, auth_handler, prompt, method, resource)

if use_pk and try_as_root:
            debugprint ("Using polkit-1 connection class")
            import asyncpk1
            c = asyncpk1.PK1Connection (reply_handler=subst_reply_handler,
                                        error_handler=subst_error_handler,
                                        host=host, port=port,
                                        encryption=encryption,
                                        parent=parent)
            self._conn = c
        else:
            debugprint ("Using IPP connection class")
            import asyncipp
            c = asyncipp.IPPAuthConnection (reply_handler=subst_reply_handler,
                                            error_handler=subst_error_handler,
                                            auth_handler=subst_auth_handler,
                                            host=host, port=port,
                                            encryption=encryption,
                                            parent=parent,
                                            try_as_root=try_as_root,
                                            prompt_allowed=prompt_allowed,
                                            semantic=self)
            self._conn = c

methodtype = type (self._conn.getPrinters)
        instancemethodtype = type (self._conn.getDevices)
        bindings = []
        for fname in dir (self._conn):
            if fname.startswith ('_'):
                continue
            fn = getattr (self._conn, fname)
            if type (fn) != methodtype and type (fn) != instancemethodtype:
                continue
            if not hasattr (self, fname):
                setattr (self, fname, self._make_binding (fn))
                bindings.append (fname)

self._bindings = bindings
        self._methodcalls = []
        debugprint ("+%s" % self)

def __del__ (self):
        debug.debugprint ("-%s" % self)

def destroy (self):
        debugprint ("DESTROY: %s" % self)
        self._destroyed = True
        try:
            self._conn.destroy ()
        except AttributeError:
            pass

for methodcall in self._methodcalls:
            methodcall.destroy ()

for binding in self._bindings:
            delattr (self, binding)

def _make_binding (self, fn):
        return lambda *args, **kwds: self._call_function (fn, *args, **kwds)

def _call_function (self, fn, *args, **kwds):
        methodcall = _AsyncMethodCall (fn,
                                       self._subst_reply_handler,
                                       self._subst_error_handler,
                                       self._subst_auth_handler)
        self._methodcalls.append (methodcall)
        methodcall.run (*args, **kwds)

def _subst_reply_handler (self, methodcall, reply_handler, *args):
        if methodcall:
            methodcall.destroy ()
            i = self._methodcalls.index (methodcall)
            del self._methodcalls[i]
            args = args[1:]
        if reply_handler and not self._destroyed:
            debugprint ("%s: chaining up to %s" % (self, reply_handler))
            reply_handler (self, *args)

def _subst_error_handler (self, methodcall, error_handler, *args):
        if methodcall:
            methodcall.destroy ()
            i = self._methodcalls.index (methodcall)
            del self._methodcalls[i]
            args = args[1:]
        if error_handler and not self._destroyed:
            debugprint ("%s: chaining up to %s" % (self, error_handler))
            error_handler (self, *args)

def _subst_auth_handler (self, methodcall, auth_handler, prompt, method, resource):
        if methodcall:
            methodcall.destroy ()
            i = self._methodcalls.index (methodcall)
            del self._methodcalls[i]
        if auth_handler and not self._destroyed:
            debugprint ("%s: chaining up to %s" % (self, auth_handler))
            auth_handler (prompt, self, method, resource)

def set_auth_info (self, password):
        """Call this from your auth_handler function."""
        self.thread.set_auth_info (password)

if __name__ == "__main__":
    # Demo
    from gi.repository import GObject
    set_debugging (True)
    GObject.threads_init ()

class Test:
        def __init__ (self, quit):
            self._conn = Connection ()
            self._quit = quit
            debugprint ("+%s" % self)

def __del__ (self):
            debug.debugprint ("-%s" % self)

def destroy (self):
            debugprint ("DESTROY: %s" % self)
            self._conn.destroy ()
            if self._quit:
                loop.quit ()

def getDevices (self):
            self._conn.getDevices (reply_handler=self.getDevices_reply,
                                   error_handler=self.getDevices_error)

def getDevices_reply (self, conn, result):
            print conn, result
            self.destroy ()

def getDevices_error (self, conn, exc):
            print repr (exc)
            self.destroy ()

t = Test (False)
    loop = GObject.MainLoop ()
    t.getDevices ()
    t.destroy ()

t = Test (True)
    t.getDevices ()
    loop.run ()

X7ROOT File Manager

Editing: asyncconn.py

Upload File

Create Folder