Modules#

class mma_wrapper.label_manager.label_manager(action_space=None, observation_space=None)[source]#

Bases: object

Manage observation/action one-hot decode/encode as well as labeling observation/action and one-hot encoded observations or actions.

static from_dict(d)[source]#

Return type:: label_manager

label_action(action, agent=None)[source]#

Label a one-hot encoded action into label :type action: Union[int, ndarray] :param action: a one-hot encoded action

Returns:: the labelized action
Return type:: label

label_observation(observation, agent=None)[source]#

Label a one-hot encoded observation into label :type observation: Union[int, ndarray] :param observation: a one-hot encoded observation

Returns:: the labelized observation
Return type:: label

label_trajectory(trajectory, agent=None)[source]#

Label a one-hot encoded trajectory into a labeled trajectory :type trajectory: List[Tuple[Union[int, ndarray], Union[int, ndarray]]] :param trajectory: a one-hot encoded trajectory

Return type:: Union[List[Tuple[str, str]], List[str], str]

Returns: trajectory: a labeled trajectory

one_hot_decode_action(action, agent=None)[source]#

One-hot decode an action :type action: Union[int, ndarray] :param action: a one-hot encoded action

Returns:: a readable action
Return type:: action

one_hot_decode_observation(observation, agent=None)[source]#

One-hot decode an observation :type observation: Union[int, ndarray] :param observation: a one-hot encoded observation

Returns:: a readable observation
Return type:: observation

one_hot_decode_trajectory(trajectory, agent=None)[source]#

One-hot decode a one-hot encoded trajectory into a trajectory of readable (observation, action) couples

Return type:: List[Tuple[Any, Any]]

one_hot_encode_action(action, agent=None)[source]#

One-hot encode an action :type action: Any :param action: a readable action

Returns:: a one-hot encoded action
Return type:: action

one_hot_encode_observation(observation, agent=None)[source]#

One-hot encode an observation :type observation: Any :param observation: a readable observation

Returns:: a one-hot encoded observation
Return type:: observation

one_hot_encode_trajectory(trajectory, agent=None)[source]#

One-hot encode a readable trajectory into a one-hot encoded trajectory :type trajectory: List[Tuple[Any, Any]] :param trajectory: a readable trajectory

Return type:: List[Tuple[Union[int, ndarray], Union[int, ndarray]]]

Returns: trajectory: a one-hot encoded trajectory

to_dict(save_source=False)[source]#

Return type:: Dict

unlabel_action(action, agent=None)[source]#

Unlabel a one-hot encoded action into a list of one-hot encoded action :type action: str :param action: the label to be mapped to the matching action

Returns:: a list of one-hot encoded action
Return type:: List[action]

unlabel_observation(observation, agent=None)[source]#

Unlabel a one-hot encoded observation into a list of one-hot encoded observation :type observation: str :param observation: the label to be mapped to the matching observations

Returns:: a list of one-hot encoded observations
Return type:: List[observation]

unlabel_trajectory(labeled_trajectory, agent=None)[source]#

Unlabel a labeled trajectory into a one-hot encoded trajectory :param trajectory: a labeled trajectory

Return type:: List[Tuple[Union[int, ndarray], Union[int, ndarray]]]

Returns: trajectory: a one-hot encoded trajectory

class mma_wrapper.organizational_model.compatibility(source, destination)[source]#

Bases: organizational_specification

destination: role#

static from_dict(d)[source]#

Return type:: compatibility

source: role#

class mma_wrapper.organizational_model.deontic_specification(role, agents, missions, time_constraint=time_constraint_type.ANY)[source]#

Bases: organizational_specification

agents: List[str]#

static from_dict(d)[source]#

Return type:: deontic_specification

missions: List[mission]#

role: role#

time_constraint: time_constraint_type = 'ANY'#

class mma_wrapper.organizational_model.deontic_specifications(permissions, obligations)[source]#

