Difference between revisions of "Implementing a Standard Library"
Jump to navigation
Jump to search
PeterHarding (talk | contribs) |
PeterHarding (talk | contribs) |
||
| Line 206: | Line 206: | ||
debug_level = 0 | debug_level = 0 | ||
verbose_flg = True | 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) | |||
#-------------------------------------------------------------------- | |||
#========================================================================== | |||
| Line 219: | Line 240: | ||
Logger.Error("Testing - Error...") | Logger.Error("Testing - Error...") | ||
time.sleep( | time.sleep(1.4) | ||
print "Get timer - %5.3f" % performiq.Timer.time() | 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 | |||
</pre> | </pre> | ||
| Line 236: | Line 275: | ||
[logger::ERROR] Testing - Error... | [logger::ERROR] Testing - Error... | ||
Get timer - 3.001 | Get timer - 3.001 | ||
Read 3 data items... | |||
-> Total data 3 | |||
a,b,c,d | |||
e,f,g,h | |||
i,j,k,l | |||
</pre> | </pre> | ||
Revision as of 12:39, 20 February 2013
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 7 -rw-r--r-- 1 pharding Users 108 Feb 20 11:06 __init__.py -rw-r--r-- 1 pharding Users 23 Feb 20 10:52 CONSTANTS.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 2121 Feb 20 11:32 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
#----------------------------------------------------------------------
#==========================================================================
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.