Implementing a Standard Library

Components

I use the following for building up utility scripts.

$ ls -lR ~/lib/py

/home/pharding/lib/py:
total 4
drwxr-xr-x+ 1 pharding Users 0 Feb 20 11:44 performiq

./performiq:
total 15
-rw-r--r-- 1 pharding Users  177 Feb 20 12:26 __init__.py
-rw-r--r-- 1 pharding Users   23 Feb 20 10:52 CONSTANTS.py
-rw-r--r-- 1 pharding Users 1248 Feb 20 12:44 CSV_Reader.py
-rw-r--r-- 1 pharding Users  361 Feb 20 10:55 Enum.py
-rw-r--r-- 1 pharding Users 2121 Feb 20 11:32 Logger.py
-rw-r--r-- 1 pharding Users 1077 Feb 20 12:08 Timer.py

env | grep PYTHONPATH

PYTHONPATH=/home/pharding/lib/py

__init__.py

$ cat __init__.py

import CONSTANTS

from CONSTANTS import VERSION
from Enum      import Enum
from Logger    import Logger
from Timer     import Timer

CONSTANTS.py

$ cat CONSTANTS.py

VERSION   = "1.0.0"

Enum.py

$ cat Enum.py

#==========================================================================

class Enum(set):
   pass

   #--------------------------------------------------------------------

   def __getattr__(self, name):
      if name in self:
         return name
      raise AttributeError

#==========================================================================

Logger.py

$ cat Logger.py

import os
import sys
import logging

#==========================================================================

class Logger:
    logger = None
    debug  = False

    @classmethod
    def Info(cls, msg):
        global debug_level, verbose_flg

        if not cls.logger:
            cls.Init()

        cls.logger.info(' ' + msg)

        if cls.debug: sys.stderr.write("[%s::INFO]  %s\n" % (cls.name, msg))

    #----------------------------------------------------------------------

    @classmethod
    def Error(cls, msg):
        global debug_level, verbose_flg

        if not cls.logger:
            cls.Init()

        cls.logger.error(msg)

        if cls.debug: sys.stderr.write("[%s::ERROR]  %s\n" % (cls.name, msg))

    #----------------------------------------------------------------------

    @classmethod
    def Warning(cls, msg):
        global debug_level, verbose_flg

        if not cls.logger:
            cls.Init()

        cls.logger.warning('*****' + msg + '*****')

        if cls.debug: sys.stderr.write("[%s::WARNING]  %s\n" % (cls.name, msg))

    #----------------------------------------------------------------------

    @classmethod
    def Init(cls, name='logger', log_dir='/c/temp', debug=False):
        cls.debug = debug
        cls.name  = name
        cls.pid   = os.getpid()

        if cls.debug: sys.stderr.write("[%s::Init] PID is %d\n" % (cls.name, cls.pid))

        cls.log_file = '%s/%s.log' % (log_dir, name)

        try:
            cls.logger  = logging.getLogger(name)
            cls.hdlr    = logging.FileHandler(cls.log_file)
            cls.fmtr    = logging.Formatter('%(asctime)s %(levelname)s %(message)s')

            cls.hdlr.setFormatter(cls.fmtr)
            cls.logger.addHandler(cls.hdlr)
            cls.logger.setLevel(logging.INFO)

            cls.logger.info("===== Started processing %s" % ('=' * 20))
 
            cls.count = 0
        except IOError, msg:
            sys.stderr.write(cls.log_file + ': cannot open: ' + `msg` + '\n')
            sys.exit(1)

#==========================================================================

Timer.py

#==========================================================================

class Timer:
    t_reference = None

    #----------------------------------------------------------------------

    @classmethod
    def init(cls):
       return float(cls.get_reference_time(init=True)) * 0.001

    #----------------------------------------------------------------------

    @classmethod
    def time(cls):
       return float(cls.get_reference_time()) * 0.001

    #----------------------------------------------------------------------

    @classmethod
    def get_reference_time(cls, init=False):
       t_now  = datetime.now()

       if (flg):
          cls.t_reference   = t_now
          t             = 0
       else:
          t_delta       = t_now - cls.t_reference
          t             = ((t_delta.seconds * 1000000) + t_delta.microseconds)/1000.0

       return t

    #----------------------------------------------------------------------

#==========================================================================

CSV_Reader.py

Just so that I can leave it around...

$ cat CSV_Reader.py

import csv
import sys

#=========================================================================

class CSV_Reader:

    #---------------------------------------------------------------------

    @classmethod
    def read(cls, fname, skip=0, obj=None, limit=0):
        csv_data  = []

        try:
            f_in      = open(fname, "rb")
        except IOError, msg:
            sys.stderr.write(cls.log_file + ': cannot open: ' + `msg` + '\n')
            sys.exit(1)

        reader = csv.reader(f_in)

        cnt   = 0

        for row in reader:
           cnt += 1

           if skip > 0:
               skip -= 1
               continue   # Skip headings

           if obj:
               data = obj(row)
           else:
               data = row

           # print data

           csv_data.append(data)

           if limit and (cnt < limit): continue

        f_in.close()  # Explicitly close the file *NOW*

        no_lines  = len(csv_data)

        print "Read %d data items..." % no_lines
        print "                 -> Total data  %d" % cnt

        return csv_data

    #---------------------------------------------------------------------

#=========================================================================

A Test Harness

Use this to test the performiq module...

$ cat tst_module.py 
#!/usr/bin/env python

import time

import performiq

from performiq import Logger

debug_level  = 0
verbose_flg  = True

#==========================================================================

class Data:

    #--------------------------------------------------------------------

    def __init__(self, row):
        self.One = row[0]
        self.Two = row[1]
        self.Three = row[2]
        self.Four = row[3]

    #--------------------------------------------------------------------

    def __str__(self):
        return "%s,%s,%s,%s" % (self.One, self.Two, self.Three, self.Four)

    #--------------------------------------------------------------------

#==========================================================================


print "performiq version [%s]" % performiq.VERSION


print "Set timer - %5.3f" % performiq.Timer.init()

Logger.Init(debug=True)

Logger.Info("Testing - Info...")
Logger.Warning("Testing - Warning...")
Logger.Error("Testing - Error...")

time.sleep(1.4)

print "Get timer - %5.3f" % performiq.Timer.time()

csv_data = """\
a,b,c,d
e,f,g,h
i,j,k,l
"""

csv_file = 'tst.csv'

f_out = open(csv_file, 'w+')

f_out.write(csv_data)
f_out.close()

data = performiq.CSV_Reader.read(csv_file, obj=Data)

if data:
    for d in data:
        print d

Running it:

$ ./tst_module.py 
performiq version [1.0.0]
Set timer - 0.000
[logger::Init] PID is 8404
[logger::INFO]  Testing - Info...
[logger::WARNING]  Testing - Warning...
[logger::ERROR]  Testing - Error...
Get timer - 3.001
Read 3 data items...
                 -> Total data  3
a,b,c,d
e,f,g,h
i,j,k,l

A Variant skel.py

This variant of my skel.py script uses the module.