Zeilenpuffer als Interface zwischen unabhängigen Modulen
Verfasst: Montag 16. August 2004, 14:23
Hallo,
immer wieder taucht das Problem auf, unterschiedliche Module miteinander kommunizieren zu lassen, z.B. eine Anwendung und ein GUI-Modul. Jetzt hab ich mit eine Linebufferklasse erstellt, die das Ganze vereinfacht.
Line ist eine Dateiartige Klasse, deren Instanzen mittels der Methode 'write' Zeichenketten übergeben werden können, die dann Zeilenweise an Empfänger weitergeleitet werden. Im Beispielcode am Ende des Scripts, wird z.B. sys.stdout als Empfänger definiert dem verschiedene Strings zugeführt werden, genausogut kann auch in einer GUI z.B. der Statuszeile von einer Applikation die Ausgabe mitgeteilt werden.
Mittels Lineobjekten ist auch die Übersetzung von Zeichenketten eines Encodings in ein anderes möglich.
Gruß
Dookie
immer wieder taucht das Problem auf, unterschiedliche Module miteinander kommunizieren zu lassen, z.B. eine Anwendung und ein GUI-Modul. Jetzt hab ich mit eine Linebufferklasse erstellt, die das Ganze vereinfacht.
Code: Alles auswählen
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
"""
Modul: Linebuffer
Description: Linebufferobject as interface for autarc modules,
for example GUIs and Applications and logfiles
Version: 0.2
Copyright: 2004 by Fritz Cizmarov fritz@sol.at
Created: 16. Aug. 2004
Last modified: 18. Aug. 2004
License: free
Requirements: Python2.3
Exports: Line
"""
from array import array
class Line(object):
"""
Buffer for textlines, written to recievers line by line
Line([initvalue,] [reciever1, [reciever2, [...]]] [encoding="utf-8"])
return a linebufferobject initialized by initvalue with any
number of recievers.
Recievers can be filelike objects (have a 'write' function) or
functions that can handle instances of Line or str.
"""
__slots__ = ["__content", "__buffer", "__recievers", "__encoding"]
def encoding(self):
return self.__encoding
encoding = property(encoding, doc="encoding of Linebuffer")
def __init__(self, *args, **kw):
self.__content = ""
self.__buffer = array('c')
self.__encoding = kw.pop("encoding", "utf-8")
self.__recievers = list(args[1:])
if args:
self.write(args[0])
def __del__(self):
if len(self.__buffer):
self.flush()
def __str__(self):
""" return last line as string """
return self.__content
def __unicode__(self):
""" return last line as unicode """
return self.__content.decode(self.encoding)
def write(self, s):
"""
write s to Linebuffer, if s contains linefeeds, split
s into lines and write them line by line
"""
if s:
if isinstance(s, unicode):
s = s.encode(self.encoding)
lfs = s.count('\n')
if lfs > 1 or (lfs == 1 and s[-1] != '\n'):
lines = s.split('\n')
for line in lines[:-1]:
self.write(line+"\n")
self.write(line)
else:
self.__buffer.fromstring(s)
if s[-1] == '\n':
self.flush()
def flush(self):
""" send linebuffer to recievers and clear linebuffer """
self.__content = self.__buffer.tostring()
del self.__buffer[:]
for reciever in self.__recievers:
if hasattr(reciever, "write"):
encoding = getattr(reciever, "encoding", self.encoding)
if encoding and encoding != self.encoding:
data = unicode(self).encode(encoding)
else:
data = self.__content
reciever.write(data)
if hasattr(reciever, "flush"):
reciever.flush()
else:
try:
reciever(self)
except TypeError:
reciever(str(self))
def from_iterable(self, iterable):
""" get lines from iterable, iterable may also be a file """
for line in iterable:
if line[-1] != "\n":
self.write(line+"\n")
else:
self.write(line)
def add_reciever(self, reciever):
""" add an reciever """
self.__recievers.append(reciever)
def remove_reciever(self, reciever):
""" remove an reciever """
self.__recievers.remove(reciever)
if __name__ == "__main__":
from time import sleep
import sys
log = file("Linebuffer.log", 'w')
l = Line("Dies ist ein Test!\n", sys.stdout, log,
encoding=sys.getfilesystemencoding())
l.write("Hallo ")
l.write("Welt!")
l.write("\n")
l.write(u"öäüß\n")
for c in "1234567890\n":
l.write(c)
l.flush()
sleep(1)
log.close()
Mittels Lineobjekten ist auch die Übersetzung von Zeichenketten eines Encodings in ein anderes möglich.
Gruß
Dookie