python – Page 2 – Acodemics

Tagpython

January 14, 2014

The Python Subprocess module is a highly flexible way of running background processes in Python. Here are two functions that show how to run external shell commands, one will wait for the command to return and the other runs in the background allowing the controlling script to stop it as required.

Passing unsanitized input to these functions will not end well.

import subprocess

def background_run_command(cmd):
    """Runs a bash command in the background, returns the Popen instance, does not block."""
    return subprocess.Popen( [ cmd ], shell=True, executable='/bin/bash')

def run_command(cmd):
    """ Runs an external command in bash, waits until the process has returned and returns output/ return code."""
    proc = subprocess.Popen( [ cmd ], shell=True, executable='/bin/bash', stdout=subprocess.PIPE, stderr=subprocess.STDOUT )
    proc.wait()
    return (proc.returncode, proc.stdout)

import subprocess

def background_run_command(cmd):

"""Runs a bash command in the background, returns the Popen instance, does not block."""

return subprocess.Popen( [ cmd ], shell=True, executable='/bin/bash')

def run_command(cmd):

""" Runs an external command in bash, waits until the process has returned and returns output/ return code."""

proc = subprocess.Popen( [ cmd ], shell=True, executable='/bin/bash', stdout=subprocess.PIPE, stderr=subprocess.STDOUT )

proc.wait()

return (proc.returncode, proc.stdout)

Generic CSV to Plot With Numpy and Python

November 8, 2013 / Jack Concanon / 0 Comments

Given multiple CSV files containing a timestamp column and some plottable values (int or float) this script will plot multiple columns of data by time.

Usage is as such – ./CSVPlot.py first.csv second.csv third.csv

#!/usr/bin/python

import numpy
from matplotlib import pyplot as plt
from datetime import datetime
import sys

timestamps = []
legend = []

for arg in sys.argv[1:]:
    data=numpy.genfromtxt(arg, delimiter=",", dtype=None, names=True)
    for columnName, dataType in data.dtype.descr:
        if not isinstance(data[columnName][0], basestring):
            if columnName.strip() != 'timestamp':
                timestamps = [datetime.fromtimestamp(x) for x in data['timestamp']]
                plt.plot(timestamps, data[columnName])
                legend.append(columnName)

plt.legend(legend, loc='upper right')
#plt.savefig('test.png')
plt.xticks(rotation=70)
plt.show()

#!/usr/bin/python

import numpy

from matplotlib import pyplot as plt

from datetime import datetime

import sys

timestamps = []

legend = []

for arg in sys.argv[1:]:

data=numpy.genfromtxt(arg, delimiter=",", dtype=None, names=True)

for columnName, dataType in data.dtype.descr:

if not isinstance(data[columnName][0], basestring):

if columnName.strip() != 'timestamp':

timestamps = [datetime.fromtimestamp(x) for x in data['timestamp']]

plt.plot(timestamps, data[columnName])

legend.append(columnName)

plt.legend(legend, loc='upper right')

#plt.savefig('test.png')

plt.xticks(rotation=70)

plt.show()

Time Bound Python Queue

August 29, 2013 / Jack Concanon / 0 Comments

An extension of the standard Python queue that only stores elements for a given number of seconds before being removed. Useful if you don’t know the volume of data being added to the queue but need to limit it in some way.

There is no decrease in performance when removing items.

Without item removal

Queue size: 10000

real 0m10.771s
user 0m0.217s
sys 0m0.028s

With item removal

Queue size: 4659

real 0m10.765s
user 0m0.203s
sys 0m0.028s

#!/usr/bin/python

"""A Time bounded queue, can use any Queue class as a base as long as it follows the standard Python Queue format (http://docs.python.org/2/library/queue.html)."""

from collections import deque
from time import time

from Queue import Queue

class TimeBoundQueue(Queue):

    def __init__(self, max_time_seconds):
        '''Elements older than max_time_seconds will be removed from the queue.'''
        Queue.__init__(self)
        self.timeout = max_time_seconds
        self.timeout_count = 100

    def _put(self, item):
        '''Add element to queue. Creates an internal tuple of (timestamp, element), also triggers removal
            of timed out elements every 100 adds.'''
        self.timeout_count -= 1
        if self.timeout_count <= 0:
            self.__timeout()
            self.timeout_count = 100
        elem = (time(), item)
        self.queue.append(elem)

    def _get(self):
        return self.queue.popleft()[1]

    def __timeout(self):
        '''Remove oldest elements up to time()-max_time_seconds.'''
        valid_timestamp = time() - self.timeout
        completed = False
        while not completed:
            try:
                timestamp, item = self.queue[0]
                if timestamp < valid_timestamp:
                    # Quicker to pop than to remove a value at random position.
                    self.queue.popleft()
                else:
                    completed = True
            except IndexError as e:
                # No elements left
                return

if __name__ == "__main__":

    from time import sleep

    t_queue = TimeBoundQueue(5)

    for i in range(0, 10000):
        t_queue.put(i)
        sleep(0.001)

    print "Queue size: {0}".format(t_queue.qsize())

#!/usr/bin/python

"""A Time bounded queue, can use any Queue class as a base as long as it follows the standard Python Queue format (http://docs.python.org/2/library/queue.html)."""