Bases: organizational_specification

static from_dict(d)[source]#

Return type:: deontic_specifications

obligations: List[deontic_specification]#

permissions: List[deontic_specification]#

class mma_wrapper.organizational_model.functional_specifications(goals, social_scheme, mission_preferences)[source]#

Bases: organizational_specification

static from_dict(d)[source]#

Return type:: functional_specifications

goals: Dict[goal, goal_logic]#

mission_preferences: List[mission_preference]#

social_scheme: Dict[social_scheme_tag, social_scheme]#

class mma_wrapper.organizational_model.goal[source]#: Bases: str, organizational_specification

class mma_wrapper.organizational_model.group_specifications(roles, sub_groups, intra_links, inter_links, intra_compatibilities, inter_compatibilities, role_cardinalities, sub_group_cardinalities)[source]#

Bases: organizational_specification

static from_dict(d)[source]#

Return type:: group_specifications

inter_compatibilities: List[compatibility]#

inter_links: List[link]#

intra_compatibilities: List[compatibility]#

intra_links: List[link]#

role_cardinalities: Dict[role, cardinality]#

roles: List[role]#

sub_group_cardinalities: Dict[group_tag, cardinality]#

sub_groups: Dict[group_tag, group_specifications]#

class mma_wrapper.organizational_model.group_tag[source]#: Bases: str

class mma_wrapper.organizational_model.link(source, destination, type)[source]#

Bases: organizational_specification

destination: role#

static from_dict(d)[source]#

Return type:: link

source: role#

type: link_type#

class mma_wrapper.organizational_model.link_type(value)[source]#

Bases: str, Enum

An enumeration.

ACQ = 'ACQ'#

AUT = 'AUT'#

COM = 'COM'#

class mma_wrapper.organizational_model.mission[source]#: Bases: str, organizational_specification

class mma_wrapper.organizational_model.mission_preference(prefer, over)[source]#

Bases: organizational_specification

static from_dict(d)[source]#

Return type:: mission_preference

over: mission#

prefer: mission#

class mma_wrapper.organizational_model.organizational_model(structural_specifications, functional_specifications, deontic_specifications)[source]#

Bases: object

deontic_specifications: deontic_specifications#

static from_dict(d)[source]#

Return type:: organizational_model

functional_specifications: functional_specifications#

structural_specifications: structural_specifications#

to_dict()[source]#

Return type:: Dict

class mma_wrapper.organizational_model.organizational_specification[source]#

Bases: object

The basic class

class mma_wrapper.organizational_model.os_encoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]#

Bases: JSONEncoder

default(o)[source]#

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)

class mma_wrapper.organizational_model.plan(goal, sub_plan=None, probability=1.0)[source]#

Bases: organizational_specification

static from_dict(d)[source]#

Return type:: plan

goal: goal#

probability: float = 1.0#

sub_plan: sub_plan = None#

class mma_wrapper.organizational_model.plan_operator(value)[source]#

Bases: str, Enum

An enumeration.

CHOICE = 'CHOICE'#

PARALLEL = 'PARALLEL'#

SEQUENCE = 'SEQUENCE'#

class mma_wrapper.organizational_model.role[source]#: Bases: str, organizational_specification

class mma_wrapper.organizational_model.social_scheme(goals_structure, mission_to_goals, mission_to_agent_cardinality)[source]#

Bases: organizational_specification

static from_dict(d)[source]#

Return type:: social_scheme

goals_structure: plan#

mission_to_agent_cardinality: Dict[mission, cardinality]#

mission_to_goals: Dict[mission, List[goal]]#

class mma_wrapper.organizational_model.social_scheme_tag[source]#: Bases: str

class mma_wrapper.organizational_model.structural_specifications(roles, role_inheritance_relations, root_groups)[source]#

Bases: organizational_specification

static from_dict(d)[source]#

Return type:: structural_specifications

