i2.routing_forest¶

Tools to specify functions through trees and forests.

Whaaa?!?

Well, you see, often – especially when writing transformers – you have a series of if/then conditions nested into eachother, in code, where it gets ugly and un-reusable.

This module explores ways to objectivy this: That is, to give us the means to create such nested conditions in a way that we can define the parts as reusable operable components.

Think of the relationship between the for loop (code) and the iterator (object), along with iterator tools (itertools). This is what we’re trying to explore, but for if/then conditions.

I said explore. Some more work is needed here to make it robust and easily usable.

Let’s look at an example involving the three main actors of our play. Each of these are Iterable and Callable (Generator to be precise).

CondNode: implements the if/then (no else) logic
FinalNode: Final – yields (both with call and iter) it’s single .val attribute.
RoutingForest: An Iterable of CondNode

You’ll note that instances of these classes are all both callables and iterables, and that when called, they return iterables. It’s this aspect that makes us be able to nest conditions within conditions, and further, control the flow of the iteration from outside. A routing node (or forest) called on an object will yield all values that match the conditions that were specified for it. For example, if you need all matches, you can wrap it with list, if you need the first match only, you can wrap it with next, if you have a default value, you can wrap it in next with a default value.

>>> import inspect
>>>
>>> def could_be_int(obj):
...     if isinstance(obj, int):
...         b = True
...     else:
...         try:
...             int(obj)
...             b = True
...         except ValueError:
...             b = False
...     if b:
...         print(f'{inspect.currentframe().f_code.co_name}')
...     return b
...
>>> def could_be_float(obj):
...     if isinstance(obj, float):
...         b = True
...     else:
...         try:
...             float(obj)
...             b = True
...         except ValueError:
...             b = False
...     if b:
...         print(f'{inspect.currentframe().f_code.co_name}')
...     return b
...
>>> print(
...     could_be_int(30),
...     could_be_int(30.3),
...     could_be_int('30.2'),
...     could_be_int('nope'),
... )
could_be_int
could_be_int
True True False False
>>> print(
...     could_be_float(30),
...     could_be_float(30.3),
...     could_be_float('30.2'),
...     could_be_float('nope'),
... )
could_be_float
could_be_float
could_be_float
True True True False
>>> assert could_be_int('30.2') is False
>>> assert could_be_float('30.2') is True
could_be_float
>>>
>>> st = RoutingForest(
...     [
...         CondNode(
...             cond=could_be_int,
...             then=RoutingForest(
...                 [
...                     CondNode(
...                         cond=lambda x: int(x) >= 10,
...                         then=FinalNode('More than a digit'),
...                     ),
...                     CondNode(
...                         cond=lambda x: (int(x) % 2) == 1,
...                         then=FinalNode("That's odd!"),
...                     ),
...                 ]
...             ),
...         ),
...         CondNode(cond=could_be_float, then=FinalNode('could be seen as a float')),
...     ]
... )
>>> assert list(st('nothing I can do with that')) == []
>>> assert list(st(8)) == ['could be seen as a float']
could_be_int
could_be_float
>>> assert list(st(9)) == ["That's odd!", 'could be seen as a float']
could_be_int
could_be_float
>>> assert list(st(10)) == ['More than a digit', 'could be seen as a float']
could_be_int
could_be_float
>>> assert list(st(11)) == [
...     'More than a digit',
...     "That's odd!",
...     'could be seen as a float',
... ]
could_be_int
could_be_float
>>>
>>> print(
...     '### RoutingForest ########################################################################################'
... )
### RoutingForest ########################################################################################
>>> rf = RoutingForest(
...     [
...         SwitchCaseNode(
...             switch=lambda x: x % 5,
...             cases={0: FinalNode('zero_mod_5'), 1: FinalNode('one_mod_5')},
...             default=FinalNode('default_mod_5'),
...         ),
...         SwitchCaseNode(
...             switch=lambda x: x % 2,
...             cases={0: FinalNode('even'), 1: FinalNode('odd')},
...             default=FinalNode('that is not an int'),
...         ),
...     ]
... )
>>>
>>> assert list(rf(5)) == ['zero_mod_5', 'odd']
>>> assert list(rf(6)) == ['one_mod_5', 'even']
>>> assert list(rf(7)) == ['default_mod_5', 'odd']
>>> assert list(rf(8)) == ['default_mod_5', 'even']
>>> assert list(rf(10)) == ['zero_mod_5', 'even']
>>>

Functions

`identity`(obj)	Return the input unchanged (the default leaf function).
`return_sentinel`(obj[, sentinel])	Return a constanc sentinel value when called.
`test_routing_forest`()	Exercise the routing nodes end to end (kept here as a runnable example).
`wrap_leafs_with_final_node`(x)	Yield the items of `x`, wrapping those that are not `RoutingNode` in `FinalNode`.