from collections import deque

from time import time

from Queue import Queue

class TimeBoundQueue(Queue):

def __init__(self, max_time_seconds):

'''Elements older than max_time_seconds will be removed from the queue.'''

Queue.__init__(self)

self.timeout = max_time_seconds

self.timeout_count = 100

def _put(self, item):

'''Add element to queue. Creates an internal tuple of (timestamp, element), also triggers removal

of timed out elements every 100 adds.'''

self.timeout_count -= 1

if self.timeout_count <= 0:

self.__timeout()

self.timeout_count = 100

elem = (time(), item)

self.queue.append(elem)

def _get(self):

return self.queue.popleft()[1]

def __timeout(self):

'''Remove oldest elements up to time()-max_time_seconds.'''

valid_timestamp = time() - self.timeout

completed = False

while not completed:

try:

timestamp, item = self.queue[0]

if timestamp < valid_timestamp:

# Quicker to pop than to remove a value at random position.

self.queue.popleft()

else:

completed = True

except IndexError as e:

# No elements left

return

if __name__ == "__main__":

from time import sleep

t_queue = TimeBoundQueue(5)

for i in range(0, 10000):

t_queue.put(i)

sleep(0.001)

print "Queue size: {0}".format(t_queue.qsize())

SQLite Database to CSV

August 23, 2013 / Jack Concanon / 0 Comments

Wrote a Python script that can dump a SQLite database to CSV, it takes the column names as the header and then writes values below. It also has a –sumarise option that can create a one line CSV file of multiple lines in a database.

Does have the option to support XML output in the future but a format would need to be defined by the user.

#!/usr/bin/python

import sqlite3
import csv
from optparse import OptionParser

class SQLiteDumper():

    def __init__(self, database_path, statistic_type, output_file_name, output_format, summarise):
        self.databaseLocation = database_path
        self.statisticType = statistic_type
        self.outputFormat = output_format
        self.summarise = summarise
        self.outputFileName = output_file_name
        self.table = self.__typeToTable()
        self.__connectToDatabase()
        self.haveWrittenHeader = False
        self.notToSummarise = ['timestamp', 'component', 'some_string']
        self.csvFile = None
        try:
            self.outputFile = open(self.outputFileName, "w+")
            if self.outputFormat == "CSV":
                self.csvFile = csv.writer(self.outputFile)
        except IOError as e:
            print e

    def __typeToTable(self):
        types = { "Views" : "views",
                    "Errors" : "errors"}
        return types[self.statisticType]

    def __connectToDatabase(self):
        self.dbConnection = sqlite3.connect(self.databaseLocation, detect_types=sqlite3.PARSE_DECLTYPES)

    def writeToCSV(self, results):
        if not self.haveWrittenHeader:
            self.csvFile.writerow(results.keys())
            self.haveWrittenHeader = True
        self.csvFile.writerow(results.values())

    def writeToXML(self, results):
        # This requires a bit more clarification about the output format.
        raise NotImplementedError("Writing to XML support currently not available.")

    def __write(self, result):
        if self.outputFormat == "CSV":
            self.writeToCSV(result)
        elif self.outputFormat == "XML":
            self.writeToXML(result)

    def processDatabase(self):
        cursor = self.dbConnection.cursor()
        result = {}
        for row in cursor.execute('SELECT * FROM {0} ORDER BY timestamp ASC'.format(self.table)):
            for i, key in enumerate(cursor.description):
                if self.summarise:
                    if result.has_key(key[0]) and key[0] not in self.notToSummarise:
                        result[key[0]] += row[i]
                    else:
                        result[key[0]] = row[i]
                else:
                    result[key[0]] = row[i]
            if not self.summarise:
                self.__write(result)
        if self.summarise:
            self.__write(result)

def getOpts():
    parser = OptionParser()
    parser.add_option("-o", "--outputFile", dest="output_file_name", action="store", default="output.csv", type="string", help="File name to store output information")
    parser.add_option("-t", "--statisticType", dest="statistic_type", action="store", default="Views", choices=["Views", "Errors"], help="Statistics type [Views, Errors]")
    parser.add_option("-d", "--database", dest="sqlite_db", action="store", type="string", help="Location of the SQLite database to read.")
    parser.add_option("-f", "--outputFormat", dest="output_format", action="store", default="CSV", choices=["XML", "CSV"], help="Output format [XML, CSV]")
    parser.add_option("-s", "--summarise", dest="summarise", action="store_true", default=False, help="Summarise all results into a single result.")

    (options, args) = parser.parse_args()

    if not options.sqlite_db:
        parser.error("--database argument is required.")

    return options

if __name__ == "__main__":

    opts = getOpts()

    dumper = SQLiteDumper(opts.sqlite_db, opts.statistic_type, opts.output_file_name, opts.output_format, opts.summarise)

    try:
        dumper.processDatabase()
    except Exception as e:
        print e

#!/usr/bin/python

import sqlite3

import csv

from optparse import OptionParser

class SQLiteDumper():

def __init__(self, database_path, statistic_type, output_file_name, output_format, summarise):

self.databaseLocation = database_path

self.statisticType = statistic_type

self.outputFormat = output_format

self.summarise = summarise

self.outputFileName = output_file_name