role_inheritance_relations: Dict[role, List[role]]#

roles: Dict[role, role_logic]#

root_groups: Dict[group_tag, group_specifications]#

class mma_wrapper.organizational_model.sub_plan(operator, sub_goals)[source]#

Bases: organizational_specification

operator: plan_operator#

sub_goals: List[plan]#

to_dict()[source]#

Return type:: Dict

class mma_wrapper.organizational_model.time_constraint_type(value)[source]#

Bases: str, Enum

An enumeration.

ANY = 'ANY'#

class mma_wrapper.organizational_specification_logic.goal_factory(pattern_rules=None, function_rules=None, label_manager=None, reward_reshape_standard=100)[source]#

Bases: object

create()[source]#

Return type:: goal_logic

class mma_wrapper.organizational_specification_logic.goal_logic(pattern_rules=None, function_rules=None, label_manager=None, reward_reshape_standard=100)[source]#

Bases: os_logic

Implements Goal Reward Guide (GRG) a way to guide agents to find a way to reach a goal.

grg: trajectory -> float

A GRG outputs a reward indicating how far an agent is from expected behavior according to previous trajectory characterizing a goal.

static from_dict(d)[source]#

Return type:: role_logic

next_reward(trajectory, agent_name)[source]#

Compute the next reward bonus/malus as the product of the ‘relative_reward_reshape’ (or ‘rrr’ strictly comprised between -1 and 1) and the ‘reward_reshape_standard’ (or ‘rss’ as a positive real number).

This bonus/malus is added to the “vanilla” reward once an agent has applied an action (after ‘env.step’ but before ‘return obs, rew, done, info’) to guide it into finding its own way to reach a specific subsequence characterizing a goal.

This reward bonus/malus is computed from the ‘rrr’ obtained from goal rules (TP and script rules)

The ‘rrr’ mechanism can be overviewed as such:

If there are no goal rules that is triggered, then the agent receives a null ‘rrr’, hence a 0 bonus/malus.
Else if some goal rules are triggered, the final ‘rrr’ is the sum of all those applied ones’ results

Parameters:: trajectory (List[Tuple[Union[int, ndarray], Union[int, ndarray]]]) – an agent’s trajectory (containing raw observation and action objects)
Return type:: float
Returns:: float the bonus/malus to add in reward to entice the agent’s into adopting the role’s expected behavior

register_pattern_rule(trajectory_pattern_str, relative_reward_reshape)[source]#

Register a pattern-based trajectory (TP) rule to the current TP rule register. A TP rule is triggered when an agent’s trajectory do match its ‘trajectory_pattern_str’ resulting in a float reward. :type trajectory_pattern_str: str :param trajectory_pattern_str: the TP to register into forming the first condition of the TP rule :type relative_reward_reshape: float :param relative_reward_reshape: the relative reward reshape value (supposedly between -1 and 1)

Returns:: None

register_pattern_rules(rules)[source]#

Register a list of TP rules to the current TP rule register :type rules: List[Tuple[str, float]] :param rules: The TP rules to be registered as a list of (trajectory pattern, float) couples

Return: None

registrer_script_rule(function, save_source=False)[source]#

Register a script-based rule. A script-based rule is a function that outputs a relative reward reshape value (supposedely between -1 and 1) from a raw trajectory.

Parameters:: function (Callable[[List[Tuple[Union[int, ndarray], Union[int, ndarray]]], str, label_manager], float]) – the callable function to be added in the register
Returns:: None

registrer_script_rules(functions)[source]#

Parameters:: functions (List[Callable[[List[Tuple[Union[int, ndarray], Union[int, ndarray]]], str, label_manager], float]]) – the list of script-based rules to be added in the register
Returns:: None

to_dict()[source]#

Return type:: Dict

class mma_wrapper.organizational_specification_logic.os_logic[source]#

Bases: object

static from_dict(d)[source]#

Return type:: os_logic

