cognition.decision.core¶

Base decision process

Attributes¶

`IO_REMOVE`	Sentinel to used in IO setting to indicate a removal
`Elaborator`	Monotonically summarizes current state
`TerminationCheck`	Detects decision process termination based upon current state
`Action`	Changes state via mutation (return `None`) or replacement (return `S`)
`ActionFactory`	Identifies viable actions in the current state
`ActionEvaluator`	Produces rankings of candidate actions

Exceptions¶

DecisionProcessExecutionError

A custom exception related to invalid decision process execution

Classes¶

`IOContainer`	Convenience bundling of input/output data access
`Phase`	Representation of decision process phases
`ActionRank`	Pairs an action and rank
`DecisionProcessErrorMessage`	Known errors with messages
`BaseDecisionProcess`	Orchestration for a sequential decision-making problem with state type `S`
`DecisionProcessIterator`	Custom iterator to facilitate easy decision process iteration via phase or cycle

Module Contents¶

cognition.decision.core.IO_REMOVE: Final[object]¶: Sentinel to used in IO setting to indicate a removal

class cognition.decision.core.IOContainer¶

Convenience bundling of input/output data access

i: cognition.util.misc.AttrReferral¶: .key access to (i)nput data

o: cognition.util.misc.AttrReferral¶: .key access to (o)utput channels

e: cognition.util.misc.AttrReferral¶: .key access to (e)laborated values

a: cognition.util.misc.AttrReferral¶: .key access to (a)rguments

input: types.MappingProxyType[str, Any]¶: Mapping view of i

output: types.MappingProxyType[str, Any]¶: Mapping view of o

elab: types.MappingProxyType[str, Any]¶: Mapping view of e

args: types.MappingProxyType[str, Any]¶: Mapping view of a

__post_init__()¶

Return type:: None

class cognition.decision.core.Phase¶

Bases: enum.IntEnum

Representation of decision process phases

Initialize self. See help(type(self)) for accurate signature.

ELABORATION = 0¶: Monotonic summarization of state

TERMINATIONCHECK = 1¶: Detect decision process termination

PROPOSE = 2¶: Factories to produce candidate actions

RANK = 3¶: Select an action (via evaluator action rankings)

APPLY = 4¶: Execute the selected action

property next: Phase¶

Cyclic enumeration

Returns:: next phase
Return type:: Phase

type cognition.decision.core.Elaborator = BiFunction[S, IOContainer, dict[str, Any]]¶: Monotonically summarizes current state

type cognition.decision.core.TerminationCheck = BiPredicate[S, IOContainer]¶: Detects decision process termination based upon current state

type cognition.decision.core.Action = BiFunction[S, IOContainer, S | None]¶: Changes state via mutation (return None) or replacement (return S)

type cognition.decision.core.ActionFactory = BiFunction[S, IOContainer, Action[S] | Iterable[Action[S]]]¶: Identifies viable actions in the current state

class cognition.decision.core.ActionRank[S]¶

Pairs an action and rank

a: Action[S]¶: Action

rank: cognition.util.misc.ImplementsLessThan¶: Rank of the action (smaller is better)

__str__()¶

Return type:: str

__lt__(other)¶

Smaller rank values come first, with a deterministic ordering in ties facilitated by action string representations

Parameters:: other (object) – self < other
Returns:: True if comparison holds
Return type:: bool

type cognition.decision.core.ActionEvaluator = TriFunction[S, IOContainer, Iterable[Action[S]], Iterable[ActionRank[S]]]¶: Produces rankings of candidate actions

class cognition.decision.core.DecisionProcessErrorMessage¶

Bases: enum.StrEnum

Known errors with messages

Initialize self. See help(type(self)) for accurate signature.

NO_PROPOSAL = 'No potential actions'¶: No candidate actions produced from factories

NO_RANK = 'No action rankings'¶: No rankings produced (given multiple candidate actions)

NO_CHOICE = 'No chosen action'¶: No action chosen to apply (should not occur)

exception cognition.decision.core.DecisionProcessExecutionError(msg)¶

Bases: Exception

A custom exception related to invalid decision process execution

Parameters:: msg (DecisionProcessErrorMessage) – decision process error message

Initialize self. See help(type(self)) for accurate signature.

property msg: DecisionProcessErrorMessage¶

Returns:: decision process error message
Return type:: DecisionProcessErrorMessage

class cognition.decision.core.BaseDecisionProcess[S](state_initializer)¶

Orchestration for a sequential decision-making problem with state type S

Parameters:: state_initializer (cognition.util.functypes.Supplier[S]) – produces state initially (and on reinit())

INPUT_KEY_TIME: Final[str] = 'clock'¶: Key associated with cycle input data