self.table = self.__typeToTable()

self.__connectToDatabase()

self.haveWrittenHeader = False

self.notToSummarise = ['timestamp', 'component', 'some_string']

self.csvFile = None

try:

self.outputFile = open(self.outputFileName, "w+")

if self.outputFormat == "CSV":

self.csvFile = csv.writer(self.outputFile)

except IOError as e:

print e

def __typeToTable(self):

types = { "Views" : "views",

"Errors" : "errors"}

return types[self.statisticType]

def __connectToDatabase(self):

self.dbConnection = sqlite3.connect(self.databaseLocation, detect_types=sqlite3.PARSE_DECLTYPES)

def writeToCSV(self, results):

if not self.haveWrittenHeader:

self.csvFile.writerow(results.keys())

self.haveWrittenHeader = True

self.csvFile.writerow(results.values())

def writeToXML(self, results):

# This requires a bit more clarification about the output format.

raise NotImplementedError("Writing to XML support currently not available.")

def __write(self, result):

if self.outputFormat == "CSV":

self.writeToCSV(result)

elif self.outputFormat == "XML":

self.writeToXML(result)

def processDatabase(self):

cursor = self.dbConnection.cursor()

result = {}

for row in cursor.execute('SELECT * FROM {0} ORDER BY timestamp ASC'.format(self.table)):

for i, key in enumerate(cursor.description):

if self.summarise:

if result.has_key(key[0]) and key[0] not in self.notToSummarise:

result[key[0]] += row[i]

else:

result[key[0]] = row[i]

else:

result[key[0]] = row[i]

if not self.summarise:

self.__write(result)

if self.summarise:

self.__write(result)

def getOpts():

parser = OptionParser()

parser.add_option("-o", "--outputFile", dest="output_file_name", action="store", default="output.csv", type="string", help="File name to store output information")

parser.add_option("-t", "--statisticType", dest="statistic_type", action="store", default="Views", choices=["Views", "Errors"], help="Statistics type [Views, Errors]")

parser.add_option("-d", "--database", dest="sqlite_db", action="store", type="string", help="Location of the SQLite database to read.")

parser.add_option("-f", "--outputFormat", dest="output_format", action="store", default="CSV", choices=["XML", "CSV"], help="Output format [XML, CSV]")

parser.add_option("-s", "--summarise", dest="summarise", action="store_true", default=False, help="Summarise all results into a single result.")

(options, args) = parser.parse_args()

if not options.sqlite_db:

parser.error("--database argument is required.")

return options

if __name__ == "__main__":

opts = getOpts()

dumper = SQLiteDumper(opts.sqlite_db, opts.statistic_type, opts.output_file_name, opts.output_format, opts.summarise)

try:

dumper.processDatabase()

except Exception as e:

print e

Python Socketed Receiver

March 14, 2013 / Jack Concanon / 0 Comments

This is a quite cool little class I wrote for receiving log messages over sockets. It is launched as a thread and uses a python Queue to store messages for retrieval by consumers in your main program. It also means I can have multiple sockets for different log sources.

#!/bin/python2.6

import threading
from Queue import Queue
import socket

class LoggingReceiver(threading.Thread):

    def __init__(self, host, port):
        threading.Thread.__init__(self)
        self.host = host
        self.port = port
        self.msg_queue = Queue()
        self.running = True
        self.max_waiting_connections = 10
        self.daemon = True
        self.debug = True

    def run(self):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind((self.host, self.port))
        s.listen(self.max_waiting_connections)
        while self.running:
            accepted_socket, address = s.accept()
            self.debug_msg('Received connection from {0}'.format(address))
            msg = ""
            while self.running:
                data = accepted_socket.recv(1024)
                if not data: 
                    break
                msg += data
            self.add_message(msg, address)
        s.close()

    def debug_msg(self, msg):
        if self.debug:
            print msg

    def add_message(self, msg, source):
        self.debug_msg('{0} sent message - {1}'.format(source, msg))
        self.msg_queue.put(msg)

    def get_message(self):
        if not self.msg_queue.empty():
            msg = self.msg_queue.get()
            self.msg_queue.task_done()
            return msg

    def flush_queue(self):
        while not self.queue.empty():
            self.msg_queue.get(block=True)
            self.msg_queue.task_done()

    def stop(self):
        self.debug_msg('Stopping main loop.')
        self.running = False

#!/bin/python2.6

import threading

from Queue import Queue

import socket

class LoggingReceiver(threading.Thread):

def __init__(self, host, port):

threading.Thread.__init__(self)

self.host = host

self.port = port

self.msg_queue = Queue()

self.running = True

self.max_waiting_connections = 10

self.daemon = True

self.debug = True

def run(self):

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

s.bind((self.host, self.port))

s.listen(self.max_waiting_connections)

while self.running:

accepted_socket, address = s.accept()

self.debug_msg('Received connection from {0}'.format(address))

msg = ""

while self.running:

data = accepted_socket.recv(1024)

if not data:

break

msg += data

self.add_message(msg, address)

s.close()

def debug_msg(self, msg):

if self.debug:

print msg

def add_message(self, msg, source):

self.debug_msg('{0} sent message - {1}'.format(source, msg))

self.msg_queue.put(msg)