Classes

`CondNode`(cond, then)	A RoutingNode that implements the if/then (no else) logic
`DelegateToMappingAttrMixin`()	A mixin to delegate `Mapping` methods to a mapping attribute called `mapping`
`FeatCondNode`(feat, feat_cond_thens)	A RoutingNode that yields multiple routes, one for each of several conditions met, where the condition is computed implements computes a feature of the obj and according to an iterable of conditions on the feature.
`FinalNode`(val)	A RoutingNode that is final.
`KeyFuncMapping`([mapping, key, default_factory])	Implements a switch-case-like mapping with a callable key function.
`NoDefault`()
`RoutingForest`(cond_nodes)
`RoutingNode`()	A RoutingNode instance needs to be callable on a single object, yielding an iterable or a final value
`SwitchCaseNode`(switch, cases[, default])	A RoutingNode that implements the switch/case/else logic.

class i2.routing_forest.CondNode(cond, then)[source]¶

Bases: RoutingNode

A RoutingNode that implements the if/then (no else) logic

class i2.routing_forest.DelegateToMappingAttrMixin[source]¶

Bases: object

A mixin to delegate Mapping methods to a mapping attribute called mapping

class i2.routing_forest.FeatCondNode(feat, feat_cond_thens)[source]¶

Bases: RoutingNode

A RoutingNode that yields multiple routes, one for each of several conditions met, where the condition is computed implements computes a feature of the obj and according to an iterable of conditions on the feature.

>>> fcn = FeatCondNode(
...     feat=lambda x: x % 5,
...     feat_cond_thens=[
...         (lambda x: x == 0, lambda x: 'zero_mod_5'),
...         (lambda x: x == 1, lambda x: 'one_mod_5'),
...         (lambda x: x == 2, lambda x: 'two_mod_5'),
...         (lambda x: x == 3, lambda x: 'three_mod_5'),
...         (lambda x: x == 4, lambda x: 'four_mod_5'),
...     ]
... )
>>> assert list(fcn(0)) == ['zero_mod_5']
>>> assert list(fcn(1)) == ['one_mod_5']
>>> assert list(fcn(2)) == ['two_mod_5']
>>> assert list(fcn(3)) == ['three_mod_5']
>>> assert list(fcn(4)) == ['four_mod_5']
>>> assert list(fcn(5)) == ['zero_mod_5']
>>> assert list(fcn(6)) == ['one_mod_5']

classmethod from_feature_val_map(feat, feat_cond_thens)[source]¶

A FeatCondNode where the conditions are equality checks on the feature value

# >>> fvn = FeatCondNode.from_feature_val_map( # … feat=lambda x: x % 3, # … feat_cond_thens={ # … 0: lambda x: ‘zero_mod_3’, # … 1: lambda x: ‘one_mod_3’, # … 2: lambda x: ‘two_mod_3’, # … } # … ) # >>> list(fvn(0)) # # >>> assert list(fvn(0)) == [‘zero_mod_3’] # >>> assert list(fvn(1)) == [‘one_mod_3’] # >>> assert list(fvn(2)) == [‘two_mod_3’] #

class i2.routing_forest.FinalNode(val)[source]¶

Bases: RoutingNode

A RoutingNode that is final. It yields (both with call and iter) it’s single .val attribute.

class i2.routing_forest.KeyFuncMapping(mapping=None, key=<function identity>, default_factory=<function return_sentinel>)[source]¶

Bases: DelegateToMappingAttrMixin, MutableMapping

Implements a switch-case-like mapping with a callable key function.

The purpose of KeyFuncMapping is to allow switch-case logic to be given as a plugin specification.

>>> from i2.routing_forest import KeyFuncMapping
>>>
>>> get_extension = lambda x: x.split('.')[-1]
>>>
>>> data_type = KeyFuncMapping(
...     {'csv': 'table', 'xls': 'table', 'wav': 'audio'}, key=get_extension
... )

Calling a KeyFuncMapping instance will call the key function on the input, then look up the result in the mapping.

>>> data_type('my_file.csv')
'table'
>>> data_type('another_file.xls')
'table'
>>> data_type('sound.wav')
'audio'

If the key is not found in the mapping, the default_factory is called with the input and the result is returned. The default default_factory is return_sentinel, which by default returns None

>>> assert data_type('poem.txt') is None

Note that instances of KeyFuncMapping are also Mapping``s, so all ``Mapping methods can be used.

>>> list(data_type)
['csv', 'xls', 'wav']
>>> dict(data_type)
{'csv': 'table', 'xls': 'table', 'wav': 'audio'}

