fifologger is a simple package you can use to log messages to a FIFO under a *nix operating system. It's a basically a simple convenience class that takes the messy parts of connecting to the “write end” of a FIFO and logging to it. This is quite handy if you want to connect and see what is happening with output from your script at different times, particularly with something you've daemonized, but you don't want to deal with logging to an actual file and you don't really care about losing messages.
#!/usr/bin/env python # $Id: fifologger.py 86 2008-12-18 20:10:39Z jay $ import os , time , stat __all__ = ['FifoError' , 'FifoLog'] """ You can use the FifoLog class to log messages to a fifo. Usage is simple: import fifologger fifo = '/var/tmp/test.fifo' l = fifologger.FifoLog(fifo) for i in xrange(1000): l.log('hello world #%d' % i) time.sleep(1) l.close() Then, just simply run: $ cat /var/tmp/test.fifo And you should see the "hello world" messages printed to your term. """ class FifoError(Exception): pass class FifoLog(): """ This class is set up to make it easy to log to a FIFO under a *nix based operating system. This allows you to simply connect to the read end of the pipe whenever you want to hop on a log stream, but you don't have the overhead of rotating log files and reattaching to new logs, etc. Note that if there is not a process connected to the read end of the pipe, all messages sent to this logger will be dropped until a reader connects. """ def __init__(self , fifoPath , showTime=True): self._fifoPath = fifoPath self._fd = -1 self._showTime = showTime def log(self , msg): """ Logs the message to the fifo. Returns True if a reader was connected and it was successfully logged, False otherwise. """ if self._fd < 0: # We don't have an open handle, attempt to open the fifo try: self._openFifo() except FifoError: # FifoErrors are bad, send those all the way up raise except Exception , err: # A simple connection error just means there is no attached # reader, just reset the file descriptor and return False self._fd = -1 return False if self._showTime: # Add a time.ctime() to the beginning of the msg msg = '%s: %s' % (time.ctime() , msg) msg = msg.strip() retries = 0 while retries < 3: # Try to write to the fifo 3 times, then fail try: os.write(self._fd , '%s\n' % msg) except Exception , err: retries += 1 else: return True return False def close(): """ Closes down the write end of the pipe, if it is open """ if self._fd > 0: os.close(self._fd) def isFifo(self , path): """ Returns a bool indicating whether the given path is a fifo """ s = os.stat(path) if s & stat.S_IFIFO: return True return False def _openFifo(self): """ Attempts to open the fifo """ if os.path.exists(self._fifoPath) and \ not self.isFifo(self._fifoPath): # We have a file with the same name as the fifo path, but # it is not a fifo, raise an exception raise FifoError , 'The file, %s, is not a fifo' % self._fifoPath elif not os.path.exists(self._fifoPath): # Nothing exists at the path so create the fifo os.mkfifo(self._fifoPath) # Attempt to open the write end of the pipe self._fd = os.open(self._fifoPath , os.O_WRONLY | os.O_NONBLOCK) if __name__ == '__main__': # Run a test fifo = '/var/tmp/test.fifo' l = FifoLog(fifo) for i in xrange(1000): l.log('hello world #%d' % i) time.sleep(1) l.close()
import fifologger fifo = '/var/tmp/test.fifo' l = fifologger.FifoLog(fifo) for i in xrange(1000): l.log('hello world #%d' % i) time.sleep(1) l.close()
Simply download the script or copy the above into a file and put the file in your
lib/python<version>/site-packages directory. This is usually something to the effect of
/usr/lib/python2.4/site-packages but it varies depending on the distro.