def get_message(self):

if not self.msg_queue.empty():

msg = self.msg_queue.get()

self.msg_queue.task_done()

return msg

def flush_queue(self):

while not self.queue.empty():

self.msg_queue.get(block=True)

self.msg_queue.task_done()

def stop(self):

self.debug_msg('Stopping main loop.')

self.running = False

Correct Setup of Python Easy-Install

August 11, 2012 / Jack Concanon / 1 Comment

Starting work on some python driver development today for the ant+ system (more on this later) and had to install a bunch of Python modules that didn’t appear on the standard Ubuntu repos.

Python has a very easy package manager of it’s own called easy tools which is found in the python-setuptools package. But this isn’t everything you need, you’ll receive a lot of errors when trying to use easy tools such as –

msgpack/_msgpack.c:4:20: fatal error: Python.h: No such file or directory
compilation terminated.
error: Setup script exited with error: command ‘gcc’ failed with exit status 1

In order to resolve these errors you’ll need to ensure you have g++ installed as well as the python header and static library files. These two packages are called g++ and python2.7-dev. Change the python version depending on your preference or download the python-all-dev package.

Language Detection Algorithm

May 27, 2012 / Jack Concanon / 0 Comments

My thesis required dealing with large quantities of Twitter data and to make it easier for myself I decided to only use English language tweets. Due to the ‘interesting’ grammar and spelling that is frequently found on line a different algorithm to a standard dictionary test was required. This one checks for common English n-grams (letter groupings) and returns a value of how sure the input is English.

Here’s some examples:

