python.general Namespace Reference

Functions
def	get_package_version (package_name, default_prefix="DataQuality")
def	kgrind_profile (filename="profile.kgrind")
def	daemonize (function, delay, error_args=None, **kwargs)
def	send_error_email (exception_info, from_, subject, body)
def	all_equal (*inputs)
def	all_permutations (input_list)
def	flattened (l)
def	_flatten (l, fn, val=[])
def	silence (log_level=None, logger=None)
def	timer (name)
def	interleave (*args)
def	date_to_nanounix (date_string)
def	nanounix_to_date (nanounix)

Variables
	log

Function Documentation

_flatten()

def python.general._flatten (   l,   fn,   val = []  )

private

Definition at line 134 of file general.py.

def _flatten(l, fn, val=[]):
    if type(l) is not list:
        return fn(l)
    if len(l) == 0:
        return fn(val)
    return [lambda x: _flatten(l[0], lambda y: _flatten(l[1:],fn,y), x), val]
 
@contextmanager

all_equal()

def python.general.all_equal

(

*

inputs

)

Returns true if all input arguments are equal (must be hashable)

Definition at line 107 of file general.py.

def all_equal(*inputs):
    """
    Returns true if all input arguments are equal (must be hashable)
    """
    return len(set(inputs)) == 1
    

all_permutations()

def python.general.all_permutations

(

input_list

)

Generate all permutations of `input_list`

Definition at line 113 of file general.py.

def all_permutations(input_list):
    """
    Generate all permutations of `input_list`
    """
    if len(input_list) <= 1:
        yield input_list
    else:
        for perm in all_permutations(input_list[1:]):
            for i in range(len(perm)+1):
                yield perm[:i] + input_list[0:1] + perm[i:]
 
# Taken from:
# http://caolanmcmahon.com/flatten_for_python

daemonize()

def python.general.daemonize	(		function,
			delay,
			error_args = None,
		**	kwargs
	)

Run a daemon which executes `function` every `delay` seconds.

Definition at line 44 of file general.py.

def daemonize(function, delay, error_args=None, **kwargs):
    """
    Run a daemon which executes `function` every `delay` seconds.
    """
    from time import sleep
    try:
        while True:
            log.info("Sleeping for %i seconds..", delay)
            sleep(delay)
            
            log.info("Current time: %s", strftime("%d/%m/%Y %H:%M:%S"))
        
            function(**kwargs)
    except Exception:
        if error_args is not None:
            import sys
            send_error_email(sys.exc_info(), **error_args)
        raise
 

date_to_nanounix()

def python.general.date_to_nanounix

(

date_string

)

Returns number of nanoseconds between unix epoch and date in the form
'dd/mm/yyyy'

Definition at line 173 of file general.py.

def date_to_nanounix(date_string):
    """
    Returns number of nanoseconds between unix epoch and date in the form
    'dd/mm/yyyy'
    """
    return int(mktime(strptime(date_string, "%d/%m/%Y")) * 1e9)
    

flattened()

def python.general.flattened

(

l

)

Definition at line 126 of file general.py.

def flattened(l):
    result = _flatten(l, lambda x: x)
    while type(result) is list and len(result) and callable(result[0]):
        if result[1] != []:
            yield result[1]
        result = result[0]([])
    yield result
 

get_package_version()

def python.general.get_package_version	(		package_name,
			default_prefix = "DataQuality"
	)

Returns package version as determined from "cmt show versions"

Definition at line 15 of file general.py.

def get_package_version(package_name, default_prefix="DataQuality"):
    """
    Returns package version as determined from "cmt show versions"
    """
    if "/" not in package_name and default_prefix:
        package_name = "/".join([default_prefix, package_name])
    output = getoutput("cmt show versions %s" % package_name).split("\n")
    output = list(map(str.strip, output))
    return " ".join(output[0].split()[1:])
 
@contextmanager

interleave()

def python.general.interleave

(

*

args

)

Definition at line 170 of file general.py.

def interleave(*args):
    return [item for items in zip(*args) for item in items]
 

kgrind_profile()

def python.general.kgrind_profile

(

filename = "profile.kgrind"

)

Generate kcachegrind profile information for a call. 

Example use:

with kgrind_profile() as p:
    p.run("main()")

Definition at line 26 of file general.py.

def kgrind_profile(filename="profile.kgrind"):
    """
    Generate kcachegrind profile information for a call. 
    
    Example use:
    
    with kgrind_profile() as p:
        p.run("main()")
    """
    from cProfile import Profile
    p = Profile()
    try:
        yield p
    finally:
        from .ext.lsprofcalltree import KCacheGrind
        with open(filename,"w") as fd:
            KCacheGrind(p).output(fd)
 

nanounix_to_date()

def python.general.nanounix_to_date

(

nanounix

)

Returns a string representation of `nanounix` nanoseconds 
since the unix epoch

Definition at line 180 of file general.py.

def nanounix_to_date(nanounix):
    """
    Returns a string representation of `nanounix` nanoseconds 
    since the unix epoch
    """
    sub_second = "%03i" % ((nanounix % int(1e9)) / 1e6)
    try:
        return strftime("%Y%m%d:%H%M%S.", gmtime(nanounix / 1e9)) + sub_second
    except ValueError:
        return "[BadTime: %s]" % int(nanounix)
 

send_error_email()

def python.general.send_error_email	(		exception_info,
			from_,
			subject,
			body
	)

Send an error email containing `exception_info`
TODO: Doesn't contain complete information

Definition at line 63 of file general.py.

def send_error_email(exception_info, from_, subject, body):
    """
    Send an error email containing `exception_info`
    TODO: Doesn't contain complete information
    """
    import smtplib
    import traceback
    
    from email.mime.text import MIMEText
    from getpass import getuser
    from socket import gethostname
 
    exceptionType, exceptionValue, exceptionTraceback = exception_info
 
    user, host = getuser(), gethostname()
 
    extra_info = ["Was running as user '%s' on machine '%s'" % (user, host),
                  #"Code located at: %s" % code
    ]
 
    msg = MIMEText("\n\n".join(body + extra_info +
                               traceback.format_tb(exceptionTraceback)))
 
    msg['Subject'] = subject
    msg['From'] = from_
    
    to = [
        #"Peter Waller <peter.waller@cern.ch>",
        "Peter Onyisi <peter.onyisi@cern.ch>"
    ]
    
    #operations = ("Data Quality Operations "
    #              "<hn-atlas-data-quality-operations@cern.ch>")
    
    msg['To'] = ", ".join(to)
    #msg['Cc'] = operations; to.append(operations)
 
    s = smtplib.SMTP()
    s.connect()
    s.sendmail(from_, to, msg.as_string())
    s.quit()
 
    log.error("Error email sent.")
    

silence()

def python.general.silence	(		log_level = None,
			logger = None
	)

Turn down the logging verbosity temporarily

Definition at line 142 of file general.py.

def silence(log_level=None, logger=None):
    """
    Turn down the logging verbosity temporarily
    """
    if log_level is None:
        log_level = log.WARNING
        
    if logger is None:
        logger = log.getLogger()
        
    orig_level = logger.level
    logger.setLevel(log_level)
    
    try:
        yield
    finally:
        logger.setLevel(orig_level)
 
@contextmanager

timer()

def python.general.timer

(

name

)

Definition at line 161 of file general.py.

def timer(name):
    "A context manager which spits out how long the block took to execute"
    start = time()
    try:
        yield
    finally:
        end = time()
        log.debug("Took %.2f to %s", (end - start, name))                
 

Variable Documentation

log

python.general.log

Definition at line 10 of file general.py.