to_dict()[source]#

class mma_wrapper.organizational_specification_logic.role_factory(pattern_rules=None, function_rules=None, label_manager=None, reward_reshape_standard=100)[source]#

Bases: object

create()[source]#

Return type:: role_logic

class mma_wrapper.organizational_specification_logic.role_logic(pattern_rules=None, function_rules=None, label_manager=None, reward_reshape_standard=100)[source]#

Bases: os_logic

Implements both Role Action Guide (RAG) and Role Reward Guide (RRG) as a common way to restrict/guide agents to expected behavior as a role.

rag : trajectory x observation -> powerset(action x float)

A RAG outputs a list of authorized (action, weight) couples from a trajectory and an observation to restrict next possible agent’s actions.

rrg = trajectory x observation x action -> float

A RRG outputs a reward indicating how far an agent is from expected behavior according to previous trajectory and the received observation and the made action.

Note: When RRG not provided, RRG is derived from RAG by default considering a default bonus/malus reward when the chosen actions in trajectory do not comply with authorized ones

static from_dict(d)[source]#

Return type:: role_logic

next_action(trajectory, observation, agent_name)[source]#

Randomly selects the next action complying with hard constraint in the TP and script rules (i.e. the associated actions’ weights are greater or equal to 0).

This function is solely used to remplace the initial chosen action by another expected one (according to the hard constraints) during the training and evaluation steps. If there are no associated hard constraints, the next action to be chosen is ‘None’, meaning the initial chosen action is kept for the rest of the step

Parameters:

trajectory (List[Tuple[Union[int, ndarray], Union[int, ndarray]]]) – the agent’s raw trajectory
observation (Union[int, ndarray]) – the agent’s raw observation
agent_name (str) – the agent’s name

Returns:

the chosen action

Return type:

action

next_reward(trajectory, observation, action, agent_name)[source]#

This bonus/malus is added to the “vanilla” reward once an agent has applied an action (after ‘env.step’ but before ‘return obs, rew, done, info’) to guide it into adopting the role expected behavior.

This reward bonus/malus is computed from soft constraint solely (i.e constraint hardness/weight is strictly between -1 et 1):

If in [0;1[ -> the associated action should be selected with a weighted priority
If in ]-1;0] -> the associated action should not be selected with a weighted priority

The ‘rrr’ mechanism can be overviewed as such:

If there are no weighted actions that would have been expected to be made / not be made for the current observation and trajectory, then the agent is free to choose any action since the ‘rrr’ is 0.
Else if weighted actions would have been expected to be made / not be made for the current observation and trajectory, then
- If the agent has chosen an action outside of any of these unwanted/wanted actions
  
  If some of actions are wanted to be chosen (at least a weight is strictly positive), then the ‘rrr’ is the least bad penalty (strictly negative weights) if there are else -1: we want to penalize the agent for not having chosen a wanted action but not as much as unwanted actions
- If the agent has chosen action within these unwanted/wanted actions, the the ‘rrr’ is directly equals to the weight: we want to penalize/reward the agent from a spectrum spanning from -1 to 1

Parameters:

trajectory (List[Tuple[Union[int, ndarray], Union[int, ndarray]]]) – an agent’s trajectory (containing raw observation and action objects)
observation (Union[int, ndarray]) – an agent’s received raw observation
action (Union[int, ndarray]) – an agent’s chosen action
agent_name (str) – an agent’s name (to be used conveniently in script)

Return type:

float

Returns:

float the bonus/malus to add in reward to entice the agent’s into adopting the role’s expected behavior

next_weighted_actions(trajectory, observation, agent_name, merge_mode='additive')[source]#

Return type:: List[Tuple[str, float]]

register_pattern_rule(trajectory_pattern_str, observation_label, weighted_actions)[source]#