Good morning everyone! : 66%
Im So Sick. Really in a bad Position : 50%
Thats the fear of unicorns : 100%
Gnt too com uma saudade daminha namorada vcs nao tem ideia :(((( : 0%
Ils testent les hologrammes pour faire des concerts par des morts. Bientôt même les pas morts feront des hologrammes et vous irez les voir. : 12%
il7ain ilnass ykhl9un men habat twitter o yntqlun to istagram=))!!! : 40%

from collections import Counter

class trigram:
    ngrams = ['the', 'and', 'tha', 'ent', 'ing', 'ion', 'tio', 'for',
        'nde', 'has', 'nce', 'edt', 'tis', 'oft', 'sth', 'men', ' th', 
        'he ', 'ed ', ' of', 'in ', 'to ', 'er ', 'as ', 'her', 'ng ',
        'of ', ' an', ' in']
    probability = 0.0

    def __init__(self):
        pass

    def parse_text(self, text):
        count = 0.0
        s = set()
        length = len(text.strip(' ').split())
        for i in range( len( text ) ): s.add( text[ i: i + 3 ] )
        c = Counter(s)
        for trigram in self.ngrams:
            try:
                count += c[trigram]
            except IndexError as e:
                pass
        if length > 0:
            self.probability = (count / length) * 100.0
        else:
            self.probability = 0
        return self.probability

    def get_probability(self):
        return self.probability

    def is_english(self):
        return (self.probability > 40)

from collections import Counter

class trigram:

ngrams = ['the', 'and', 'tha', 'ent', 'ing', 'ion', 'tio', 'for',

'nde', 'has', 'nce', 'edt', 'tis', 'oft', 'sth', 'men', ' th',

'he ', 'ed ', ' of', 'in ', 'to ', 'er ', 'as ', 'her', 'ng ',

'of ', ' an', ' in']

probability = 0.0

def __init__(self):

pass

def parse_text(self, text):

count = 0.0

s = set()

length = len(text.strip(' ').split())

for i in range( len( text ) ): s.add( text[ i: i + 3 ] )

c = Counter(s)

for trigram in self.ngrams:

try:

count += c[trigram]

except IndexError as e:

pass

if length > 0:

self.probability = (count / length) * 100.0

else:

self.probability = 0

return self.probability

def get_probability(self):

return self.probability

def is_english(self):

return (self.probability > 40)

Python Feed Forward Perceptron

May 27, 2012 / Jack Concanon / 1 Comment

I thought I would post a couple of useful bits of code from my final year of university.

The first is a feed forward perceptron written in Python. I tried to expand the examples on the internet as much as possible to help me understand how they work.

#!/usr/bin/python2.7

import random

class Perceptron:

    def __init__(self, weights=None, threshold=0.5):
        self.weights = weights
        self.threshold = threshold

    def activation(self, input_vector):
        # activation of the perceptron. 
        val = sum(value * self.weights[index] for index, value in enumerate(input_vector))
        if val > self.threshold:
            return 1
        else:
            return 0

    # learning rate is also the alpha
    def train(self, training_set, learning_rate = 0.1):
        # set the initial weights to random numbers in the range [-0.5,0.5] or 0
        self.weights = [0 for i in training_set[0][0]]

        while True:
            error_count = 0
            for input_vector, desired_output in training_set:
                print self.weights
                result = self.activation(input_vector)
                error = desired_output - result
                if error != 0:
                    error_count += 1
                    # weight training
                    for index, value in enumerate(input_vector):
                        #delta rule : learning rate * error * current value                    
                        self.weights[index] += (learning_rate * error * value)
            if error_count == 0:
                break

    def test(self, training_set):
        for input_vector, desired_result in training_set:
            if(self.activation(input_vector) != desired_result):
                return False
        return True

if __name__ == "__main__":

    # values for a AND gate.
    training_set_AND = [((0, 0), 0),
    ((0, 1), 0),
    ((1, 0), 0),
    ((-1, 1), 1)]

    training_set_OR = [((0,1), 1),
                       ((1,0), 1),
                       ((0,0), 0),
                       ((1,1), 1)]

    # values for a NAND gate.
    training_set_NAND = [((1, 0, 0), 1), 
                        ((1, 1, 0), 1), 
                        ((1, 0, 1), 1),
                        ((1, 1, 1), 0)
                        ]

    p = Perceptron(None, 0.5)
    p.train(training_set_AND, 0.1)
    print p.test(training_set_AND)

#!/usr/bin/python2.7

import random

class Perceptron:

def __init__(self, weights=None, threshold=0.5):

self.weights = weights

self.threshold = threshold

def activation(self, input_vector):

# activation of the perceptron.

val = sum(value * self.weights[index] for index, value in enumerate(input_vector))

if val > self.threshold:

return 1

else:

return 0

# learning rate is also the alpha

def train(self, training_set, learning_rate = 0.1):

# set the initial weights to random numbers in the range [-0.5,0.5] or 0

self.weights = [0 for i in training_set[0][0]]

while True:

error_count = 0

for input_vector, desired_output in training_set:

print self.weights

result = self.activation(input_vector)

error = desired_output - result

if error != 0:

error_count += 1

# weight training

for index, value in enumerate(input_vector):

#delta rule : learning rate * error * current value

self.weights[index] += (learning_rate * error * value)

if error_count == 0:

break

def test(self, training_set):

for input_vector, desired_result in training_set:

if(self.activation(input_vector) != desired_result):

return False

return True

if __name__ == "__main__":

# values for a AND gate.

training_set_AND = [((0, 0), 0),

((0, 1), 0),

((1, 0), 0),

((-1, 1), 1)]

training_set_OR = [((0,1), 1),

((1,0), 1),

((0,0), 0),

((1,1), 1)]

# values for a NAND gate.

training_set_NAND = [((1, 0, 0), 1),

((1, 1, 0), 1),

((1, 0, 1), 1),

((1, 1, 1), 0)

]

p = Perceptron(None, 0.5)

p.train(training_set_AND, 0.1)

print p.test(training_set_AND)

Free Twitter Data Set

May 1, 2012 / Jack Concanon / 1 Comment

I needed a twitter data set for my thesis and struggled to find one that was freely available. I ended up downloading the data I needed so I thought I would release it here. It does come in a slightly annoying format which is a MongoDB dump file but it should be easy enough to extract to there and use it. The structure of each record is shown below.

So if you wanted to go over all records in the database you would do it like this –

from pymongo import Connection

if __name__ == "__main__":

    # connect to mongodb and get tweets collection
    db_connection = Connection()
    fyp_db = db_connection.fyp
    tweets_db = fyp_db.tweets
    num_tweets = 0

    for user_col in tweets_db.find():
        num_tweets += len(user_col['tweets'])
        for friend in user_col['friends']:
            if len(user_col['friends'][friend]) > 0:
                num_tweets += len(user_col['friends'][friend])

from pymongo import Connection

if __name__ == "__main__":

# connect to mongodb and get tweets collection

db_connection = Connection()

fyp_db = db_connection.fyp

tweets_db = fyp_db.tweets

num_tweets = 0

for user_col in tweets_db.find():

num_tweets += len(user_col['tweets'])

for friend in user_col['friends']:

if len(user_col['friends'][friend]) > 0:

num_tweets += len(user_col['friends'][friend])

Download Twitter Mongo Dump

Code Golf Progress Bar

April 3, 2012 / Jack Concanon / 0 Comments

Had a go at some code golf today, went for this challenge: http://codegolf.stackexchange.com/questions/5382/create-an-ascii-progress-bar

Here is my solution which is 125 characters small.

a,b=raw_input().split()
a=int(a)
b=float(b)
c=int(a*b)
d="|"*c+" "*(a-c)
e=len(d)/2
print"["+d[:e]+`int(b*100)`+"%"+d[e:]+"]"

a,b=raw_input().split()

a=int(a)

b=float(b)

c=int(a*b)

d="|"*c+" "*(a-c)

e=len(d)/2

print"["+d[:e]+`int(b*100)`+"%"+d[e:]+"]"

So the output for this would be as such:

jack@Icarus:~/Projects/code golf$ python progress_bar.py 
10 0.7
[|||||70%||   ]
jack@Icarus:~/Projects/code golf$ python progress_bar.py 
10 0.56
[|||||56%     ]
jack@Icarus:~/Projects/code golf$ python progress_bar.py 
10 0.8
[|||||80%|||  ]

jack@Icarus:~/Projects/code golf$ python progress_bar.py

10 0.7

[|||||70%|| ]

jack@Icarus:~/Projects/code golf$ python progress_bar.py

10 0.56

[|||||56% ]

jack@Icarus:~/Projects/code golf$ python progress_bar.py

10 0.8

[|||||80%||| ]

Pi Approximator in Python Using Liebniz Formula

March 22, 2012 / Jack Concanon / 0 Comments

Calculates Pi to an increasing degree of accuracy using Leibniz formula. The more cycles that it runs for then the greater the accuracy of the result.

num_cycles = raw_input("Please enter the number of cycles for estimating Pi: ")
pi = 0.0
sign = 1
denom = 1.0

for i in range(0, int(num_cycles)):
    pi += sign/denom
    sign *= -1
    denom += 2

pi *= 4.0

print "Pi approximation is: " + str(pi)

num_cycles = raw_input("Please enter the number of cycles for estimating Pi: ")

pi = 0.0

sign = 1

denom = 1.0

for i in range(0, int(num_cycles)):

pi += sign/denom

sign *= -1

denom += 2

pi *= 4.0

print "Pi approximation is: " + str(pi)

Python Watchman System Monitor

July 9, 2011 / Jack Concanon / 0 Comments

A ‘live’ console based systems monitor in Python. Can be easily extended to monitor various other things. Currently only monitors disk usage, memory usage and zombie processes. Can provide advice for a user. Written to aid some non technical system users. Update: Added percentage bars and tidied up. Also added a CPU monitor.

#!/usr/bin/python

import sys
import os
import subprocess
import time
import re

class MessageStore:
    def __init__(self, id):
        self._id = id
        self._buffer = ''
        self._recommendation = ''
        self._last_update = 0

    def add_message(self, msg, level):
        self._buffer = '%s [%15s] %s' % (self.__colourise_level(level), self._id, msg)
        self.last_update = time.strftime('%a %H:%M:%S')

    def add_recommendation(self, msg):
        self._recommendation = msg

    def percentage_bar(self, percentage):
        percent_to_fill = int(percentage/10)
        return '[%s%s]' % ("+" * percent_to_fill*2, " " * int(10-percent_to_fill)*2)

    def __colourise_level(self,level):

        _red = '\033[1;37;41m'
        _green = '\033[1;37;42m'
        _yellow = '\033[1;37;43m'
        _no_colour = '\033[0m'

        if level == 'error':
            return '%s SERIOUS %s' % (_red, _no_colour)
        elif level == 'warning':
            return '%s WARNING %s' % (_yellow, _no_colour)
        else:
            return '%s    OK   %s' % (_green, _no_colour)

    def get_last_update_time(self):
        return self._last_update

    def get_recommendation(self):
        return self._recommendation

    def get_buffer(self):
        return self._buffer

class MemoryMonitor(MessageStore):

    def __init__(self, target, interval, limit_mb):
        MessageStore.__init__(self, 'Memory Monitor')
        self.target = target
        self.interval = interval
        self.last_update_t = 0
        self.limit_mb = int(limit_mb)

    def update(self):
        if time.time() &gt;= self.last_update_t + self.interval:
            r = self.get_memory_usage()
            self.add_message(r[0], r[1])
            self.add_recommendation(r[2])
            self.last_update_t = time.time()

    def generate_recommendation(self, level):
        recommendation = ''
        if level == 'error':
            recommendation = 'High memory usage detected, consider restarting process (pkill )'

        return recommendation

    def get_memory_usage(self):
        p = subprocess.Popen("pgrep -f \"%s\" | xargs -n 1 pmap" % self.target, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE);
        (stdout, stderr) = p.communicate()
        sum = 0.0
        for line in stdout.splitlines():
            if "heap" in line or "stack" in line or "anon" in line:
                #if "dev:" in line:
                heap = re.split(" +", line)
                sum = sum + int(heap[1].replace('K',''))
        if sum/1024 &gt;= self.limit_mb:
            level = 'error'
        elif sum/1024 &gt;= self.limit_mb*0.9:
            level = 'warning'
        else:
            level = 'ok'
        #msg = '%15s %9.1fM [%s]' % (self.target, sum/1024, "+" * int(sum/102400))
        msg = '%15s %9.1fM %s' % (self.target, sum/1024, self.percentage_bar((sum/1024)/self.limit_mb*100))
        return [msg, level, self.generate_recommendation(level)]

class DiskMonitor(MessageStore):

    def __init__(self, target, interval, limit_perc):
        MessageStore.__init__(self, 'Disk Monitor')
        self.target = target
        self.interval = interval
        self.last_update_t = 0
        self.limit_perc = int(limit_perc)

    def update(self):
        if time.time() &gt;= self.last_update_t + self.interval:
            r = self.get_disk_usage()
            self.add_message(r[0], r[1])
            self.add_recommendation(r[2])
            self.last_update_t = time.time()

    def generate_recommendation(self, level):
        recommendation = ''
        if level == 'error':
            recommendation = 'You got a problem.'
        elif level == 'warning':
            recommendation = 'Better watch out.'    
        return recommendation

    def get_disk_usage(self):
        p = subprocess.Popen("df -h %s | tail -1 | awk '{print $5}'" % self.target, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE);
        (stdout, stderr) = p.communicate()
        usage_perc = int(stdout.strip()[0:-1])
        recommendation = ''
        if usage_perc &gt;= self.limit_perc:
            level = 'error'
        elif usage_perc &gt;= self.limit_perc*0.9:
            level = 'warning'
        else:
            level = 'ok'
        msg = '%15s   %7d%% %s' % (self.target, usage_perc, self.percentage_bar(usage_perc))
        return [msg, level, self.generate_recommendation(level)]

class ZombieProcesses(MessageStore):

    def __init__(self, interval):
        MessageStore.__init__(self, 'Zombie Monitor')
        self.interval = interval
        self.last_update_t = 0

    def update(self):
        if time.time() &gt;= self.last_update_t + self.interval:
            r = self.find_zombies()
            self.add_message(r[0], r[1])
            self.last_update_t = time.time()

    def find_zombies(self):
        p = subprocess.Popen("ps -ef | grep defunct | grep -v grep | awk '{print $2,$8}'" , shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE);
        (stdout, stderr) = p.communicate()
        lines = []
        recommendation = ''
        if not stdout == '':
            lines = stdout.strip().split("\n")
        if len(lines) &gt; 0:
            level = 'warning'
        else:
            level = 'ok'
        msg = "              Found %d zombie processes" % len(lines)
        if len(lines) &gt; 0:
            for line in lines:
                print line
                msg += "\n %30s" % line
        return [msg, level, recommendation]

class CPUMonitor(MessageStore):

    def __init__(self, interval, limit):
        MessageStore.__init__(self, 'CPU Monitor')
        self.interval = interval
        self.limit = int(limit)
        self.last_update_t = 0

    def update(self):
        if time.time() &gt;= self.last_update_t + self.interval:
            r = self.get_cpu_usage()
            self.add_message(r[0], r[1])
            self.last_update_t = time.time()

    def generate_recommendation(self, level):
        recommendation = ''
        if level == 'error':
            recommendation = 'Very high CPU percentage detected, a high CPU percentage could affect stability. Try closing unnecessary scripts running on the machine.'
        return recommendation

    def get_cpu_usage(self):
        p = subprocess.Popen("sar 1 | tail -1 | awk '{print $2}'" , shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE);
        (stdout, stderr) = p.communicate()
        cpu_usage = int(stdout.strip())
        recommendation = ''
        if cpu_usage &gt;= self.limit:
            level = 'error'
        elif cpu_usage &gt;= self.limit*0.9:
            level = 'warning'
        else:
            level = 'ok'
        msg = "      CPU usage   %7d%% %s" % (cpu_usage, self.percentage_bar(cpu_usage))

        return [msg, level, self.generate_recommendation(level)]

def clear_screen():
    print "\033[1;1f\033[2K"

if __name__ == "__main__":

    processes = [MemoryMonitor('chrome', 60, 400),
                MemoryMonitor('compiz', 60, 12000),
                DiskMonitor('/tmp', 60, 90),
                DiskMonitor('/home', 120, 80),
                CPUMonitor(20, 80)
                ]
    print "\033[2J"

    while True:
        clear_screen()
        print time.strftime('%a %H:%M:%S')
        last_process_id = ''
        recommendations = []
        for process in processes:
            if process._id != last_process_id:
                print "\n"
                last_process_id = process._id
            process.update()
            if process.get_recommendation() != '':
                recommendations.append(process.get_recommendation())
            print process.get_buffer()
        if len(recommendations):
            print ''
            print '---------------------------------------------'
            print ''
            print ("\n").join(recommendations)
        time.sleep(2)

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

#!/usr/bin/python

import sys

import os

import subprocess

import time

import re

class MessageStore:

def __init__(self, id):

self._id = id

self._buffer = ''

self._recommendation = ''

self._last_update = 0

def add_message(self, msg, level):

self._buffer = '%s [%15s] %s' % (self.__colourise_level(level), self._id, msg)

self.last_update = time.strftime('%a %H:%M:%S')

def add_recommendation(self, msg):

self._recommendation = msg

def percentage_bar(self, percentage):

percent_to_fill = int(percentage/10)

return '[%s%s]' % ("+" * percent_to_fill*2, " " * int(10-percent_to_fill)*2)

def __colourise_level(self,level):

_red = '\033[1;37;41m'

_green = '\033[1;37;42m'

_yellow = '\033[1;37;43m'

_no_colour = '\033[0m'

if level == 'error':

return '%s SERIOUS %s' % (_red, _no_colour)

elif level == 'warning':

return '%s WARNING %s' % (_yellow, _no_colour)

else:

return '%s OK %s' % (_green, _no_colour)

def get_last_update_time(self):

return self._last_update

def get_recommendation(self):

return self._recommendation

def get_buffer(self):

return self._buffer

class MemoryMonitor(MessageStore):

def __init__(self, target, interval, limit_mb):

MessageStore.__init__(self, 'Memory Monitor')

self.target = target

self.interval = interval

self.last_update_t = 0

self.limit_mb = int(limit_mb)

def update(self):

if time.time() >= self.last_update_t + self.interval:

r = self.get_memory_usage()

self.add_message(r[0], r[1])

self.add_recommendation(r[2])

self.last_update_t = time.time()

def generate_recommendation(self, level):

recommendation = ''

if level == 'error':

recommendation = 'High memory usage detected, consider restarting process (pkill )'

return recommendation

def get_memory_usage(self):

p = subprocess.Popen("pgrep -f \"%s\" | xargs -n 1 pmap" % self.target, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE);

(stdout, stderr) = p.communicate()

sum = 0.0

for line in stdout.splitlines():

if "heap" in line or "stack" in line or "anon" in line:

#if "dev:" in line:

heap = re.split(" +", line)

sum = sum + int(heap[1].replace('K',''))

if sum/1024 >= self.limit_mb:

level = 'error'

elif sum/1024 >= self.limit_mb*0.9:

level = 'warning'

else:

level = 'ok'

#msg = '%15s %9.1fM [%s]' % (self.target, sum/1024, "+" * int(sum/102400))

msg = '%15s %9.1fM %s' % (self.target, sum/1024, self.percentage_bar((sum/1024)/self.limit_mb*100))

return [msg, level, self.generate_recommendation(level)]

class DiskMonitor(MessageStore):

def __init__(self, target, interval, limit_perc):

MessageStore.__init__(self, 'Disk Monitor')

self.target = target

self.interval = interval

self.last_update_t = 0

self.limit_perc = int(limit_perc)

def update(self):

if time.time() >= self.last_update_t + self.interval:

r = self.get_disk_usage()

self.add_message(r[0], r[1])

self.add_recommendation(r[2])

self.last_update_t = time.time()

def generate_recommendation(self, level):

recommendation = ''

if level == 'error':

recommendation = 'You got a problem.'

elif level == 'warning':

recommendation = 'Better watch out.'

return recommendation

def get_disk_usage(self):

p = subprocess.Popen("df -h %s | tail -1 | awk '{print $5}'" % self.target, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE);

(stdout, stderr) = p.communicate()

usage_perc = int(stdout.strip()[0:-1])

recommendation = ''

if usage_perc >= self.limit_perc:

level = 'error'

elif usage_perc >= self.limit_perc*0.9:

level = 'warning'

else:

level = 'ok'

msg = '%15s %7d%% %s' % (self.target, usage_perc, self.percentage_bar(usage_perc))

return [msg, level, self.generate_recommendation(level)]

class ZombieProcesses(MessageStore):

def __init__(self, interval):

MessageStore.__init__(self, 'Zombie Monitor')

self.interval = interval

self.last_update_t = 0

def update(self):

if time.time() >= self.last_update_t + self.interval:

r = self.find_zombies()

self.add_message(r[0], r[1])

self.last_update_t = time.time()

def find_zombies(self):

p = subprocess.Popen("ps -ef | grep defunct | grep -v grep | awk '{print $2,$8}'" , shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE);

(stdout, stderr) = p.communicate()

lines = []

recommendation = ''

if not stdout == '':

lines = stdout.strip().split("\n")

if len(lines) > 0:

level = 'warning'

else:

level = 'ok'

msg = " Found %d zombie processes" % len(lines)

if len(lines) > 0:

for line in lines:

print line

msg += "\n %30s" % line

return [msg, level, recommendation]

class CPUMonitor(MessageStore):

def __init__(self, interval, limit):

MessageStore.__init__(self, 'CPU Monitor')

self.interval = interval

self.limit = int(limit)

self.last_update_t = 0

def update(self):

if time.time() >= self.last_update_t + self.interval:

r = self.get_cpu_usage()

self.add_message(r[0], r[1])

self.last_update_t = time.time()

def generate_recommendation(self, level):

recommendation = ''

if level == 'error':

recommendation = 'Very high CPU percentage detected, a high CPU percentage could affect stability. Try closing unnecessary scripts running on the machine.'

return recommendation

def get_cpu_usage(self):

p = subprocess.Popen("sar 1 | tail -1 | awk '{print $2}'" , shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE);

(stdout, stderr) = p.communicate()

cpu_usage = int(stdout.strip())

recommendation = ''

if cpu_usage >= self.limit:

level = 'error'

elif cpu_usage >= self.limit*0.9:

level = 'warning'

else:

level = 'ok'

msg = " CPU usage %7d%% %s" % (cpu_usage, self.percentage_bar(cpu_usage))

return [msg, level, self.generate_recommendation(level)]

def clear_screen():

print "\033[1;1f\033[2K"

if __name__ == "__main__":

processes = [MemoryMonitor('chrome', 60, 400),

MemoryMonitor('compiz', 60, 12000),

DiskMonitor('/tmp', 60, 90),

DiskMonitor('/home', 120, 80),

CPUMonitor(20, 80)

]

print "\033[2J"

while True:

clear_screen()

print time.strftime('%a %H:%M:%S')

last_process_id = ''

recommendations = []

for process in processes:

if process._id != last_process_id:

print "\n"

last_process_id = process._id

process.update()

if process.get_recommendation() != '':

recommendations.append(process.get_recommendation())

print process.get_buffer()

if len(recommendations):

print ''

print '---------------------------------------------'

print ''

print ("\n").join(recommendations)

time.sleep(2)

Theme by Anders Norén — Up ↑

Categories

Meta

Tagpython

Generic CSV to Plot With Numpy and Python

Time Bound Python Queue

SQLite Database to CSV

Python Socketed Receiver

Correct Setup of Python Easy-Install

Language Detection Algorithm

Python Feed Forward Perceptron

Free Twitter Data Set

Code Golf Progress Bar

Pi Approximator in Python Using Liebniz Formula

Python Watchman System Monitor

Recent Comments