putput package

Submodules

putput.combiner module

putput.combiner.combine(utterance_combo: Sequence[Sequence[str]], tokens: Sequence[str], groups: Sequence[Tuple[str, int]], *, token_handler_map: Optional[Mapping[str, Callable[[str, str], str]]] = None, group_handler_map: Optional[Mapping[str, Callable[[str, Sequence[str]], str]]] = None, combo_options: Optional[putput.joiner.ComboOptions] = None) → Tuple[int, Iterable[Tuple[str, Sequence[str], Sequence[str]]]][source]

Generates an utterance, handled tokens, and handled groups.

Parameters:
  • utterance_combo – An utterance_combo from pipeline.expander.expand.
  • tokens – Tokens that have yet to be handled from pipeline.expander.expand.
  • groups – Groups that have yet to be handled from pipeline.expander.expand.
  • token_handler_map – A mapping between a token and a function with args (token, phrase generated by token) that returns a handled token. If ‘DEFAULT’ is specified as the token, the handler will apply to all tokens not otherwise specified in the mapping.
  • combo_options – Options for randomly sampling the combination of ‘utterance_combo’.
Returns:

The length of the Iterable, and the Iterable consisting of an utterance and handled tokens.

Examples

>>> def _iob_token_handler(token: str, phrase: str) -> str:
...     tokens = ['{}-{}'.format('B' if i == 0 else 'I', token)
...               for i, _ in enumerate(phrase.replace(" '", "'").split())]
...     return ' '.join(tokens)
>>> def _just_groups(group_name: str, _: Sequence[str]) -> str:
...     return '[{}]'.format(group_name)
>>> token_handler_map = {'DEFAULT': _iob_token_handler}
>>> group_handler_map = {'DEFAULT': _just_groups}
>>> combo_options = ComboOptions(max_sample_size=2, with_replacement=False)
>>> utterance_combo = (('can she get', 'may she get'), ('fries',))
>>> tokens = ('ADD', 'ITEM')
>>> groups = (('ADD_ITEM', 2),)
>>> sample_size, generator = combine(utterance_combo,
...                                  tokens,
...                                  groups,
...                                  token_handler_map=token_handler_map,
...                                  group_handler_map=group_handler_map,
...                                  combo_options=combo_options)
>>> sample_size
2
>>> for utterance, handled_tokens, handled_groups in generator:
...     print(utterance)
...     print(handled_tokens)
...     print(handled_groups)
can she get fries
('B-ADD I-ADD I-ADD', 'B-ITEM')
('[ADD_ITEM]',)
may she get fries
('B-ADD I-ADD I-ADD', 'B-ITEM')
('[ADD_ITEM]',)

putput.expander module

putput.expander.expand(pattern_def: Mapping[KT, VT_co], *, dynamic_token_patterns_map: Optional[Mapping[str, Sequence[str]]] = None) → Tuple[int, Iterable[Tuple[Sequence[Sequence[str]], Sequence[str], Sequence[Tuple[str, int]]]]][source]

Expands the pattern_def to prepare for combination.

Parameters:
  • pattern_def – A dictionary representation of the pattern definition.
  • dynamic_token_patterns_map – The ‘dynamic’ counterpart to the ‘static’ section in the pattern definition. This mapping between token and token patterns is useful in scenarios where tokens and token patterns cannot be known before runtime.
Returns:

The length of the Iterable, and the Iterable consisting of an utterance_combo, tokens (that have yet to be handled), and groups (that have yet to be handled).

Examples

>>> from pathlib import Path
>>> from putput.pipeline import _load_pattern_def
>>> pattern_def_path = Path(__file__).parent.parent / 'tests' / 'doc' / 'example_pattern_definition.yml'
>>> pattern_def = _load_pattern_def(pattern_def_path)
>>> dynamic_token_patterns_map = {'ITEM': ('fries',)}
>>> num_utterance_patterns, generator = expand(pattern_def,
...                                            dynamic_token_patterns_map=dynamic_token_patterns_map)
>>> num_utterance_patterns
1
>>> for utterance_combo, unhandled_tokens, unhandled_groups in generator:
...     print(utterance_combo)
...     print(unhandled_tokens)
...     print(unhandled_groups)
(('can she get', 'may she get'), ('fries',), ('can she get', 'may she get'), ('fries',), ('and',), ('fries',))
('ADD', 'ITEM', 'ADD', 'ITEM', 'CONJUNCTION', 'ITEM')
(('ADD_ITEM', 2), ('ADD_ITEM', 2), ('None', 1), ('None', 1))
putput.expander.expand_utterance_patterns_ranges_and_groups(utterance_patterns: Sequence[Sequence[str]], group_map: Mapping[str, Sequence[str]]) → Tuple[Sequence[Sequence[str]], Sequence[Sequence[Tuple[str, int]]]][source]

