agents.rules package

Note

All rule should be imported into this module for insatiate to work

agents.rules.create_default_rules(gameframework: GameFramework | None = None, random_lane_change: bool = False) → Iterable[Rule]
Parameters:
Return type:

Iterable[Rule]

agents.rules.rule_from_config(cfg: CallFunctionFromConfig | DictConfig | CreateRuleFromConfig) → Rule | Iterable[Rule]

Instantiates Rules through Hydra’s instantiate function.

Note

The _target_ interface also allows to call functions, e.g. create_default_rules(), hence you need to check if the return value is a Rule or an Iterable[Rule]

Returns:

Rule or Iterable[Rule]

Parameters:

cfg (CallFunctionFromConfig | DictConfig | CreateRuleFromConfig)

Return type:

Rule | Iterable[Rule]

See also

Subpackages

Submodules

agents.rules.behaviour_templates module

agents.rules.behaviour_templates.if_config(config_path, value)

Returns a partial function that checks if a value in the config is set to a certain value.

agents.rules.behaviour_templates.set_default_intersection_speed(ctx: classes.rule.Context)

Slow down the car when turning at a junction.

Parameters:

ctx (classes.rule.Context)

class agents.rules.behaviour_templates.SlowDownAtIntersectionRule(phases: Phase | Iterable[Phase] | None | 'type[Any]' | Self | str | None = None, bases: tuple[type[Any], ...] | None = None, clsdict: dict[str, Any] | None = None, **kwargs: Any)

Bases: Rule

Slow down the car when turning at a junction.

Parameters:

  • phases (Union[Phase, Iterable[Phase]])

  • condition (Optional[ConditionFunctionLikeT])

  • action (Optional[Union[CallableAction[Self, []], Dict[Any, CallableAction[Self, []]]]])

  • false_action (Optional[CallableAction[Self, []]])

  • actions (Optional[Dict[Any, CallableAction[Self, []]]])

  • description (str)

  • overwrite_settings (Optional[Dict[str, Any]])

  • self_config (Optional[Dict[str, Any]])

  • priority (RulePriority)

  • cooldown_reset_value (Optional[int])

  • group (Optional[str])

  • enabled (bool)

phase: Phase = 8193

For the Class API the phase attribute be set to a single Phase object.

condition() → Literal[True]

This is an ConditionFunction that always returns True. It can be used to always execute an action.

Parameters:

ctx (classes.rule.Context)

Return type:

Literal[True]

action()

Slow down the car when turning at a junction.

Parameters:

ctx (classes.rule.Context)

overwrite_settings: Dict[str, Any] = {'speed': {'intersection_speed_decrease': 10}}

Settings that should overwrite the agent’s settings for this rule.

Note

The overwrite settings are primitive dict objects, omegaconf.DictConfig objects are converted.

description: str = 'Set speed to intersection speed'

Description of what this rule should do

__init__(phases: Phase | Iterable[Phase], condition: ConditionFunctionLikeT | None = None, action: classes.type_protocols.CallableActionT[Self, []] | Dict[Any, classes.type_protocols.CallableActionT[Self, []]] | None = None, false_action: classes.type_protocols.CallableActionT[Self, []] | None = None, *, actions: Dict[Any, classes.type_protocols.CallableActionT[Self, []]] | None = None, description: str = 'What does this rule do?', overwrite_settings: Dict[str, Any] | None = None, self_config: Dict[str, Any] | None = None, priority: RulePriority = RulePriority.NORMAL, cooldown_reset_value: int | None = None, group: str | None = None, enabled: bool = True)

Initializes a Rule object.

