Skip to content

Instantly share code, notes, and snippets.

@dgulino
Last active August 24, 2021 14:13
Show Gist options
  • Save dgulino/4750106 to your computer and use it in GitHub Desktop.
Save dgulino/4750106 to your computer and use it in GitHub Desktop.
Python console application that takes a stream of newline delimited characters (lines) and returns the count of these lines every N seconds.Nice way, in combination with grep, to keep track of certain activity in a log
#!/usr/bin/env python
#Copyright 2007 Drew Gulino
##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
import sys, os, getopt, time
from getopt import GetoptError
import threading
class TaskThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
self._finished = threading.Event()
self._interval = 5.0
def setInterval(self, interval):
"""Set the number of seconds we sleep between executing our task"""
self._interval = interval
def shutdown(self):
"""Stop this thread"""
self._finished.set()
def run(self):
while 1:
if self._finished.isSet(): return
self.task()
# sleep for interval or until shutdown
self._finished.wait(self._interval)
def task(self):
"""The task done by this thread - override in subclasses"""
pass
class Print(TaskThread):
"""Say 'hello' once a minute"""
def __init__(self, count, interval, reset):
TaskThread.__init__(self)
self.setInterval(interval)
self._reset = reset
self._count = count
def task(self):
"""This will be executed every 60 seconds"""
# print self._count.total
sys.stdout.write(str(self._count.total))
sys.stdout.write("\n")
sys.stdout.flush()
if self._reset == True:
self._count.total = 0
class Count:
def __init__(self):
self.total = 0
self.time = time.time()
self.oldtime = self.time
def readLine(self,fileObj):
bytes = []
while True:
try:
b = fileObj.read()
except IOError as e:
if e.args[0] == sys.errno.EAGAIN:
break
raise
except KeyboardInterrupt:
break
bytes.append(b)
if b == chr(10) or b == chr(13):
# bytes.append(chr(10))
break
if b == chr(3):
sys.exit()
return ''.join(bytes)
def nonBlockingReadAll(self,fileObj):
bytes = []
while True:
try:
b = fileObj.read(1024)
except IOError as e:
if e.args[0] == sys.errno.EAGAIN:
break
raise
bytes.append(b)
if len(b) < 1024:
break
return ''.join(bytes)
def usage(progname):
print("usage: " + progname)
version()
print("[-h --help]")
print("[-v --version]")
print("[-i --interval (in seconds)]")
print("[-n --noreset (defaults to zero every interval) ]")
def version():
print("version: 0.4")
def main(argv, stdout, environ):
progname = argv[0]
interval = 5.0
reset = True
try:
arglist, args = getopt.getopt(argv[1:], "hnvi:", ["help", "version", "noreset", "interval="])
except GetoptError:
print("Invalid Option!")
usage(progname)
return
# Parse command line arguments
for (field, val) in arglist:
if field in ("-h", "--help"):
usage(progname)
return
if field in ("-v", "--version"):
version()
return
if field in ("-i", "--interval"):
interval = float(val)
if field in ("-n", "--noreset"):
reset = False
count = Count()
print_out = Print(count, interval, reset)
print_out.start()
while 1:
try:
line = sys.stdin.readline()
if not line: break
count.total=count.total + 1
except KeyboardInterrupt:
print("count stopped with keyboard input")
break
print_out.shutdown()
if __name__ == "__main__":
main(sys.argv, sys.stdout, os.environ)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment