advice needed for lazy evaluation mechanism

markolopa marko.loparic at gmail.com
Sun Nov 8 23:41:27 CET 2009


Hi,

Could you please give me some advice on the piece of code I am
writing?

My system has several possible outputs, some of them are not always
needed. I started to get confused with the code flow conditions needed
to avoid doing unnecessary work. So I am trying to restructure it
using lazy evaluation.

In the new mechanism I am coding I have a repository with two types of
objects: infos and routines. In the begining I have a list of
routines. Each routine tells which infos it can compute. The execution
is triggered when the value of an info is requested. In the example
below I have 3 routines

Routine "ReadData" computes info "gender" and info "birth_year"
Routine "YearToAge" computes info "age" (using info "birth_year")
Routine "ComputeMHF" computes info "max_heart_frequency" (using info
"gender" and info "age")

           /--> gender ----------------------------\
ReadData --|                                       | --> ComputeMHF --
> max_heart_frequency
           \--> birth_year --> YearToAge --> age --/

So for instance if all I need is info "age", only the routines
"ReadData" and "YearToAge" are computed.

The code below implements the example. There are 3 files:
  - test.py: the test case for the example
  - routines.py: the routines (classes) of the example
  - repository.py: the lazy evaluation mechanism (independent of the
example)

My questions are:
- Is there a more standard (pythonic) way to do what I am trying to
do? Are there libraries, design patterns, functional programming
structures to use to achieve what I am looking for (i.e. am I trying
to reinvent the wheel)?
- Is the coding style good?
- Can I avoid the eval command in Repository.add_routine? What I want
there is to be able to have a generic code for the repository which
does not depend on the files containing the routines I want it to
hold.

Note: The routines do not need to declare the info they depend on.
They request the info in the computation phase.

test.py
===
import unittest
from repository import Repository

ROUTINE_LIST = """
ReadData routines
YearToAge routines
ComputeMHF routines
"""

class Test(unittest.TestCase):

    def test_age(self):
        repo = Repository(ROUTINE_LIST)
        self.assertEqual(repo['age'], 30)

    def test_max_heart_frequency(self):
        repo = Repository(ROUTINE_LIST)
        self.assertEqual(repo['max_heart_frequency'], 181)
===

routines.py
===
from repository import AbstractRoutine

class ReadData(AbstractRoutine):
    def __init__(self):
        super(ReadData, self).__init__(self.__class__.__name__,
                                       ['birth_year', 'gender'])
    def compute(self, repo):
        repo['birth_year'] = 1979
        repo['gender'] = 'F'

class YearToAge(AbstractRoutine):
    def __init__(self):
        super(YearToAge, self).__init__(self.__class__.__name__,
                                        ['age'])
    def compute(self, repo):
        repo['age'] = 2009 - repo['birth_year']

class ComputeMHF(AbstractRoutine):
    def __init__(self):
        super(ComputeMHF, self).__init__(self.__class__.__name__,
                                         ['max_heart_frequency'])
    def compute(self, repo):
        gender = repo['gender']
        age = repo['age']
        mhf = 211 - age if gender == 'F' else 205 - age
        repo['max_heart_frequency'] = mhf
===

repostory.py
===
from StringIO import StringIO

class AbstractRoutine(object):

    def __init__(self, name, infos_provided):
        self.name = name
        self.infos_provided = infos_provided
        self.computed = False

    def compute(self):
        raise NotImplementedError

class Info(object):
    def __init__(self, name, routine):
        self.name = name
        self.routine = routine
        self.computed = False
        self.value = None

class Repository(object):

    def __init__(self, routine_definition_lines):
        self._infos = {}
        self.add_routines(routine_definition_lines)

    def add_routines(self, definition_lines):
        for line in StringIO(definition_lines):
            line = line.strip()
            if line == '':
                continue
            name, file_name = line.split()
            self.add_routine(name, file_name)

    def add_routine(self, class_name, file_name):
        routine = None  # only to cheat pylint
        cmd = "from %s import %s\nroutine = %s()" % (file_name,
                                                     class_name,
 
class_name)
        exec(cmd)  # XXX: ugly
        if not isinstance(routine, AbstractRoutine):
            raise ValueError('Class %s is not AbstractRoutine'
                             % class_name)
        for info_name in routine.infos_provided:
            info = Info(info_name, routine)
            self._infos[info_name] = info

    def __setitem__(self, key, value):
        if key not in self._infos:
            raise ValueError('info %s not defined in repository' %
key)
        info = self._infos[key]
        if info.computed:
            raise ValueError('info %s has already been computed' %
key)
        info.value = value
        info.computed = True

    def __getitem__(self, key):
        if key not in self._infos:
            raise ValueError('info %s not defined in repository' %
key)
        info = self._infos[key]
        if not info.computed:
            print('Calling routine %s to compute info %s'
                  % (info.routine.name, info.name))
            info.routine.compute(self)
            if not info.computed:
                raise ValueError('routine %s did not compute info %s'
%
                                 (info.routine.name, key))
        return info.value
===

Thanks a lot!
Marko




More information about the Python-list mailing list