### Navigation

- index
- modules |
- CCHR 1.0 documentation »
- Module code »
- src »

# Source code for src.util

    import math, re, sys, os, MySQLdb, shutil, hashlib
    from functools import wraps
    from datetime import datetime
    from collections import defaultdict
    from contextlib import closing

    import cardsharp as cs
    from cardsharp.errors import LoadError
    from configuration import config, db_info
    from errors import CalcError

    __all__ = ['RUNDATETIME', 'FORMAT_INFO', 'FORMAT_DICT', 'PHASE_MAP',
               'key_as_str', 'check_file', 'calc', 're_escape', 'get_state_key',
               'get_cnxn', 'get_conv_cnxn', 'is_file', 'is_none', 'memoize',
               'str_to_boolean', 'get_state_name', 'get_max_review_round', 'copy_to_review',
               'set_LOOKUPS', 'LOOKUPS', 'check_format','capture_input', 'get_hash']

    RUNDATETIME = datetime.now().isoformat('_').replace(':', '-')

    FORMAT_INFO = {
             #'sas' : '.sas7bdat',
            'text' : '.txt',
           'excel' : '.xls',
            'spss' : '.sav',
          'csharp' : '.csharp',
             'csv' : '.csv',
        #'sas7bdat' : '.sas7bdat',
             'txt' : '.txt',
             'xls' : '.xls',
             'sav' : '.sav',
             'del' : '.del',
       #'.sas7bdat' : 'sas',
            '.txt' : 'text',
            '.xls' : 'excel',
            '.sav' : 'spss',
         '.csharp' : 'csharp',
            '.csv' : 'csv',
       #'.sas7bdat' : 'sas7bdat',
            '.del' : 'del',
    }

    FORMAT_DICT = {'float':float,
                 'integer':int,
                 'boolean':bool,
                 'string':str,
                 }

    PHASE_MAP = {0: 'stage',
                 1: 'pre_process',
                 2: 'process',
                 3: 'transform',
                 4: 'compare'}

    try:
        region_ds = cs.load(source=os.path.join(db_info['stand']['dir'], r'lookups\region.xls'), format='excel')
        region_ds.variables['id'].convert('integer')

    except LoadError:
        region_ds = None
        print 'Unable to load region dataset'

    regions = {}
    LOOKUPS = defaultdict(list)

    def get_hash(key):
        return hashlib.md5(key).hexdigest()

    def capture_input(text, float_range=None, int_range=None, str_opt=None, open=None, default=None, as_tuple=False, as_list=False, allow_int=False, as_bool=False, **kw):
        ans = None

        while not ans:
            ans = raw_input(text).lower()
            if ans == '' and default is not None:
                return default
            elif ans == '' and kw.get('required'): continue

            if int_range and float_range:
                print 'This option not yet enabled'
                print 'TODO allow for both integers and floats'

            if int_range:
                try:
                    _range = int_range[0] if len(int_range) == 1 else range(int_range[0], int_range[-1]+1)
                    if as_list or as_tuple:
                        ans = [a for a in ans.split(',')]

                        for a in ans:
                            if int(a) not in _range:
                                print '%s is not a valid option please enter a value %i - %i.' % (ans, int_range[0], int_range[-1])
                                ans = None
                                break

                        if ans != None:
                            if as_list: return [int(a) for a in ans]
                            else: return tuple([int(a) for a in ans])

                    else:
                        if int(ans) not in _range:
                            print '%s is not a valid option please enter a value %i - %i.' % (ans, int_range[0], int_range[-1])
                            ans = None
                        else:
                            return int(ans)

                except ValueError:
                    print '%s is not a valid option please enter a value %i - %i.' % (ans, int_range[0], int_range[-1])
                    ans = None

            elif float_range:
                try:
                    if as_list or as_tuple:
                        _valid = True

                        for a in ans:
                            #allow integers in float_range option
                            if allow_int and '.' not in a:
                                try: int(a)
                                except ValueError:
                                    print '%s is not a valid integer.' % a

                            elif not float_range[0] < float(a) < float_range[1]:
                                print '%s is not a valid option please enter a value between 0 and 1 .' % ans
                                ans = None
                                break

                        if ans != None:
                            temp = [float(a) if '.' in a else int(a) for a in ans] if allow_int else [float(a) for a in ans]
                            return temp if as_list else tuple(temp)

                    else:
                        if not float_range[0] < float(ans) < float_range[1]:
                            print '%s is not a valid option please enter a value between 0 and 1 .' % ans
                            ans = None
                        else:
                            return float(ans)

                except ValueError:
                    print '%s is not a valid option please enter a value between 0 and 1 .' % ans
                    ans = None

            elif str_opt:
                if as_list or as_tuple:
                    ans = [a.strip().lower() for a in ans.split(',')]
                    ans = ans if as_list else tuple(ans)

                    for _a in ans:
                        if _a not in str_opt:
                            print '%s is not a valid option please enter a one of the following: %s.' % (_a, str_opt)
                            ans = None
                            break

                    if ans != None:
                        return ans

                else:
                    if ans not in str_opt:
                        print '%s is not a valid option please enter a one of the following: %s.' % (ans, str_opt)
                        ans = None
                    else:
                        return ans

            elif as_bool:
                if ans not in ('y', 'yes', 'n', 'no'):
                    print '%s is not a valid option please enter yes or no.' % ans
                    ans = None
                elif ans in ('y', 'yes'): return True
                else: return False

            else:
                return ans

    def set_LOOKUPS(db_info):
        for file in os.listdir(os.path.join(config.meta_dir, 'database', 'standardized',  'lookups')):
            lookup_name = file.replace('.xls', '')
            with closing(MySQLdb.connect(host=config.db_host, user=db_info['user'], passwd=db_info['pass'],
                         charset='utf8', db=db_info['name'], port=config.db_port, use_unicode=True)) as conn:
                with closing(conn.cursor()) as cur:
                    cur.execute('SELECT id from %s' % lookup_name)
                    for row in cur.fetchall():
                        LOOKUPS[lookup_name].append(row[0])

    if region_ds:
        for row in region_ds:
            regions[row['id']] = row['abbr']

        for row in region_ds:
            regions[row['abbr']] = row['id']

    def get_cnxn(db_name, opt):
        return MySQLdb.connect(host=config.db_host, user=opt['db_info']['user'], passwd=opt['db_info']['pass'],
                               charset='utf8', db=db_name, port=config.db_port, use_unicode=True)

    def get_conv_cnxn(db_name, opt):
        from MySQLdb.converters import conversions
        return MySQLdb.connect(host=config.db_host, user=opt['db_info']['user'], passwd=opt['db_info']['pass'],
                               charset='utf8', db=db_name, port=config.db_port, use_unicode=True, conv=conversions)

    def check_format(format):
        """Helper funtion to ensure that cradsharp is able to handle the
        format arg."""
        if not format:
            return None

        if format not in ['csharp', 'excel', 'text', 'csv', 'mysql', 'spss','del']:
            print "%s not a valid file format." % format
            sys.exit(1)

        return format

    def key_as_str(value, inner_trim=True):
        """Produces a key value for looking up in a crosswalk.

        If value is None than returns an empty string.
        >>> v = None
        >>> key_as_str(v) + 'XXX'
        XXX

        If value is a basestring removes leading and trailing whitespace if inner_trim = True,
        multiple line spaces made into one space, value is lowered:
        >>> v = '  DS     sS    '
        >>> key_as_str(v)
        ds ss

        If inner_trime is false just strim leading and trailing and lower
        >>> v = '   DF     sdsD'
        >>> v
        df     sdsd
        """

        if value is None:
            return ''
        else:
            if isinstance(value, basestring):
                if inner_trim:
                    return ' '.join(value.split()).lower()
                else:
                    return value.strip().lower()
            else:
                return str(value)

    def get_state_key(value):
        if not value:
            return ''
        else:
            if isinstance(value, basestring):
                try:
                    #check to see if the string passed in is an int
                    return str(regions.get(int(value)))
                except:
                    return value.lower().strip()
            else:
                return str(regions.get(value))

    def get_state_name(value):
        if not value:
            return ''
        else:
            if isinstance(value, basestring):
                try:
                    #check to see if the string passed in is an int
                    return str(regions.get(int(value)))
                except:
                    return str(regions.get(value))

            return str(regions.get(value))

    def check_file(file_path, as_dir = False):
        """Helper function to check if file or directory is valid."""
        if not os.path.exists(file_path):
            print "%s does not exist." % file_path
            sys.exit(1)

        if as_dir:
            if not os.path.isdir(file_path):
                print "%s is not a directory." % file_path
                sys.exit(1)
        else:
            if not os.path.isfile(file_path):
                print "%s is not a file." % file_path
                sys.exit(1)

        return file_path

    #get the highest round for a state
    def get_max_review_round(segment, region, set=config.current_set, **kw):
        _path = os.path.join(config.review_dir, 'set_%s' % set, segment, region)
        #check to see if no review files exist
        #if no files exist return lowest to allow creation of initial review round
        if not os.path.isdir(_path):
            return -1

        round = -1
        _round = -1

        for dir in os.listdir(_path):
            try:
                _round = int(re.search('round(\d+)', dir).groups()[0])
            except AttributeError, ValueError:
                print 'warning: bad review folder %s' % os.path.join(_path, dir)

            if _round > round:
                round = _round
                max_dir = dir

        if kw.get('as_dir'):
            return os.path.join(_path, max_dir)
        else:
            return round

    def copy_to_review(name, region, orig_path):
        """Copy output file to review directory and return the path to the review directory"""
        _round = get_max_review_round(name, region)
        _round = 1 if _round == -1 else _round + 1
        _review_path = os.path.join(config.review_dir, 'set_%s' % config.current_set, name, region, 'round%s' % _round)
        if not os.path.isdir(_review_path): os.makedirs(_review_path)
        shutil.copy(orig_path, _review_path)
        return _review_path

    def is_file(file_path):
        if not os.path.isfile(file_path):
            return False
        return True

    integers_regex = re.compile(r'\b[\d\.]+\b')

    def paranreplace(matchobj):
        return re.sub('\(|\)(?!\()', lambda a: '*(' if a.group(0) == '(' else ')*', matchobj.group(0))
    parans = re.compile('\d\(|\)\(|\)\d')

    def calc(expr, advanced=False):
        def safe_eval(expr, symbols={}):
            expr = re.sub(parans, paranreplace, expr)
            return eval(expr, dict(__builtins__=None), symbols)
        def whole_number_to_float(match):
            group = match.group()
            if group.find('.') == -1:
                return group + '.0'
            return group
        expr = expr.replace('^','**')
        expr = expr.replace(',','')
        expr = integers_regex.sub(whole_number_to_float, expr)
        try:
            if advanced:
                return safe_eval(expr, vars(math))
            else:
                return safe_eval(expr)
        except:
            raise CalcError

    _escape_map = {'\\' : '\\\\',
                   '.' : '\\.',
                   '^' : '\\^',
                   '$' : '\\$',
                   '*' : '\\*',
                   '+' : '\\+',
                   '?' : '\\?',
                   '{' : '\\{',
                   '}' : '\\}',
                   '[' : '\\[',
                   ']' : '\\]',
                   '|' : '\\|',
                   '(' : '\\(',
                   ')' : '\\)'}

    def re_escape(value):
        '''
        Escape a regular expression.

        >>> re_escape(ab(#))
        u'ab\((\d+|#\d+|#)\)'

        '''
        return ''.join([_escape_map.get(c, c) for c in value])

    def is_none(value):
        return True if value is None or value is '' else False

    def memoize(func):
        """This function integrates memoization, an optimization technique, into cardsharp.
        Memoize wraps a function with a single argument.  When it is called the first time, the result
        is cached with the argument as the key.  If it is called in the future with the same
        argument, it returns the cached value.

        >>> from random import random
        >>> func = lambda x: random()
        >>> func(1) == func(1)
        False
        >>> func = memoize(func)
        >>> func(1) == func(1)
        True
        """
        memo = dict()

        @wraps(func)
        def _func(arg):
            if arg in memo:
                return memo[arg]

            memo[arg] = value = func(arg)
            return value

        return _func

    def str_to_boolean(s):
        """Converts a string to a boolean.

        Convert map:
        -------------------
        | 1     | *True*  |
        -------------------
        | true  | *True*  |
        -------------------
        | 0     | *False* |
        -------------------
        | false | *False* |
        -------------------

        If *s* is *None* returns *None* instead.

        >>> str_to_boolean('1')
        True
        >>> str_to_boolean(0)
        False
        >>> str_to_boolean('  FAlse  ')
        False
        >>> str_to_boolean('3')
        Traceback (most recent call last):
            ...
        FormatError: Invalid input data, 3: must be true, false, 1 or 0
        """
        if s is None:
            return None
        else:
            if str(s) in ['1', '1.0'] or str(s).strip().lower() == 'true':
                return True
            elif str(s) in ['0', '0.0'] or str(s).strip().lower() == 'false':
                return False
            else:
                raise FormatError('Invalid input data, %s: must be true, false, 1 or 0' % s)

### Navigation

- index
- modules |
- CCHR 1.0 documentation »
- Module code »
- src »