Parameters:

  • phases (Union[Phase, Iterable[Phase]]) – The phase(s) when the rule should be evaluated. An iterable of Phase objects or a single Phase object.

  • condition (Optional[ConditionFunctionLikeT]) – A function that takes a Context object as input and returns a Hashable value. If not provided, the class must implement a condition function.

  • action (Optional[Union[classes.type_protocols.CallableActionT[Self, []], Dict[Any, classes.type_protocols.CallableActionT[Self, []]]]]) – A function or a dictionary of functions that take a Context object as input. If action behaves like actions. Only one of action and actions can be set.

  • false_action (Optional[classes.type_protocols.CallableActionT[Self, []]]) – A function that takes a Context object as input and returns any value. It is used when action is a single function and represents the action to be taken when the condition is False.

  • actions (Optional[Dict[Any, classes.type_protocols.CallableActionT[Self, []]]]) – A dictionary of action functions It should map the return values of condition to the corresponding action function. If action is None, actions must be provided.

  • description (str) – A string that describes what this rule does.

  • overwrite_settings (Optional[Dict[str, Any]]) – A dictionary of settings that will overwrite the agent’s setting for this Rule.

  • priority (RulePriority) – The priority of the rule. It can be a float, an integer, or a RulePriority enum value.

  • cooldown_reset_value (Optional[int]) – An optional integer value that represents the cooldown reset value for the rule. If not provided falls back to the class attribute Rule.DEFAULT_COOLDOWN_RESET.

  • group (Optional[str]) – An optional string that specifies the group to which this rule belongs.

  • enabled (bool) – A boolean value indicating whether the rule is enabled or not.

  • self_config (Optional[Dict[str, Any]])

Raises:

  • ValueError – If phases is empty or None, or if phases contains an object that is not of type Phase.

  • TypeError – If condition is None and the class does not implement a condition() function, or if both action and actions are None and the class does not have an actions attribute or an action() function.

  • TypeError – if actions is not a Mapping object.

  • ValueError – If both action and actions are provided.

  • ValueError – If action is a Mapping and either false_action or actions is not None.

  • ValueError – If an action function is not callable.

  • ValueError – If description is not a string.

agents.rules.behaviour_templates.set_default_speed(ctx: classes.rule.Context)

Speed to apply when the car drives under normal circumstances, i.e. no junctions, no obstacles, etc. detected.

Parameters:

ctx (classes.rule.Context)

class agents.rules.behaviour_templates.NormalSpeedRule(phases: Phase | Iterable[Phase] | None | 'type[Any]' | Self | str | None = None, bases: tuple[type[Any], ...] | None = None, clsdict: dict[str, Any] | None = None, **kwargs: Any)

Bases: Rule

Speed to apply when the car drives under normal circumstances, i.e. no junctions, no obstacles, etc. detected.

Parameters:
phases: FrozenSet[Phase] = 257

The phase or phases in which the rule should be evaluated. For instantiation the phases attribute can be any Iterable [Phase].

condition() → Literal[True]

This is an ConditionFunction that always returns True. It can be used to always execute an action.

Parameters:

ctx (classes.rule.Context)

Return type:

Literal[True]

action()

Speed to apply when the car drives under normal circumstances, i.e. no junctions, no obstacles, etc. detected.

Parameters:

ctx (classes.rule.Context)

description: str = 'Set speed to normal speed'

Description of what this rule should do

__init__(phases: Phase | Iterable[Phase], condition: ConditionFunctionLikeT | None = None, action: classes.type_protocols.CallableActionT[Self, []] | Dict[Any, classes.type_protocols.CallableActionT[Self, []]] | None = None, false_action: classes.type_protocols.CallableActionT[Self, []] | None = None, *, actions: Dict[Any, classes.type_protocols.CallableActionT[Self, []]] | None = None, description: str = 'What does this rule do?', overwrite_settings: Dict[str, Any] | None = None, self_config: Dict[str, Any] | None = None, priority: RulePriority = RulePriority.NORMAL, cooldown_reset_value: int | None = None, group: str | None = None, enabled: bool = True)

Initializes a Rule object.

Parameters:

  • phases (Union[Phase, Iterable[Phase]]) – The phase(s) when the rule should be evaluated. An iterable of Phase objects or a single Phase object.

  • condition (Optional[ConditionFunctionLikeT]) – A function that takes a Context object as input and returns a Hashable value. If not provided, the class must implement a condition function.

  • action (Optional[Union[classes.type_protocols.CallableActionT[Self, []], Dict[Any, classes.type_protocols.CallableActionT[Self, []]]]]) – A function or a dictionary of functions that take a Context object as input. If action behaves like actions. Only one of action and actions can be set.

  • false_action (Optional[classes.type_protocols.CallableActionT[Self, []]]) – A function that takes a Context object as input and returns any value. It is used when action is a single function and represents the action to be taken when the condition is False.

  • actions (Optional[Dict[Any, classes.type_protocols.CallableActionT[Self, []]]]) – A dictionary of action functions It should map the return values of condition to the corresponding action function. If action is None, actions must be provided.

  • description (str) – A string that describes what this rule does.

  • overwrite_settings (Optional[Dict[str, Any]]) – A dictionary of settings that will overwrite the agent’s setting for this Rule.

  • priority (RulePriority) – The priority of the rule. It can be a float, an integer, or a RulePriority enum value.

  • cooldown_reset_value (Optional[int]) – An optional integer value that represents the cooldown reset value for the rule. If not provided falls back to the class attribute Rule.DEFAULT_COOLDOWN_RESET.

  • group (Optional[str]) – An optional string that specifies the group to which this rule belongs.

  • enabled (bool) – A boolean value indicating whether the rule is enabled or not.

  • self_config (Optional[Dict[str, Any]])