Including update, which constitutes a convenient way to extend the mapping.

>>> data_type.update(txt='text')
>>> data_type('poem.txt')
'text'

The default_factory can be set to any callable, including a KeyFuncMapping itself, which enables us to define an else for the switch-case logic that a KeyFuncMapping implements. Say, for example, if no handled extension is found, we want to check the protocol of the input string instead. This is not only a new mapping, but also a new key function. We can do it as such:

>>> get_protocol = lambda x: x.split('://')[0]
>>> protocol = KeyFuncMapping({'https': 'url'}, get_protocol)
>>> new_data_type = KeyFuncMapping(
...     data_type.mapping, data_type.key, default_factory=protocol
... )
>>> new_data_type('notes.txt')
'text'
>>> new_data_type('https://www.python.org/')
'url'

Given how useful this pattern is, we made the + operator implement this. Note that here, + is not associative or commutative (as with numbers). It should be understood to function more like the + for iterables like list and tuple.

>>> nested = data_type + protocol
>>> nested('https://www.python.org/')
'url'
>>> nested('jazz.wav')
'audio'

default_factory(sentinel=None)¶

Return a constanc sentinel value when called. Use partial to set sentinel

key()¶

Return the input unchanged (the default leaf function).

class i2.routing_forest.RoutingForest(cond_nodes)[source]¶

Bases: RoutingNode

>>> rf = RoutingForest([
...     CondNode(cond=lambda x: isinstance(x, int),
...              then=RoutingForest([
...                  CondNode(cond=lambda x: int(x) >= 10, then=FinalNode('More than a digit')),
...                  CondNode(cond=lambda x: (int(x) % 2) == 1, then=FinalNode("That's odd!"))])
...             ),
...     CondNode(cond=lambda x: isinstance(x, (int, float)),
...              then=FinalNode('could be seen as a float')),
... ])
>>> assert list(rf('nothing I can do with that')) == []
>>> assert list(rf(8)) == ['could be seen as a float']
>>> assert list(rf(9)) == ["That's odd!", 'could be seen as a float']
>>> assert list(rf(10)) == ['More than a digit', 'could be seen as a float']
>>> assert list(rf(11)) == ['More than a digit', "That's odd!", 'could be seen as a float']

class i2.routing_forest.RoutingNode[source]¶

Bases: object

A RoutingNode instance needs to be callable on a single object, yielding an iterable or a final value

static from_object(x, mini_lang=<function _default_mini_lang>)[source]¶: Converts an object to a RoutingNode instance. Enables mini-languages to be developed for defining routing trees.

class i2.routing_forest.SwitchCaseNode(switch, cases, default=<i2.routing_forest.NoDefault object>)[source]¶

Bases: RoutingNode

A RoutingNode that implements the switch/case/else logic. It’s just a specialization (enhanced with a “default” option) of the FeatCondNode class to a situation where the cond function of feat_cond_thens is equality, therefore the routing can be implemented with a {value_to_compare_to_feature: then_node} map.

Parameters:

switch (Callable) – A function returning the feature of an object we want to switch on
cases (Mapping) – The mapping from feature to RoutingNode that should be yield for that feature. It is often a dict, but only requirement is that it implements the cases.get(val, default) method.
default (Any) – Default RoutingNode to yield if no

>>> rf = RoutingForest([
...     SwitchCaseNode(switch=lambda x: x % 5,
...                    cases={0: FinalNode('zero_mod_5'), 1: FinalNode('one_mod_5')},
...                    default=FinalNode('default_mod_5')),
...     SwitchCaseNode(switch=lambda x: x % 2,
...                    cases={0: FinalNode('even'), 1: FinalNode('odd')},
...                    default=FinalNode('that is not an int')),
... ])
>>>
>>> assert(list(rf(5)) == ['zero_mod_5', 'odd'])
>>> assert(list(rf(6)) == ['one_mod_5', 'even'])
>>> assert(list(rf(7)) == ['default_mod_5', 'odd'])
>>> assert(list(rf(8)) == ['default_mod_5', 'even'])
>>> assert(list(rf(10)) == ['zero_mod_5', 'even'])

i2.routing_forest.identity(obj)[source]¶

Return the input unchanged (the default leaf function).

i2.routing_forest.return_sentinel(obj, sentinel=None)[source]¶

Return a constanc sentinel value when called. Use partial to set sentinel

i2.routing_forest.test_routing_forest()[source]¶: Exercise the routing nodes end to end (kept here as a runnable example).

i2.routing_forest.wrap_leafs_with_final_node(x)[source]¶: Yield the items of x, wrapping those that are not RoutingNode in FinalNode.