INPUT_ATTR_TIME: Final[str] = 'cycles'¶: Attribute produced by the cycle input data

OUTPUT_KEY_LOG: Final[str] = 'log'¶: Key associated with the log output channel

__getstate__()¶

__setstate__(state)¶

reinit()¶

Restarts the decision process

Returns:: this decision process (for chaining)
Return type:: Self

__str__()¶

Return type:: str

property phase: Phase¶

Returns:: current decision process phase
Return type:: Phase

property done: bool¶

Returns:: True if any termination check has returned True
Return type:: bool

property state: S¶

Returns:: current decision process state
Return type:: S

property io: IOContainer¶

Returns:: current decision process io (for debugging)
Return type:: IOContainer

property num_cycles: int¶

Returns:: how many decision process cycles have occurred since last initialization
Return type:: int

property chosen_action: str | None¶

Returns:: str() of the most recently chosen action
Return type:: str | None

add_elaborator(e)¶

Adds a state summarizer to the decision process

Parameters:: e (Elaborator[S]) – elaborator to add
Returns:: this decision process (for chaining)
Return type:: Self

elaborator(e)¶

Decorator version of add_elaborator()

Parameters:: e (Elaborator[S]) – elaborator to add
Returns:: added elaborator
Return type:: Elaborator[S]

add_termination_check(p)¶

Add a state predicate to identify a cause of decision process termination

Parameters:: p (TerminationCheck[S]) – predicate to detect termination
Returns:: this decision process (for chaining)
Return type:: Self

termination_check(p)¶

Decorator version of add_termination_check()

Parameters:: p (TerminationCheck[S]) – predicate to add
Returns:: added predicate
Return type:: TerminationCheck[S]

add_action_factory(f)¶

Adds a factory to propose potential action(s) given current state

Parameters:: f (ActionFactory[S]) – factory to add
Returns:: this decision process (for chaining)
Return type:: Self

action_factory(f)¶

Decorator version of add_action_factory()

Parameters:: f (ActionFactory[S]) – factory to add
Returns:: added factory
Return type:: ActionFactory[S]

add_action_evaluator(ae)¶

Adds an evaluator of potential actions

Parameters:: ae (ActionEvaluator[S]) – evaluator to add
Returns:: this decision process (for chaining)
Return type:: Self

action_evaluator(ae)¶

Decorator version of add_action_evaluator()

Parameters:: ae (ActionEvaluator[S]) – evaluator to add
Returns:: added evaluator
Return type:: ActionEvaluator[S]

run_phase()¶

Executes the current decision process phase

Returns:: this decision process (for chaining)
Return type:: Self

run_cycles(n=1)¶

Executes (up to) n cycles of the full phases

Parameters:: n (int) – number of phases to run
Returns:: this decision process (for chaining)
Return type:: Self

run_until_done()¶

Runs until decision process completion

Returns:: this decision process (for chaining)
Return type:: Self

phases()¶

Facilitates iteration by phase

Returns:: (potentially infinite) iterator over phases
Return type:: collections.abc.Iterator[Self]

cycles()¶

Facilitates iteration by cycle

Returns:: (potentially infinite) iterator over cycles
Return type:: collections.abc.Iterator[Self]

set_input_data(input_key, data)¶

Sets value of io.i.input_key

Parameters:

name – input data key
buffer – arbitrary object reference (or IO_REMOVE to remove)
input_key (str)
data (Any)

Returns:

this decision process (for chaining)

Return type:

Self

set_output_channel(output_key, data)¶

Sets value of io.o.output_key

Parameters:

name – output channel key
buffer – arbitrary object reference (or IO_REMOVE to remove)
output_key (str)
data (Any)

Returns:

this decision process (for chaining)

Return type:

Self

set_arg_value(arg_key, data)¶

Sets value of io.a.arg_key

Parameters:

name – argument key
buffer – arbitrary object reference (or IO_REMOVE to remove)
arg_key (str)
data (Any)

Returns:

this decision process (for chaining)

Return type:

Self

property log: str¶

Returns:: any data provided to the OUTPUT_KEY_LOG channel
Return type:: str

clear_log()¶

Clears any data provided to the OUTPUT_KEY_LOG channel

Returns:: this decision process (for chaining)
Return type:: Self

class cognition.decision.core.DecisionProcessIterator[S, DP: BaseDecisionProcess[S]](dp, by_phase=True)¶

Bases: collections.abc.Iterator[DP]

Custom iterator to facilitate easy decision process iteration via phase or cycle

Parameters:

dp (DP) – associated decision process
by_phase (bool) – True if iteration by phase; by cycle otherwise

__next__()¶

Provides the next phase/cycle if the decision process is not complete

Return type:: DP