Raises:

  • ValueError – If phases is empty or None, or if phases contains an object that is not of type Phase.

  • TypeError – If condition is None and the class does not implement a condition() function, or if both action and actions are None and the class does not have an actions attribute or an action() function.

  • TypeError – if actions is not a Mapping object.

  • ValueError – If both action and actions are provided.

  • ValueError – If action is a Mapping and either false_action or actions is not None.

  • ValueError – If an action function is not callable.

  • ValueError – If description is not a string.

agents.rules.behaviour_templates.random_spawnpoint_destination(ctx: classes.rule.Context, waypoints: List[carla.Waypoint] | None = None)

Set a random waypoint as the next target.

Parameters:
agents.rules.behaviour_templates.is_agent_done(ctx: Context) → bool

Agent has reached its destination.

Parameters:

ctx (Context)

Return type:

bool

class agents.rules.behaviour_templates.TargetRandomSpawnpointWhenDone(phases: Phase | Iterable[Phase] | None | 'type[Any]' | Self | str | None = None, bases: tuple[type[Any], ...] | None = None, clsdict: dict[str, Any] | None = None, **kwargs: Any)

Bases: Rule

Sets random waypoint when done

Parameters:

  • phases (Union[Phase, typing.Iterable[Phase]])

  • condition (Optional[ConditionFunctionLikeT])

  • action (Optional[Union[classes.type_protocols.classes.type_protocols.CallableActionTT[Self, []], Dict[Any, classes.type_protocols.classes.type_protocols.CallableActionTT[Self, []]]]])

  • false_action (Optional[classes.type_protocols.classes.type_protocols.CallableActionTT[Self, []]])

  • actions (Optional[Dict[Any, classes.type_protocols.classes.type_protocols.CallableActionTT[Self, []]]])

  • description (str)

  • overwrite_settings (Optional[Dict[str, Any]])

  • self_config (Optional[Dict[str, Any]])

  • priority (RulePriority)

  • cooldown_reset_value (Optional[int])

  • group (Optional[str])

  • enabled (bool)

phases: FrozenSet[Phase] = 131073

The phase or phases in which the rule should be evaluated. For instantiation the phases attribute can be any Iterable [Phase].

condition() → bool

Agent has reached its destination.

Parameters:

ctx (Context)

Return type:

bool

action(waypoints: List[carla.Waypoint] | None = None)

Set a random waypoint as the next target.

Parameters:
description: str = 'Sets random waypoint when done'

Description of what this rule should do

__init__(phases: Phase | typing.Iterable[Phase], condition: ConditionFunctionLikeT | None = None, action: classes.type_protocols.classes.type_protocols.CallableActionTT[Self, []] | Dict[Any, classes.type_protocols.classes.type_protocols.CallableActionTT[Self, []]] | None = None, false_action: classes.type_protocols.classes.type_protocols.CallableActionTT[Self, []] | None = None, *, actions: Dict[Any, classes.type_protocols.classes.type_protocols.CallableActionTT[Self, []]] | None = None, description: str = 'What does this rule do?', overwrite_settings: Dict[str, Any] | None = None, self_config: Dict[str, Any] | None = None, priority: RulePriority = RulePriority.NORMAL, cooldown_reset_value: int | None = None, group: str | None = None, enabled: bool = True)

Initializes a Rule object.