Register a pattern-based trajectory (TP) rule to the current TP rule register. A TP rule is triggered when an agent’s trajectory and received observation do match its ‘trajectory_pattern_str’ and ‘observation’ resulting in a list of weighted actions. :type trajectory_pattern_str: str :param trajectory_pattern_str: the TP to register into forming the first condition of the TP rule :type observation_label: str :param observation_label: the observation label to register into forming the second condition of the TP rule :type weighted_actions: List[Tuple[str, float]] :param weighted_actions: the weighted actions to be outputed when the two conditions are satisfied

Returns:: None

register_pattern_rules(rules)[source]#

Register a list of TP rules to the current TP rule register :type rules: List[Tuple[Tuple[str, str], List[Tuple[str, float]]]] :param rules: The TP rules to be registered as a list of ((trajectory pattern, observation label), weighted actions) couples

Return: None

registrer_script_rule(function, save_source=False)[source]#

Register a script-based rule. A script-based rule is a function that outputs weighted actions from a raw trajectory and a raw observation.

Parameters:: function (Callable[[List[Tuple[Union[int, ndarray], Union[int, ndarray]]], Union[int, ndarray], str, label_manager], List[Tuple[Union[int, ndarray], float]]]) – the callable function to be added in the register
Returns:: None

registrer_script_rules(functions)[source]#

Parameters:: functions (List[Callable[[List[Tuple[Union[int, ndarray], Union[int, ndarray]]], Union[int, ndarray], str, label_manager], List[Tuple[Union[int, ndarray], float]]]]) – the list of script-based rules to be added in the register
Returns:: None

to_dict()[source]#

Return type:: Dict

class mma_wrapper.utils.cardinality(lower_bound, upper_bound)[source]#

Bases: object

static from_dict(d)[source]#

Return type:: cardinality

lower_bound: Union[int, str]#

upper_bound: Union[int, str]#

mma_wrapper.utils.draw_networkx_edge_labels(G, pos, edge_labels=None, label_pos=0.5, font_size=10, font_color='k', font_family='sans-serif', font_weight='normal', alpha=None, bbox=None, horizontalalignment='center', verticalalignment='center', ax=None, rotate=True, clip_on=True, rad=0)[source]#

Draw edge labels.

Parameters:

G (graph) – A networkx graph
pos (dictionary) – A dictionary with nodes as keys and positions as values. Positions should be sequences of length 2.
edge_labels (dictionary (default={})) – Edge labels in a dictionary of labels keyed by edge two-tuple. Only labels for the keys in the dictionary are drawn.
label_pos (float (default=0.5)) – Position of edge label along edge (0=head, 0.5=center, 1=tail)
font_size (int (default=10)) – Font size for text labels
font_color (string (default='k' black)) – Font color string
font_weight (string (default='normal')) – Font weight
font_family (string (default='sans-serif')) – Font family
alpha (float or None (default=None)) – The text transparency
bbox (Matplotlib bbox, optional) – Specify text box properties (e.g. shape, color etc.) for edge labels. Default is {boxstyle=’round’, ec=(1.0, 1.0, 1.0), fc=(1.0, 1.0, 1.0)}.
horizontalalignment (string (default='center')) – Horizontal alignment {‘center’, ‘right’, ‘left’}
verticalalignment (string (default='center')) – Vertical alignment {‘center’, ‘top’, ‘bottom’, ‘baseline’, ‘center_baseline’}
ax (Matplotlib Axes object, optional) – Draw the graph in the specified Matplotlib axes.
rotate (bool (deafult=True)) – Rotate edge labels to lie parallel to edges
clip_on (bool (default=True)) – Turn on clipping of edge labels at axis boundaries

Returns:

dict of labels keyed by edge

Return type:

dict

Examples

>>> G = nx.dodecahedral_graph()
>>> edge_labels = nx.draw_networkx_edge_labels(G, pos=nx.spring_layout(G))

Also see the NetworkX drawing examples at https://networkx.org/documentation/latest/auto_examples/index.html