Package zephir :: Package monitor :: Package agents :: Module printers
[hide private]
[frames] | no frames]

Source Code for Module zephir.monitor.agents.printers

 1  # -*- coding: UTF-8 -*- 
 2  ########################################################################### 
 3  # Eole NG - 2007 
 4  # Copyright Pole de Competence Eole  (Ministere Education - Academie Dijon) 
 5  # Licence CeCill  cf /root/LicenceEole.txt 
 6  # eole@ac-dijon.fr 
 7  ########################################################################### 
 8   
 9  """ 
10  Affichages des imprimantes CUPS et de leur état 
11  """ 
12   
13  from twisted.python import log 
14  from twisted.internet import defer 
15  from twisted.internet.utils import getProcessOutput, getProcessValue 
16   
17  from zephir.monitor.agentmanager.agent import Agent 
18  from zephir.monitor.agentmanager.data import HTMLData, TableData 
19  from zephir.monitor.agentmanager import status 
20  from zephir.monitor.agentmanager.util import status_to_img 
21   
22   
23  PRINTER_CHECK = '/usr/bin/lpstat' 
24   
25 -class Printers(Agent):
26
27 - def __init__(self, name, 28 **params):
29 Agent.__init__(self, name, **params) 30 self.table = TableData([ 31 ('status', "état", {'align':'center'}, status_to_img), 32 ('printer', "Imprimante", {'align':'left'}, None), 33 ]) 34 self.data = [self.table]
35
36 - def measure(self):
37 test = getProcessOutput( PRINTER_CHECK, 38 args = ['-p'], 39 env = {'LC_ALL': 'C'}) 40 test.addCallback(self.measure_process) 41 test.addErrback(self.measure_error) 42 return test
43
44 - def measure_process(self, result):
45 l=[] 46 lines = result.splitlines() 47 for line in lines : 48 if line.startswith('printer') : 49 if 'disabled' in line : 50 l.append( { 'printer' : line.split()[1], 51 'status' : 'Off' } ) 52 self.measure_data[line.split()[1]] = 'Off' 53 elif 'is idle' in line : 54 l.append( { 'printer' : line.split()[1], 55 'status' : 'On' } ) 56 self.measure_data[line.split()[1]] = 'On' 57 if l == [] : 58 l.append( { 'printer' : '** Aucune imprimante détectée **', 59 'status' : '---' } ) 60 61 return { 'statistics' : l}
62 63
64 - def measure_error(self, result):
65 return { 'statistics' : [{ 'printer' : '** Mesure impossible **', 'status' : '---' }] }
66 67
68 - def write_data(self):
69 Agent.write_data(self) 70 if self.last_measure is not None: 71 self.table.table_data = self.last_measure.value['statistics']
72 73
74 - def check_status(self):
75 #if self.last_measure is not None: 76 # for service in self.last_measure.value['s']: 77 # if service['status'] != 'On': 78 # return status.Error() 79 #else: 80 # # pas de mesure connue 81 # return status.Unknown() 82 return status.OK()
83