Parameters:

  • phases (Union[Phase, typing.Iterable[Phase]]) – The phase(s) when the rule should be evaluated. An iterable of Phase objects or a single Phase object.

  • condition (Optional[ConditionFunctionLikeT]) – A function that takes a Context object as input and returns a Hashable value. If not provided, the class must implement a condition function.

  • action (Optional[Union[classes.type_protocols.classes.type_protocols.CallableActionTT[Self, []], Dict[Any, classes.type_protocols.classes.type_protocols.CallableActionTT[Self, []]]]]) – A function or a dictionary of functions that take a Context object as input. If action behaves like actions. Only one of action and actions can be set.

  • false_action (Optional[classes.type_protocols.classes.type_protocols.CallableActionTT[Self, []]]) – A function that takes a Context object as input and returns any value. It is used when action is a single function and represents the action to be taken when the condition is False.

  • actions (Optional[Dict[Any, classes.type_protocols.classes.type_protocols.CallableActionTT[Self, []]]]) – A dictionary of action functions It should map the return values of condition to the corresponding action function. If action is None, actions must be provided.

  • description (str) – A string that describes what this rule does.

  • overwrite_settings (Optional[Dict[str, Any]]) – A dictionary of settings that will overwrite the agent’s setting for this Rule.

  • priority (RulePriority) – The priority of the rule. It can be a float, an integer, or a RulePriority enum value.

  • cooldown_reset_value (Optional[int]) – An optional integer value that represents the cooldown reset value for the rule. If not provided falls back to the class attribute Rule.DEFAULT_COOLDOWN_RESET.

  • group (Optional[str]) – An optional string that specifies the group to which this rule belongs.

  • enabled (bool) – A boolean value indicating whether the rule is enabled or not.

  • self_config (Optional[Dict[str, Any]])

Raises:

  • ValueError – If phases is empty or None, or if phases contains an object that is not of type Phase.

  • TypeError – If condition is None and the class does not implement a condition() function, or if both action and actions are None and the class does not have an actions attribute or an action() function.

  • TypeError – if actions is not a Mapping object.

  • ValueError – If both action and actions are provided.

  • ValueError – If action is a Mapping and either false_action or actions is not None.

  • ValueError – If an action function is not callable.

  • ValueError – If description is not a string.

agents.rules.behaviour_templates.set_next_waypoint_nearby(ctx: classes.rule.Context)
Parameters:

ctx (classes.rule.Context)

class agents.rules.behaviour_templates.SetNextWaypointNearby(phases: Phase | Iterable[Phase] | None | 'type[Any]' | Self | str | None = None, bases: tuple[type[Any], ...] | None = None, clsdict: dict[str, Any] | None = None, **kwargs: Any)

Bases: Rule

Sets random waypoint when done to a nearby point ahead

Parameters:

  • phases (Union[Phase, typing.typing.Iterable[Phase]])

  • condition (Optional[ConditionFunctionLikeT])

  • action (Optional[Union[classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTT[Self, []], Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTT[Self, []]]]])

  • false_action (Optional[classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTT[Self, []]])

  • actions (Optional[Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTT[Self, []]]])

  • description (str)

  • overwrite_settings (Optional[Dict[str, Any]])

  • self_config (Optional[Dict[str, Any]])

  • priority (RulePriority)

  • cooldown_reset_value (Optional[int])

  • group (Optional[str])

  • enabled (bool)

phases: FrozenSet[Phase] = 131073

The phase or phases in which the rule should be evaluated. For instantiation the phases attribute can be any Iterable [Phase].

condition() → bool

Agent has reached its destination.

Parameters:

ctx (Context)

Return type:

bool

action()
Parameters:

ctx (classes.rule.Context)

__init__(phases: Phase | typing.typing.Iterable[Phase], condition: ConditionFunctionLikeT | None = None, action: classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTT[Self, []] | Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTT[Self, []]] | None = None, false_action: classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTT[Self, []] | None = None, *, actions: Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTT[Self, []]] | None = None, description: str = 'What does this rule do?', overwrite_settings: Dict[str, Any] | None = None, self_config: Dict[str, Any] | None = None, priority: RulePriority = RulePriority.NORMAL, cooldown_reset_value: int | None = None, group: str | None = None, enabled: bool = True)

Initializes a Rule object.