Expands ranges and groups in utterance patterns, ensuring each utterance pattern is unique.

Parameters:
  • utterance_patterns – utterance_patterns section of pattern_def.
  • group_map – A mapping between a group name and the tokens that make up the group.
Returns:

A tuple of utterance patterns with group names and ranges replaced by tokens, and groups which are tuples of (group_name, number of tokens the group spans).

Examples

>>> utterance_patterns = [['WAKE', 'PLAY_ARTIST', '1-2']]
>>> group_map = {'PLAY_ARTIST': ('PLAY', 'ARTIST')}
>>> patterns, groups = expand_utterance_patterns_ranges_and_groups(utterance_patterns, group_map)
>>> tuple(sorted(patterns, key=lambda item: len(item)))
(('WAKE', 'PLAY', 'ARTIST'), ('WAKE', 'PLAY', 'ARTIST', 'PLAY', 'ARTIST'))
>>> tuple(sorted(groups, key=lambda item: len(item)))
((('None', 1), ('PLAY_ARTIST', 2)), (('None', 1), ('PLAY_ARTIST', 2), ('PLAY_ARTIST', 2)))
putput.expander.get_base_item_map(pattern_def: Mapping[KT, VT_co], base_key: str) → Mapping[str, Sequence[str]][source]

Returns base item map or an empty dictionary if one does not exist.

Parameters:
  • pattern_def – A dictionary representation of the pattern definition.
  • base_key – Key in pattern_def corresponding to the base item map.

Examples

>>> from pathlib import Path
>>> from putput.pipeline import _load_pattern_def
>>> pattern_def_path = Path(__file__).parent.parent / 'tests' / 'doc' / 'example_pattern_definition.yml'
>>> pattern_def = _load_pattern_def(pattern_def_path)
>>> get_base_item_map(pattern_def, 'groups')
{'ADD_ITEM': ('ADD', 'ITEM')}
>>> get_base_item_map(pattern_def, 'base_tokens')
{'PRONOUNS': ('she',)}
>>> get_base_item_map(pattern_def, 'not_a_key')
{}

putput.joiner module

class putput.joiner.ComboOptions(*, max_sample_size: int, with_replacement: bool)[source]

Bases: object

Options for join_combo via random sampling.

max_sample_size

Ceiling for number of components to sample.

with_replacement

Option to include duplicates when randomly sampling. If True, will sample max_sample_size. If False, will sample up to ‘max_sample_size’ unique combinations.

Raises:ValueError – If max_sample_size <= 0.
max_sample_size

Ceiling for number of components to sample.

with_replacement

Option to include duplicates when randomly sampling.

putput.joiner.join_combo(combo: Sequence[Sequence[T]], *, combo_options: Optional[putput.joiner.ComboOptions] = None) → Iterable[Sequence[T]][source]

Generates the product of a combo, subject to ‘combo_options’.

If ‘combo_options’ is not specified, ‘join_combo’ returns an Iterable of the product of combo. If ‘combo_options’ is specified, ‘join_combo’ returns an Iterable of samples of the product of combo.

Sampling should be used to speed up the consumption of the returned Iterable as well as to control size of the product, especially in cases where oversampling/undersampling is desired.

Parameters:
  • combo – Sequences to join.
  • combo_options – Options for randomly sampling.
Yields:

A joined combo.

Examples

>>> random.seed(0)
>>> combo = (('hey', 'ok'), ('speaker', 'sound system'), ('play',))
>>> tuple(join_combo(combo))
(('hey', 'speaker', 'play'), ('hey', 'sound system', 'play'),
('ok', 'speaker', 'play'), ('ok', 'sound system', 'play'))
>>> combo_options = ComboOptions(max_sample_size=1, with_replacement=False)
>>> tuple(join_combo(combo, combo_options=combo_options))
(('ok', 'sound system', 'play'),)