Parameters:

  • phases (Union[Phase, typing.typing.Iterable[Phase]]) – The phase(s) when the rule should be evaluated. An iterable of Phase objects or a single Phase object.

  • condition (Optional[ConditionFunctionLikeT]) – A function that takes a Context object as input and returns a Hashable value. If not provided, the class must implement a condition function.

  • action (Optional[Union[classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTT[Self, []], Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTT[Self, []]]]]) – A function or a dictionary of functions that take a Context object as input. If action behaves like actions. Only one of action and actions can be set.

  • false_action (Optional[classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTT[Self, []]]) – A function that takes a Context object as input and returns any value. It is used when action is a single function and represents the action to be taken when the condition is False.

  • actions (Optional[Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTT[Self, []]]]) – A dictionary of action functions It should map the return values of condition to the corresponding action function. If action is None, actions must be provided.

  • description (str) – A string that describes what this rule does.

  • overwrite_settings (Optional[Dict[str, Any]]) – A dictionary of settings that will overwrite the agent’s setting for this Rule.

  • priority (RulePriority) – The priority of the rule. It can be a float, an integer, or a RulePriority enum value.

  • cooldown_reset_value (Optional[int]) – An optional integer value that represents the cooldown reset value for the rule. If not provided falls back to the class attribute Rule.DEFAULT_COOLDOWN_RESET.

  • group (Optional[str]) – An optional string that specifies the group to which this rule belongs.

  • enabled (bool) – A boolean value indicating whether the rule is enabled or not.

  • self_config (Optional[Dict[str, Any]])

Raises:

  • ValueError – If phases is empty or None, or if phases contains an object that is not of type Phase.

  • TypeError – If condition is None and the class does not implement a condition() function, or if both action and actions are None and the class does not have an actions attribute or an action() function.

  • TypeError – if actions is not a Mapping object.

  • ValueError – If both action and actions are provided.

  • ValueError – If action is a Mapping and either false_action or actions is not None.

  • ValueError – If an action function is not callable.

  • ValueError – If description is not a string.

description: str = 'Sets random waypoint when done to a nearby point ahead'

Description of what this rule should do

agents.rules.behaviour_templates.accept_rss_updates(ctx: Context)

Accept RSS updates from the RSS manager.

Parameters:

ctx (Context)

class agents.rules.behaviour_templates.AlwaysAcceptRSSUpdates(phases: Phase | Iterable[Phase] | None | 'type[Any]' | Self | str | None = None, bases: tuple[type[Any], ...] | None = None, clsdict: dict[str, Any] | None = None, **kwargs: Any)

Bases: Rule

Always accept RSS updates if rss is enabled in the config.

Parameters:

  • phases (Union[Phase, typing.typing.typing.Iterable[Phase]])

  • condition (Optional[ConditionFunctionLikeT])

  • action (Optional[Union[classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTT[Self, []], Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTT[Self, []]]]])

  • false_action (Optional[classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTT[Self, []]])

  • actions (Optional[Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTT[Self, []]]])

  • description (str)

  • overwrite_settings (Optional[Dict[str, Any]])

  • self_config (Optional[Dict[str, Any]])

  • priority (RulePriority)

  • cooldown_reset_value (Optional[int])

  • group (Optional[str])

  • enabled (bool)

phases: FrozenSet[Phase] = 514

The phase or phases in which the rule should be evaluated. For instantiation the phases attribute can be any Iterable [Phase].

condition(config_path: str, value) → bool

Check if a value in the config is set to a certain value.

Parameters:
Return type:

bool

action()

Accept RSS updates from the RSS manager.

Parameters:

ctx (Context)

description: str = 'Always accepts the updates calculated by the RSS System.'

Description of what this rule should do

__init__(phases: Phase | typing.typing.typing.Iterable[Phase], condition: ConditionFunctionLikeT | None = None, action: classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTT[Self, []] | Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTT[Self, []]] | None = None, false_action: classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTT[Self, []] | None = None, *, actions: Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTT[Self, []]] | None = None, description: str = 'What does this rule do?', overwrite_settings: Dict[str, Any] | None = None, self_config: Dict[str, Any] | None = None, priority: RulePriority = RulePriority.NORMAL, cooldown_reset_value: int | None = None, group: str | None = None, enabled: bool = True)

Initializes a Rule object.

Parameters:

  • phases (Union[Phase, typing.typing.typing.Iterable[Phase]]) – The phase(s) when the rule should be evaluated. An iterable of Phase objects or a single Phase object.

  • condition (Optional[ConditionFunctionLikeT]) – A function that takes a Context object as input and returns a Hashable value. If not provided, the class must implement a condition function.

  • action (Optional[Union[classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTT[Self, []], Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTT[Self, []]]]]) – A function or a dictionary of functions that take a Context object as input. If action behaves like actions. Only one of action and actions can be set.

  • false_action (Optional[classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTT[Self, []]]) – A function that takes a Context object as input and returns any value. It is used when action is a single function and represents the action to be taken when the condition is False.

  • actions (Optional[Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTT[Self, []]]]) – A dictionary of action functions It should map the return values of condition to the corresponding action function. If action is None, actions must be provided.

  • description (str) – A string that describes what this rule does.

  • overwrite_settings (Optional[Dict[str, Any]]) – A dictionary of settings that will overwrite the agent’s setting for this Rule.

  • priority (RulePriority) – The priority of the rule. It can be a float, an integer, or a RulePriority enum value.

  • cooldown_reset_value (Optional[int]) – An optional integer value that represents the cooldown reset value for the rule. If not provided falls back to the class attribute Rule.DEFAULT_COOLDOWN_RESET.

  • group (Optional[str]) – An optional string that specifies the group to which this rule belongs.

  • enabled (bool) – A boolean value indicating whether the rule is enabled or not.

  • self_config (Optional[Dict[str, Any]])

Raises:

  • ValueError – If phases is empty or None, or if phases contains an object that is not of type Phase.

  • TypeError – If condition is None and the class does not implement a condition() function, or if both action and actions are None and the class does not have an actions attribute or an action() function.

  • TypeError – if actions is not a Mapping object.

  • ValueError – If both action and actions are provided.

  • ValueError – If action is a Mapping and either false_action or actions is not None.

  • ValueError – If an action function is not callable.

  • ValueError – If description is not a string.

class agents.rules.behaviour_templates.ConfigBasedRSSUpdates(phases: Phase | Iterable[Phase] | None | 'type[Any]' | Self | str | None = None, bases: tuple[type[Any], ...] | None = None, clsdict: dict[str, Any] | None = None, **kwargs: Any)

Bases: Rule

Always accept RSS updates if rss.always_accept_update is set to True in the config.

Parameters:

  • phases (Union[Phase, typing.typing.typing.typing.Iterable[Phase]])

  • condition (Optional[ConditionFunctionLikeT])

  • action (Optional[Union[classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTTT[Self, []], Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTTT[Self, []]]]])

  • false_action (Optional[classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTTT[Self, []]])

  • actions (Optional[Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTTT[Self, []]]])

  • description (str)

  • overwrite_settings (Optional[Dict[str, Any]])

  • self_config (Optional[Dict[str, Any]])

  • priority (RulePriority)

  • cooldown_reset_value (Optional[int])

  • group (Optional[str])

  • enabled (bool)

phases: FrozenSet[Phase] = 514

The phase or phases in which the rule should be evaluated. For instantiation the phases attribute can be any Iterable [Phase].

condition(config_path: str, value) → bool

Check if a value in the config is set to a certain value.

Parameters:
Return type:

bool

action()

Accept RSS updates from the RSS manager.

Parameters:

ctx (Context)

__init__(phases: Phase | typing.typing.typing.typing.Iterable[Phase], condition: ConditionFunctionLikeT | None = None, action: classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTTT[Self, []] | Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTTT[Self, []]] | None = None, false_action: classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTTT[Self, []] | None = None, *, actions: Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTTT[Self, []]] | None = None, description: str = 'What does this rule do?', overwrite_settings: Dict[str, Any] | None = None, self_config: Dict[str, Any] | None = None, priority: RulePriority = RulePriority.NORMAL, cooldown_reset_value: int | None = None, group: str | None = None, enabled: bool = True)

Initializes a Rule object.

Parameters:

  • phases (Union[Phase, typing.typing.typing.typing.Iterable[Phase]]) – The phase(s) when the rule should be evaluated. An iterable of Phase objects or a single Phase object.

  • condition (Optional[ConditionFunctionLikeT]) – A function that takes a Context object as input and returns a Hashable value. If not provided, the class must implement a condition function.

  • action (Optional[Union[classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTTT[Self, []], Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTTT[Self, []]]]]) – A function or a dictionary of functions that take a Context object as input. If action behaves like actions. Only one of action and actions can be set.

  • false_action (Optional[classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTTT[Self, []]]) – A function that takes a Context object as input and returns any value. It is used when action is a single function and represents the action to be taken when the condition is False.

  • actions (Optional[Dict[Any, classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.classes.type_protocols.CallableActionTTTTT[Self, []]]]) – A dictionary of action functions It should map the return values of condition to the corresponding action function. If action is None, actions must be provided.

  • description (str) – A string that describes what this rule does.

  • overwrite_settings (Optional[Dict[str, Any]]) – A dictionary of settings that will overwrite the agent’s setting for this Rule.

  • priority (RulePriority) – The priority of the rule. It can be a float, an integer, or a RulePriority enum value.

  • cooldown_reset_value (Optional[int]) – An optional integer value that represents the cooldown reset value for the rule. If not provided falls back to the class attribute Rule.DEFAULT_COOLDOWN_RESET.

  • group (Optional[str]) – An optional string that specifies the group to which this rule belongs.

  • enabled (bool) – A boolean value indicating whether the rule is enabled or not.

  • self_config (Optional[Dict[str, Any]])