putput.logger module

putput.logger.get_logger(module_name: str, *, level: int = 20, stream: IO[str] = <_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'>) → logging.Logger[source]

Returns a configured logger for the module.

Parameters:
  • module_name – __name__ for the calling module.
  • level – Minimum logging level. Messages with this level or higher will be shown.
  • stream – ‘stream’ argument to logging.StreamHandler, typically sys.stdout or sys.stderr.
Raises:

ValueError – If stream is not ‘stderr’ or ‘stdout’.

putput.pipeline module

class putput.pipeline.Pipeline(pattern_def_path: pathlib.Path, *, dynamic_token_patterns_map: Optional[Mapping[str, Sequence[str]]] = None, token_handler_map: Optional[Mapping[str, Callable[[str, str], str]]] = None, group_handler_map: Optional[Mapping[str, Callable[[str, Sequence[str]], str]]] = None, expansion_hooks_map: Optional[Mapping[str, Sequence[Callable[[Sequence[Sequence[str]], Sequence[str], Sequence[Tuple[str, int]]], Tuple[Sequence[Sequence[str]], Sequence[str], Sequence[Tuple[str, int]]]]]]] = None, combo_hooks_map: Optional[Mapping[str, Sequence[Callable]]] = None, combo_options_map: Optional[Mapping[str, putput.joiner.ComboOptions]] = None, seed: Optional[int] = None)[source]

Bases: object

Transforms a pattern definition into labeled data.

To perform this transformation, initialize ‘Pipeline’ and call ‘flow’.

There are two ways to initialize ‘Pipeline’: by passing in desired arguments or through the use of a ‘preset’ in the method ‘from_preset’. ‘Presets’ instantiate the ‘Pipeline’ with arguments that cover common use cases. As these arguments become attributes that the user can modify, using a ‘preset’ does not give up customizability.

Once ‘Pipeline’ has been initialized, calling the method ‘flow’ will cause labeled data to flow through ‘Pipeline’ to the user.

There are two stages in ‘flow’. The first stage, ‘expansion’, expands the pattern definition file into an ‘utterance_combo’, ‘tokens’, and ‘groups’ for each utterance pattern. At the end of the first stage, if hooks in ‘expansion_hooks_map’ are specified for the current utterance pattern, they are applied in order where the output of a previous hook becomes the input to the next hook.

The second stage, ‘combination’, yields a sequence of ‘utterance’, ‘handled_tokens’, and ‘handled_groups’. This stage applies handlers from ‘token_handler_map’ and ‘group_handler_map’ and is subject to constraints specified in ‘combo_options_map’. At the end of the second stage, if hooks in ‘combo_hooks_map’ are specified for the current ‘utterance_pattern’, they are applied in order where the output of a previous hook becomes the input to the next hook.

Examples

Default behavior

>>> pattern_def_path = Path(__file__).parent.parent / 'tests' / 'doc' / 'example_pattern_definition.yml'
>>> dynamic_token_patterns_map = {'ITEM': ('fries',)}
>>> p = Pipeline(pattern_def_path, dynamic_token_patterns_map=dynamic_token_patterns_map)
>>> generator = p.flow(disable_progress_bar=True)
>>> for utterance, tokens, groups in generator:
...     print(utterance)
...     print(tokens)
...     print(groups)
can she get fries can she get fries and fries
('[ADD(can she get)]', '[ITEM(fries)]', '[ADD(can she get)]', '[ITEM(fries)]', '[CONJUNCTION(and)]',
'[ITEM(fries)]')
('{ADD_ITEM([ADD(can she get)] [ITEM(fries)])}', '{ADD_ITEM([ADD(can she get)] [ITEM(fries)])}',
'{None([CONJUNCTION(and)])}', '{None([ITEM(fries)])}')
can she get fries may she get fries and fries
('[ADD(can she get)]', '[ITEM(fries)]', '[ADD(may she get)]', '[ITEM(fries)]', '[CONJUNCTION(and)]',
'[ITEM(fries)]')
('{ADD_ITEM([ADD(can she get)] [ITEM(fries)])}', '{ADD_ITEM([ADD(may she get)] [ITEM(fries)])}',
'{None([CONJUNCTION(and)])}', '{None([ITEM(fries)])}')
may she get fries can she get fries and fries
('[ADD(may she get)]', '[ITEM(fries)]', '[ADD(can she get)]', '[ITEM(fries)]', '[CONJUNCTION(and)]',
'[ITEM(fries)]')
('{ADD_ITEM([ADD(may she get)] [ITEM(fries)])}', '{ADD_ITEM([ADD(can she get)] [ITEM(fries)])}',
'{None([CONJUNCTION(and)])}', '{None([ITEM(fries)])}')
may she get fries may she get fries and fries
('[ADD(may she get)]', '[ITEM(fries)]', '[ADD(may she get)]', '[ITEM(fries)]', '[CONJUNCTION(and)]',
'[ITEM(fries)]')
('{ADD_ITEM([ADD(may she get)] [ITEM(fries)])}', '{ADD_ITEM([ADD(may she get)] [ITEM(fries)])}',
'{None([CONJUNCTION(and)])}', '{None([ITEM(fries)])}')

With arguments

>>> import json
>>> import random
>>> def _just_tokens(token: str, _: str) -> str:
...     return '[{token}]'.format(token=token)
>>> def _just_groups(group_name: str, _: Sequence[str]) -> str:
...     return '[{group_name}]'.format(group_name=group_name)
>>> def _add_random_words(utterance: str,
...                       handled_tokens: Sequence[str],
...                       handled_groups: Sequence[str]
...                       ) -> Tuple[str, Sequence[str], Sequence[str]]:
...     utterances = utterance.split()
...     random_words = ['hmmmm', 'uh', 'um', 'please']
...     insert_index = random.randint(0, len(utterances))
...     random_word = random.choice(random_words)
...     utterances.insert(insert_index, random_word)
...     utterance = ' '.join(utterances)
...     return utterance, handled_tokens, handled_groups
>>> def _jsonify(utterance: str,
...              handled_tokens: Sequence[str],
...              handled_groups: Sequence[str]
...              ) -> str:
...     return json.dumps(dict(utterance=utterance,
...                            handled_tokens=handled_tokens,
...                            handled_groups=handled_groups),
...                       sort_keys=True)
>>> def _sample_utterance_combo(utterance_combo: Sequence[Sequence[str]],
...                             tokens: Sequence[str],
...                             groups: Sequence[Tuple[str, int]]
...                             ) -> Tuple[Sequence[Sequence[str]], Sequence[str], Sequence[Tuple[str, int]]]:
...        TOKEN_INDEX = tokens.index('ADD')
...        utterance_combo_list = list(utterance_combo)
...        sampled_combos = tuple(random.sample(utterance_combo_list.pop(TOKEN_INDEX), 1))
...        utterance_combo_list.insert(TOKEN_INDEX, sampled_combos)
...        utterance_combo = tuple(utterance_combo_list)
...        return utterance_combo, tokens, groups
>>> token_handler_map = {'ITEM': _just_tokens}
>>> group_handler_map = {'ADD_ITEM': _just_groups}
>>> expansion_hooks_map = {'ADD_ITEM, 2, CONJUNCTION, ITEM': (_sample_utterance_combo,)}
>>> combo_hooks_map = {'ADD_ITEM, 2, CONJUNCTION, ITEM': (_add_random_words, _add_random_words, _jsonify),
...                    'DEFAULT': (_jsonify,)}
>>> combo_options_map = {'DEFAULT': ComboOptions(max_sample_size=2, with_replacement=False)}
>>> p = Pipeline(pattern_def_path,
...              dynamic_token_patterns_map=dynamic_token_patterns_map,
...              token_handler_map=token_handler_map,
...              group_handler_map=group_handler_map,
...              expansion_hooks_map=expansion_hooks_map,
...              combo_hooks_map=combo_hooks_map,
...              combo_options_map=combo_options_map,
...              seed=0)
>>> for json_result in p.flow(disable_progress_bar=True):
...     print(json_result)
{"handled_groups": ["[ADD_ITEM]", "[ADD_ITEM]", "{None([CONJUNCTION(and)])}", "{None([ITEM])}"],
 "handled_tokens": ["[ADD(may she get)]", "[ITEM]", "[ADD(can she get)]", "[ITEM]", "[CONJUNCTION(and)]",
                    "[ITEM]"],
 "utterance": "may she get fries please can she hmmmm get fries and fries"}
{"handled_groups": ["[ADD_ITEM]", "[ADD_ITEM]", "{None([CONJUNCTION(and)])}", "{None([ITEM])}"],
 "handled_tokens": ["[ADD(may she get)]", "[ITEM]", "[ADD(may she get)]", "[ITEM]", "[CONJUNCTION(and)]",
                    "[ITEM]"],
 "utterance": "may she get fries may she um um get fries and fries"}

With a preset

>>> dynamic_token_patterns_map = {'ITEM': ('fries',)}
>>> p = Pipeline.from_preset('IOB2',
...                          pattern_def_path,
...                          dynamic_token_patterns_map=dynamic_token_patterns_map)
>>> generator = p.flow(disable_progress_bar=True)
>>> for utterance, tokens, groups in generator:
...     print(utterance)
...     print(tokens)
...     print(groups)
...     break
can she get fries can she get fries and fries
('B-ADD I-ADD I-ADD', 'B-ITEM', 'B-ADD I-ADD I-ADD', 'B-ITEM', 'B-CONJUNCTION', 'B-ITEM')
('B-ADD_ITEM I-ADD_ITEM I-ADD_ITEM I-ADD_ITEM', 'B-ADD_ITEM I-ADD_ITEM I-ADD_ITEM I-ADD_ITEM', 'B-None',
 'B-None')
combo_hooks_map

A mapping between an utterance pattern and hooks to apply after the combination phase. If ‘DEFAULT’ is specified as the utterance pattern, the hooks will apply to all utterance patterns not otherwise specified in the mapping. During, ‘flow’, hooks are applied in order where the output of the previous hook becomes the input to the next hook.

combo_options_map

A mapping between an utterance pattern and ComboOptions to apply during the combination phase. If ‘DEFAULT’ is specified as the utterance pattern, the options will apply to all utterance patterns not otherwise specified in the mapping.

dynamic_token_patterns_map

The dynamic counterpart to the static section in the pattern definition. This mapping between token and token patterns is useful in scenarios where tokens and token patterns cannot be known before runtime.

expansion_hooks_map

A mapping between an utterance pattern and hooks to apply after the expansion phase. If ‘DEFAULT’ is specified as the utterance pattern, the hooks will apply to all utterance patterns not otherwise specified in the mapping. During, ‘flow’, hooks are applied in order where the output of the previous hook becomes the input to the next hook.

flow(*, disable_progress_bar: bool = False) → Iterable[T_co][source]

Generates labeled data one utterance at a time.

Parameters:disable_progress_bar – Option to display progress of expansion and combination stages as the Iterable is consumed.
Yields:Labeled data.

Examples

>>> from pathlib import Path
>>> from putput.pipeline import Pipeline
>>> pattern_def_path = Path(__file__).parent.parent / 'tests' / 'doc' / 'example_pattern_definition.yml'
>>> dynamic_token_patterns_map = {'ITEM': ('fries',)}
>>> p = Pipeline(pattern_def_path, dynamic_token_patterns_map=dynamic_token_patterns_map)
>>> generator = p.flow(disable_progress_bar=True)
>>> for utterance, tokens, groups in generator:
...     print(utterance)
...     print(tokens)
...     print(groups)
...     break
can she get fries can she get fries and fries
('[ADD(can she get)]', '[ITEM(fries)]', '[ADD(can she get)]', '[ITEM(fries)]',
'[CONJUNCTION(and)]', '[ITEM(fries)]')
('{ADD_ITEM([ADD(can she get)] [ITEM(fries)])}', '{ADD_ITEM([ADD(can she get)] [ITEM(fries)])}',
'{None([CONJUNCTION(and)])}', '{None([ITEM(fries)])}')
classmethod from_preset(preset: Union[str, Callable, Sequence[Union[str, Callable]]], *args, **kwargs) → T_PIPELINE[source]

Instantiates ‘Pipeline’ from a preset configuration.

There are two ways to use ‘from_preset’. The simplest way is to use the preset’s name. However, presets may have optional arguments that allow for more control. In that case, use a call to the preset’s method, ‘preset’, with the desired arguments.

Parameters:
  • preset – A str that is the preset’s name, a Callable that is the result of calling the preset’s ‘preset’ function, or a Sequence of the two. The Callable form allows more control over the preset’s behavior. If a Sequence is specified, the result of calling the presets’ ‘preset’ function may only overlap in ‘combo_hooks_map’ and ‘expansion_hooks_map’. If there is overlap, functions will be applied in the order of the Sequence.
  • args – See __init__ docstring.
  • kwargs – See __init__ docstring.
Raises:

ValueError – If presets or kwargs contain the same keys, and those keys are not ‘combo_hooks_map’ or ‘expansion_hooks_map’.

Returns:

An instance of Pipeline.

Examples

Preset str

>>> from pathlib import Path
>>> from putput.pipeline import Pipeline
>>> pattern_def_path = Path(__file__).parent.parent / 'tests' / 'doc' / 'example_pattern_definition.yml'
>>> dynamic_token_patterns_map = {'ITEM': ('fries',)}
>>> p = Pipeline.from_preset('IOB2',
...                          pattern_def_path,
...                          dynamic_token_patterns_map=dynamic_token_patterns_map)
>>> generator = p.flow(disable_progress_bar=True)
>>> for utterance, tokens, groups in generator:
...     print(utterance)
...     print(tokens)
...     print(groups)
...     break
can she get fries can she get fries and fries
('B-ADD I-ADD I-ADD', 'B-ITEM', 'B-ADD I-ADD I-ADD', 'B-ITEM', 'B-CONJUNCTION', 'B-ITEM')
('B-ADD_ITEM I-ADD_ITEM I-ADD_ITEM I-ADD_ITEM', 'B-ADD_ITEM I-ADD_ITEM I-ADD_ITEM I-ADD_ITEM',
'B-None', 'B-None')

Preset function with arguments

>>> from putput.presets import iob2
>>> p = Pipeline.from_preset(iob2.preset(tokens_to_include=('ITEM',), groups_to_include=('ADD_ITEM',)),
...                          pattern_def_path,
...                          dynamic_token_patterns_map=dynamic_token_patterns_map)
>>> generator = p.flow(disable_progress_bar=True)
>>> for utterance, tokens, groups in generator:
...     print(utterance)
...     print(tokens)
...     print(groups)
...     break
can she get fries can she get fries and fries
('O O O', 'B-ITEM', 'O O O', 'B-ITEM', 'O', 'B-ITEM')
('B-ADD_ITEM I-ADD_ITEM I-ADD_ITEM I-ADD_ITEM', 'B-ADD_ITEM I-ADD_ITEM I-ADD_ITEM I-ADD_ITEM', 'O', 'O')
group_handler_map

A mapping between a group name and a function with args (group name, handled tokens) that returns a handled group. If ‘DEFAULT’ is specified as the group name, the handler will apply to all groups not otherwise specified in the mapping.

pattern_def_path

Read-only path to the pattern definition.

seed

Seed to control random behavior for Pipeline.

token_handler_map

A mapping between a token and a function with args (token, phrase to tokenize) that returns a handled token. If ‘DEFAULT’ is specified as the token, the handler will apply to all tokens not otherwise specified in the mapping.

putput.validator module

exception putput.validator.PatternDefinitionValidationError[source]

Bases: Exception

Exception that describes an invalid pattern defintion.

putput.validator.validate_pattern_def(pattern_def: Mapping[KT, VT_co]) → None[source]

Ensures the pattern definition is defined properly.

Parameters:pattern_def – A dictionary representation of the pattern definition.
Raises:PatternDefinitionValidationError – If the pattern definition file is invalid.

Examples

>>> from pathlib import Path
>>> from putput.pipeline import _load_pattern_def
>>> pattern_def_path = Path(__file__).parent.parent / 'tests' / 'doc' / 'example_pattern_definition.yml'
>>> pattern_def = _load_pattern_def(pattern_def_path)
>>> validate_pattern_def(pattern_def)

Module contents

Package settings for putput.