Raises:

  • ValueError – If phases is empty or None, or if phases contains an object that is not of type Phase.

  • TypeError – If condition is None and the class does not implement a condition() function, or if both action and actions are None and the class does not have an actions attribute or an action() function.

  • TypeError – if actions is not a Mapping object.

  • ValueError – If both action and actions are provided.

  • ValueError – If action is a Mapping and either false_action or actions is not None.

  • ValueError – If an action function is not callable.

  • ValueError – If description is not a string.

description: str = 'Always accept RSS updates if :any:`rss.always_accept_update <LunaticAgentSettings.rss>` is set to True in the config.'

Description of what this rule should do

agents.rules.stopped_long_trigger module

class agents.rules.stopped_long_trigger.StoppedTooLongTrigger(phases: Phase | Iterable[Phase] | None | 'type[Any]' | Self | str | None = None, bases: tuple[type[Any], ...] | None = None, clsdict: dict[str, Any] | None = None, **kwargs: Any)

Bases: MultiRule

Triggers child rules if the agent has stopped for a too long time

Parameters:
phase: Phase = 6

For the Class API the phase attribute be set to a single Phase object.

stop_time_threshold = 60

Time in seconds the agent is allowed to stop before triggering the rule.

condition(ctx: Context) → bool
Parameters:

ctx (Context)

Return type:

bool

action(ctx: Context)
Parameters:

ctx (Context)

__init__(phases: Phase | Iterable[Phase], rules: List[Rule], condition: ConditionFunctionLikeT | None = None, *, description: str = 'If its own condition is true calls the passed rules.', priority: RulePriority = RulePriority.NORMAL, sort_rules: bool = True, execute_all_rules: bool = False, action: Callable[[Context], Any] | None = None, ignore_phase: bool = True, overwrite_settings: Dict[str, Any] | None = None, self_config: Dict[str, Any] | None = None, cooldown_reset_value: int | None = None, group: str | None = None, enabled: bool = True)

Initializes a Rule object that can have further rules as children.

Parameters:

  • phases (Union[Phase, Iterable]) – The phase or phases in which the rule should be active.

  • rules (List[Rule]) – The list of child rules to be called if the rule’s condition is true.

  • condition (Callable[[Context]], optional) – The condition that determines if the rules should be evaluated. Defaults to always_execute().

  • execute_all_rules (bool, optional) – If False will only execute the first rule with a applicable condition, i.e. this MultiRule is like a node in a decision tree. If True all rules are evaluated, unless one raises a DoNotEvaluateChildRules exception. Defaults to False.

  • sort_rules (bool, optional) – Flag indicating whether to sort the rules by priority. Defaults to True.

  • action (Callable[[Context]], optional) – The action to be executed before the passed rules are evaluated. Defaults to None.

  • ignore_phase (bool, optional) – Flag indicating whether to ignore the Phase of the passed child rules. Defaults to True.

  • overwrite_settings (Dict[str, Any], optional) – Additional settings to overwrite the agent’s settings. Defaults to None.

  • priority (RulePriority, optional) – The priority of the rule. RulePriority.NORMAL.

  • description (str, optional) – The description of the rule. Defaults to "If its own rule is true calls the passed rules.".

  • group (str | None, optional) – The group name of the rule. Defaults to None.

  • enabled (bool, optional) – Flag indicating whether the rule is enabled after creation. Defaults to True.

  • self_config (Optional[Dict[str, Any]])

  • cooldown_reset_value (Optional[int])

description: str = 'Triggers child rules if the agent has stopped for a too long time'

Description of what this rule should do

rules: List[Rule] = []

The list of child rules to be called if this rule’s condition is true.