aind_behavior_curriculum package¶

Submodules¶

aind_behavior_curriculum.base module¶

Base behavior pydantic object

pydantic model aind_behavior_curriculum.base.AindBehaviorModel¶

Bases: BaseModel

Defines Pydantic configurations applied to all behavior models. BaseModel: Validate arguments on initialization. Configurations:

extra=’forbid’:
Do not allow a model to be initalized with undocumented parameters.

validate_assignment=True:
Revalidate fields against schema on any change to model instance.

validate_defaults=True:
Validate default fields on subclasses.

strict=True:
Enforce strict typing.

Show JSON schema

{
   "title": "AindBehaviorModel",
   "description": "Defines Pydantic configurations applied to all behavior models.\nBaseModel: Validate arguments on initialization.\nConfigurations:\n    - extra='forbid':\n        Do not allow a model to be initalized with undocumented parameters.\n    - validate_assignment=True:\n        Revalidate fields against schema on any change to model instance.\n    - validate_defaults=True:\n        Validate default fields on subclasses.\n    - strict=True:\n        Enforce strict typing.",
   "type": "object",
   "properties": {},
   "additionalProperties": false
}

Config:

extra: str = forbid
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

pydantic model aind_behavior_curriculum.base.AindBehaviorModelExtra¶

Bases: BaseModel

Same as AindBehaviorModel w/ extra = “allow”. Helpful for deserialization of nested subclasses.

Show JSON schema

{
   "title": "AindBehaviorModelExtra",
   "description": "Same as AindBehaviorModel w/ extra = \"allow\".\nHelpful for deserialization of nested subclasses.",
   "type": "object",
   "properties": {},
   "additionalProperties": true
}

Config:

extra: str = allow
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

aind_behavior_curriculum.curriculum module¶

Core Stage and Curriculum Primitives.

pydantic model aind_behavior_curriculum.curriculum.BehaviorGraph¶

Bases: AindBehaviorModel, Generic[NodeTypes, EdgeType]

Core directed graph data structure used in Stage and Curriculum.

Show JSON schema

{
   "title": "BehaviorGraph",
   "description": "Core directed graph data structure used in Stage and Curriculum.",
   "type": "object",
   "properties": {
      "nodes": {
         "default": {},
         "title": "Nodes",
         "type": "object"
      },
      "graph": {
         "additionalProperties": {
            "items": {
               "maxItems": 2,
               "minItems": 2,
               "prefixItems": [
                  {},
                  {
                     "type": "integer"
                  }
               ],
               "type": "array"
            },
            "type": "array"
         },
         "default": {},
         "title": "Graph",
         "type": "object"
      }
   },
   "additionalProperties": false
}

Config:

extra: str = forbid
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

Fields:

graph (Dict[int, List[Tuple[aind_behavior_curriculum.curriculum.EdgeType, int]]])
nodes (Dict[int, aind_behavior_curriculum.curriculum.NodeTypes])

field graph: Dict[int, List[Tuple[EdgeType, int]]] = {}¶

field nodes: Dict[int, NodeTypes] = {}¶

add_node(node: NodeTypes) → None¶: Adds a floating node to the behavior graph.

add_transition(start_node: NodeTypes, dest_node: NodeTypes, rule: EdgeType) → None¶

Add transition between two nodes: start_node -> dest_node.

If start_node has been added to graph before, this method starts a transition: from the exisiting start_node.
If dest_node has been added to graph before, this method creates a transition: into the existing dest_node.

NOTE: The order in which this method is called sets the order of transition priority.

remove_node(node: NodeTypes) → None¶: Removes node and all associated incoming/outgoing transition rules from the stage graph. NOTE: Removed nodes and transitions have the side effect of changing transition priority.

remove_node_transition(start_node: NodeTypes, dest_node: NodeTypes, rule: EdgeType, remove_start_node: bool = False, remove_dest_node: bool = False) → None¶: Removes transition with options to remove start/end nodes associated with the transition. NOTE: Removed nodes and transitions has the side effect of changing transition priority.

see_node_transitions(node: NodeTypes) → List[Tuple[EdgeType, NodeTypes]]¶: See transitions of node in behavior graph.

see_nodes() → List[NodeTypes]¶: See nodes of behavior graph.

set_transition_priority(node: NodeTypes, node_transitions: List[Tuple[EdgeType, NodeTypes]]) → None¶: Change order of node transitions listed under a node. Highest priority is ordered in node_transition from left -> right.

pydantic model aind_behavior_curriculum.curriculum.BehaviorGraph¶

Bases: AindBehaviorModel, Generic[NodeTypes, EdgeType]

Core directed graph data structure used in Stage and Curriculum.

Show JSON schema

{
   "title": "BehaviorGraph",
   "description": "Core directed graph data structure used in Stage and Curriculum.",
   "type": "object",
   "properties": {
      "nodes": {
         "default": {},
         "title": "Nodes",
         "type": "object"
      },
      "graph": {
         "additionalProperties": {
            "items": {
               "maxItems": 2,
               "minItems": 2,
               "prefixItems": [
                  {},
                  {
                     "type": "integer"
                  }
               ],
               "type": "array"
            },
            "type": "array"
         },
         "default": {},
         "title": "Graph",
         "type": "object"
      }
   },
   "additionalProperties": false
}

Config:

extra: str = forbid
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

Fields:

graph (Dict[int, List[Tuple[EdgeType, int]]])
nodes (Dict[int, NodeTypes])

field graph: Dict[int, List[Tuple[EdgeType, int]]] = {}¶

field nodes: Dict[int, NodeTypes] = {}¶

add_node(node: NodeTypes) → None¶: Adds a floating node to the behavior graph.

add_transition(start_node: NodeTypes, dest_node: NodeTypes, rule: EdgeType) → None¶

Add transition between two nodes: start_node -> dest_node.

If start_node has been added to graph before, this method starts a transition: from the exisiting start_node.
If dest_node has been added to graph before, this method creates a transition: into the existing dest_node.

NOTE: The order in which this method is called sets the order of transition priority.

remove_node(node: NodeTypes) → None¶: Removes node and all associated incoming/outgoing transition rules from the stage graph. NOTE: Removed nodes and transitions have the side effect of changing transition priority.

remove_node_transition(start_node: NodeTypes, dest_node: NodeTypes, rule: EdgeType, remove_start_node: bool = False, remove_dest_node: bool = False) → None¶: Removes transition with options to remove start/end nodes associated with the transition. NOTE: Removed nodes and transitions has the side effect of changing transition priority.

see_node_transitions(node: NodeTypes) → List[Tuple[EdgeType, NodeTypes]]¶: See transitions of node in behavior graph.

see_nodes() → List[NodeTypes]¶: See nodes of behavior graph.

set_transition_priority(node: NodeTypes, node_transitions: List[Tuple[EdgeType, NodeTypes]]) → None¶: Change order of node transitions listed under a node. Highest priority is ordered in node_transition from left -> right.

pydantic model aind_behavior_curriculum.curriculum.BehaviorGraph¶

Bases: AindBehaviorModel, Generic[NodeTypes, EdgeType]

Core directed graph data structure used in Stage and Curriculum.

Show JSON schema

{
   "title": "BehaviorGraph",
   "description": "Core directed graph data structure used in Stage and Curriculum.",
   "type": "object",
   "properties": {
      "nodes": {
         "default": {},
         "title": "Nodes",
         "type": "object"
      },
      "graph": {
         "additionalProperties": {
            "items": {
               "maxItems": 2,
               "minItems": 2,
               "prefixItems": [
                  {},
                  {
                     "type": "integer"
                  }
               ],
               "type": "array"
            },
            "type": "array"
         },
         "default": {},
         "title": "Graph",
         "type": "object"
      }
   },
   "additionalProperties": false
}

Config:

extra: str = forbid
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

Fields:

graph (Dict[int, List[Tuple[EdgeType, int]]])
nodes (Dict[int, NodeTypes])

field graph: Dict[int, List[Tuple[EdgeType, int]]] = {}¶

field nodes: Dict[int, NodeTypes] = {}¶

add_node(node: NodeTypes) → None¶: Adds a floating node to the behavior graph.

add_transition(start_node: NodeTypes, dest_node: NodeTypes, rule: EdgeType) → None¶

Add transition between two nodes: start_node -> dest_node.

If start_node has been added to graph before, this method starts a transition: from the exisiting start_node.
If dest_node has been added to graph before, this method creates a transition: into the existing dest_node.

NOTE: The order in which this method is called sets the order of transition priority.

remove_node(node: NodeTypes) → None¶: Removes node and all associated incoming/outgoing transition rules from the stage graph. NOTE: Removed nodes and transitions have the side effect of changing transition priority.

remove_node_transition(start_node: NodeTypes, dest_node: NodeTypes, rule: EdgeType, remove_start_node: bool = False, remove_dest_node: bool = False) → None¶: Removes transition with options to remove start/end nodes associated with the transition. NOTE: Removed nodes and transitions has the side effect of changing transition priority.

see_node_transitions(node: NodeTypes) → List[Tuple[EdgeType, NodeTypes]]¶: See transitions of node in behavior graph.

see_nodes() → List[NodeTypes]¶: See nodes of behavior graph.

set_transition_priority(node: NodeTypes, node_transitions: List[Tuple[EdgeType, NodeTypes]]) → None¶: Change order of node transitions listed under a node. Highest priority is ordered in node_transition from left -> right.

pydantic model aind_behavior_curriculum.curriculum.Curriculum¶

Bases: AindBehaviorModel

Curriculum manages a StageGraph instance with a read/write API. To use, subclass this and add subclass metrics.

Show JSON schema

{
   "title": "Curriculum",
   "description": "Curriculum manages a StageGraph instance with a read/write API.\nTo use, subclass this and add subclass metrics.",
   "type": "object",
   "properties": {
      "pkg_location": {
         "default": "",
         "title": "Pkg Location",
         "type": "string"
      },
      "name": {
         "default": "Please subclass, rename, and define                  a StageGraph with your own Stage objs                  Ex: StageGraph[Union[StageA, StageB, Graduated]]",
         "title": "Name",
         "type": "string"
      },
      "graph": {
         "allOf": [
            {
               "$ref": "#/$defs/StageGraph"
            }
         ],
         "default": {
            "nodes": {},
            "graph": {}
         }
      }
   },
   "$defs": {
      "Policy": {
         "additionalProperties": false,
         "description": "User-defined function that defines\nhow current Task parameters change according to metrics.",
         "properties": {
            "rule": {
               "description": "Callable with Serialization.",
               "title": "Rule",
               "type": "string"
            }
         },
         "required": [
            "rule"
         ],
         "title": "Policy",
         "type": "object"
      },
      "PolicyGraph": {
         "additionalProperties": false,
         "description": "Graph for Stage.",
         "properties": {
            "nodes": {
               "additionalProperties": {
                  "$ref": "#/$defs/Policy"
               },
               "default": {},
               "title": "Nodes",
               "type": "object"
            },
            "graph": {
               "additionalProperties": {
                  "items": {
                     "maxItems": 2,
                     "minItems": 2,
                     "prefixItems": [
                        {
                           "$ref": "#/$defs/PolicyTransition"
                        },
                        {
                           "type": "integer"
                        }
                     ],
                     "type": "array"
                  },
                  "type": "array"
               },
               "default": {},
               "title": "Graph",
               "type": "object"
            }
         },
         "title": "PolicyGraph",
         "type": "object"
      },
      "PolicyTransition": {
         "additionalProperties": false,
         "description": "User-defined function that defines\ncriteria for transitioning between policies based on metrics.",
         "properties": {
            "rule": {
               "description": "Callable with Serialization.",
               "title": "Rule",
               "type": "string"
            }
         },
         "required": [
            "rule"
         ],
         "title": "PolicyTransition",
         "type": "object"
      },
      "Stage": {
         "additionalProperties": false,
         "description": "Instance of a Task.\nTask Parameters may change according to rules defined in BehaviorGraph.\nStage manages a BehaviorGraph instance with a read/write API.",
         "properties": {
            "name": {
               "description": "Stage name.",
               "title": "Name",
               "type": "string"
            },
            "task": {
               "allOf": [
                  {
                     "$ref": "#/$defs/Task"
                  }
               ],
               "description": "Task in which this stage is based off of."
            },
            "graph": {
               "allOf": [
                  {
                     "$ref": "#/$defs/PolicyGraph"
                  }
               ],
               "default": {
                  "nodes": {},
                  "graph": {}
               }
            },
            "start_policies": {
               "default": [],
               "items": {
                  "$ref": "#/$defs/Policy"
               },
               "title": "Start Policies",
               "type": "array"
            }
         },
         "required": [
            "name",
            "task"
         ],
         "title": "Stage",
         "type": "object"
      },
      "StageGraph": {
         "additionalProperties": false,
         "description": "Graph for Curriculum.",
         "properties": {
            "nodes": {
               "additionalProperties": {
                  "$ref": "#/$defs/Stage"
               },
               "default": {},
               "title": "Nodes",
               "type": "object"
            },
            "graph": {
               "additionalProperties": {
                  "items": {
                     "maxItems": 2,
                     "minItems": 2,
                     "prefixItems": [
                        {
                           "$ref": "#/$defs/StageTransition"
                        },
                        {
                           "type": "integer"
                        }
                     ],
                     "type": "array"
                  },
                  "type": "array"
               },
               "default": {},
               "title": "Graph",
               "type": "object"
            }
         },
         "title": "StageGraph",
         "type": "object"
      },
      "StageTransition": {
         "additionalProperties": false,
         "description": "User-defined function that defines\ncriteria for transitioning stages based on metrics.",
         "properties": {
            "rule": {
               "description": "Callable with Serialization.",
               "title": "Rule",
               "type": "string"
            }
         },
         "required": [
            "rule"
         ],
         "title": "StageTransition",
         "type": "object"
      },
      "Task": {
         "additionalProperties": false,
         "description": "Base Task Primitive.\nHolds Task metadata and parameters.",
         "properties": {
            "name": {
               "description": "Name of the task.",
               "title": "Name",
               "type": "string"
            },
            "description": {
               "default": "",
               "description": "Description of the task.",
               "title": "Description",
               "type": "string"
            },
            "task_parameters": {
               "allOf": [
                  {
                     "$ref": "#/$defs/TaskParameters"
                  }
               ],
               "description": "Set of parameters associated with a subject task.\n    Subclass with Task Parameters."
            }
         },
         "required": [
            "name",
            "task_parameters"
         ],
         "title": "Task",
         "type": "object"
      },
      "TaskParameters": {
         "additionalProperties": true,
         "description": "Set of parameters associated with a subject task.\nSubclass with Task Parameters.",
         "properties": {},
         "title": "TaskParameters",
         "type": "object"
      }
   },
   "additionalProperties": false
}

Config:

extra: str = forbid
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

Fields:

graph (aind_behavior_curriculum.curriculum.StageGraph)
name (str)
pkg_location (str)

field graph: StageGraph = StageGraph(nodes={}, graph={})¶

field name: str = 'Please subclass, rename, and define a StageGraph with your own Stage objs Ex: StageGraph[Union[StageA, StageB, Graduated]]'¶

field pkg_location: str = ''¶

add_stage(stage: Stage) → None¶: Adds a floating stage to the Curriculum adjacency graph.

add_stage_transition(start_stage: Stage, dest_stage: Stage, rule: StageTransition) → None¶

Add stage transition between two stages: Stage_A -> Stage_B.

If Stage_A has been added to stage before, this method starts a transition: from the exisiting Stage_A.
If Stage_B has been added to stage before, this method creates a transition: into the existing Stage_B.

NOTE: The order in which this method is called sets the order of transition priority.

classmethod download_curriculum(name: str, version: str, bucket='aind-behavior-curriculum-prod-o5171v') → Curriculum¶: Reconstruct curriculum object from cloud json.

export_curriculum(export_dir: str) → None¶: Export json and diagram into export dir.

export_diagram(png_path: str) → None¶: Makes diagram for input Curriculum and writes to output png_path.

export_json(json_path: str) → None¶: Export curriculum json to export path

model_post_init(_Curriculum__context: Any) → None¶: Add Curriculum pkg location

remove_stage(stage: Stage) → None¶: Removes stage and all associated incoming/outgoing transition rules from the curriculum graph. NOTE: Removed nodes and transitions have the side effect of changing transition priority.

remove_stage_transition(start_stage: Policy, dest_stage: Policy, rule: PolicyTransition, remove_start_stage: bool = False, remove_dest_stage: bool = False) → None¶: Removes transition with options to remove start/end stages associated with the transition. NOTE: Removed nodes and transitions has the side effect of changing transition priority.

see_stage_transitions(stage: Stage) → List[Tuple[StageTransition, Stage]]¶: See transitions of stage in curriculum graph.

see_stages() → List[Stage]¶: See stages of curriculum graph.

set_stage_transition_priority(stage: Stage, stage_transitions: List[Tuple[StageTransition, Stage]]) → None¶: Change the order of stage transitions listed under a stage. To use, call see_stage_transitions() and order the transitions in the desired priority from left -> right.

validate_curriculum() → Curriculum¶: Validate curriculum for export/serialization.

pydantic model aind_behavior_curriculum.curriculum.Metrics¶

Bases: AindBehaviorModelExtra

Abstract Metrics class. Subclass with Metric values.

Show JSON schema

{
   "title": "Metrics",
   "description": "Abstract Metrics class.\nSubclass with Metric values.",
   "type": "object",
   "properties": {},
   "additionalProperties": true
}

Config:

extra: str = allow
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

pydantic model aind_behavior_curriculum.curriculum.Policy¶

Bases: AindBehaviorModel

User-defined function that defines how current Task parameters change according to metrics.

Show JSON schema

{
   "title": "Policy",
   "description": "User-defined function that defines\nhow current Task parameters change according to metrics.",
   "type": "object",
   "properties": {
      "rule": {
         "description": "Callable with Serialization.",
         "title": "Rule",
         "type": "string"
      }
   },
   "additionalProperties": false,
   "required": [
      "rule"
   ]
}

Config:

extra: str = forbid
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

Fields:

rule (aind_behavior_curriculum.curriculum.Rule)

Validators:

validate_rule » rule

field rule: Rule [Required]¶

Callable with Serialization.

Validated by:

validate_rule

validator validate_rule » rule¶

Policy Signature: I: - metrics: Metrics object - task_parameters: TaskParameters object

O: - result: TaskParameters object

pydantic model aind_behavior_curriculum.curriculum.PolicyGraph¶

Bases: BehaviorGraph[Policy, PolicyTransition]

Graph for Stage.

Show JSON schema

{
   "title": "PolicyGraph",
   "description": "Graph for Stage.",
   "type": "object",
   "properties": {
      "nodes": {
         "additionalProperties": {
            "$ref": "#/$defs/Policy"
         },
         "default": {},
         "title": "Nodes",
         "type": "object"
      },
      "graph": {
         "additionalProperties": {
            "items": {
               "maxItems": 2,
               "minItems": 2,
               "prefixItems": [
                  {
                     "$ref": "#/$defs/PolicyTransition"
                  },
                  {
                     "type": "integer"
                  }
               ],
               "type": "array"
            },
            "type": "array"
         },
         "default": {},
         "title": "Graph",
         "type": "object"
      }
   },
   "$defs": {
      "Policy": {
         "additionalProperties": false,
         "description": "User-defined function that defines\nhow current Task parameters change according to metrics.",
         "properties": {
            "rule": {
               "description": "Callable with Serialization.",
               "title": "Rule",
               "type": "string"
            }
         },
         "required": [
            "rule"
         ],
         "title": "Policy",
         "type": "object"
      },
      "PolicyTransition": {
         "additionalProperties": false,
         "description": "User-defined function that defines\ncriteria for transitioning between policies based on metrics.",
         "properties": {
            "rule": {
               "description": "Callable with Serialization.",
               "title": "Rule",
               "type": "string"
            }
         },
         "required": [
            "rule"
         ],
         "title": "PolicyTransition",
         "type": "object"
      }
   },
   "additionalProperties": false
}

Config:

extra: str = forbid
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

Fields:

pydantic model aind_behavior_curriculum.curriculum.PolicyTransition¶

Bases: AindBehaviorModel

User-defined function that defines criteria for transitioning between policies based on metrics.

Show JSON schema

{
   "title": "PolicyTransition",
   "description": "User-defined function that defines\ncriteria for transitioning between policies based on metrics.",
   "type": "object",
   "properties": {
      "rule": {
         "description": "Callable with Serialization.",
         "title": "Rule",
         "type": "string"
      }
   },
   "additionalProperties": false,
   "required": [
      "rule"
   ]
}

Config:

extra: str = forbid
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

Fields:

rule (aind_behavior_curriculum.curriculum.Rule)

Validators:

validate_rule » rule

field rule: Rule [Required]¶

Callable with Serialization.

Validated by:

validate_rule

validator validate_rule » rule¶

Policy Transition Signature: I: - metrics: Metrics object

O: - result: bool

class aind_behavior_curriculum.curriculum.Rule¶

Bases: object

Custom Pydantic Type that defines de/serialiation for Callables.

pydantic model aind_behavior_curriculum.curriculum.Stage¶

Bases: AindBehaviorModel, Generic[TTask]

Instance of a Task. Task Parameters may change according to rules defined in BehaviorGraph. Stage manages a BehaviorGraph instance with a read/write API.

Show JSON schema

{
   "title": "Stage",
   "description": "Instance of a Task.\nTask Parameters may change according to rules defined in BehaviorGraph.\nStage manages a BehaviorGraph instance with a read/write API.",
   "type": "object",
   "properties": {
      "name": {
         "description": "Stage name.",
         "title": "Name",
         "type": "string"
      },
      "task": {
         "allOf": [
            {
               "$ref": "#/$defs/Task"
            }
         ],
         "description": "Task in which this stage is based off of."
      },
      "graph": {
         "allOf": [
            {
               "$ref": "#/$defs/PolicyGraph"
            }
         ],
         "default": {
            "nodes": {},
            "graph": {}
         }
      },
      "start_policies": {
         "default": [],
         "items": {
            "$ref": "#/$defs/Policy"
         },
         "title": "Start Policies",
         "type": "array"
      }
   },
   "$defs": {
      "Policy": {
         "additionalProperties": false,
         "description": "User-defined function that defines\nhow current Task parameters change according to metrics.",
         "properties": {
            "rule": {
               "description": "Callable with Serialization.",
               "title": "Rule",
               "type": "string"
            }
         },
         "required": [
            "rule"
         ],
         "title": "Policy",
         "type": "object"
      },
      "PolicyGraph": {
         "additionalProperties": false,
         "description": "Graph for Stage.",
         "properties": {
            "nodes": {
               "additionalProperties": {
                  "$ref": "#/$defs/Policy"
               },
               "default": {},
               "title": "Nodes",
               "type": "object"
            },
            "graph": {
               "additionalProperties": {
                  "items": {
                     "maxItems": 2,
                     "minItems": 2,
                     "prefixItems": [
                        {
                           "$ref": "#/$defs/PolicyTransition"
                        },
                        {
                           "type": "integer"
                        }
                     ],
                     "type": "array"
                  },
                  "type": "array"
               },
               "default": {},
               "title": "Graph",
               "type": "object"
            }
         },
         "title": "PolicyGraph",
         "type": "object"
      },
      "PolicyTransition": {
         "additionalProperties": false,
         "description": "User-defined function that defines\ncriteria for transitioning between policies based on metrics.",
         "properties": {
            "rule": {
               "description": "Callable with Serialization.",
               "title": "Rule",
               "type": "string"
            }
         },
         "required": [
            "rule"
         ],
         "title": "PolicyTransition",
         "type": "object"
      },
      "Task": {
         "additionalProperties": false,
         "description": "Base Task Primitive.\nHolds Task metadata and parameters.",
         "properties": {
            "name": {
               "description": "Name of the task.",
               "title": "Name",
               "type": "string"
            },
            "description": {
               "default": "",
               "description": "Description of the task.",
               "title": "Description",
               "type": "string"
            },
            "task_parameters": {
               "allOf": [
                  {
                     "$ref": "#/$defs/TaskParameters"
                  }
               ],
               "description": "Set of parameters associated with a subject task.\n    Subclass with Task Parameters."
            }
         },
         "required": [
            "name",
            "task_parameters"
         ],
         "title": "Task",
         "type": "object"
      },
      "TaskParameters": {
         "additionalProperties": true,
         "description": "Set of parameters associated with a subject task.\nSubclass with Task Parameters.",
         "properties": {},
         "title": "TaskParameters",
         "type": "object"
      }
   },
   "additionalProperties": false,
   "required": [
      "name",
      "task"
   ]
}

Config:

extra: str = forbid
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

Fields:

graph (aind_behavior_curriculum.curriculum.PolicyGraph)
name (str)
start_policies (List[aind_behavior_curriculum.curriculum.Policy])
task (aind_behavior_curriculum.curriculum.TTask)

field graph: PolicyGraph = PolicyGraph(nodes={}, graph={})¶

field name: str [Required]¶: Stage name.

field start_policies: List[Policy] = []¶

field task: TTask [Required]¶: Task in which this stage is based off of.

add_policy(policy: Policy) → None¶: Adds a floating policy to the Stage adjacency graph.

add_policy_transition(start_policy: Policy, dest_policy: Policy, rule: PolicyTransition) → None¶

Add policy transition between two policies: Policy_A -> Policy_B.

If Policy_A has been added to stage before, this method starts a transition: from the existing Policy_A.
If Policy_B has been added to stage before, this method creates a transition: into the existing Policy_B.

NOTE: The order in which this method is called sets the order of transition priority.

get_task_parameters() → TaskParameters¶: See current task parameters of Task.

remove_policy(policy: Policy) → None¶: Removes policy and all associated incoming/outgoing transition rules from the stage graph. NOTE: Removed nodes and transitions have the side effect of changing transition priority.

remove_policy_transition(start_policy: Policy, dest_policy: Policy, rule: PolicyTransition, remove_start_policy: bool = False, remove_dest_policy: bool = False) → None¶: Removes transition with options to remove start/end policies associated with the transition. NOTE: Removed nodes and transitions has the side effect of changing transition priority.

see_policies() → List[Policy]¶: See policies of policy graph.

see_policy_transitions(policy: Policy) → List[Tuple[PolicyTransition, Policy]]¶: TTask See transitions of stage in policy graph.

set_policy_transition_priority(policy: Policy, policy_transitions: List[Tuple[PolicyTransition, Policy]]) → None¶: Change the order of policy transitions listed under a policy. To use, call see_policy_transitions() and order the transitions in the desired priority from left -> right.

set_start_policies(start_policies: Policy | List[Policy])¶: Sets stage’s start policies to start policies provided. Input overwrites existing start policies.

set_task_parameters(task_params: TaskParameters) → None¶: Set task with new set of task parameters. Task revalidates TaskParameters on assignment.

validate_stage() → Stage¶: Check if stage is non-empty and specifies start policies.

pydantic model aind_behavior_curriculum.curriculum.StageGraph¶

Bases: BehaviorGraph[Stage, StageTransition], Generic[TTask]

Graph for Curriculum.

Show JSON schema

{
   "title": "StageGraph",
   "description": "Graph for Curriculum.",
   "type": "object",
   "properties": {
      "nodes": {
         "additionalProperties": {
            "$ref": "#/$defs/Stage"
         },
         "default": {},
         "title": "Nodes",
         "type": "object"
      },
      "graph": {
         "additionalProperties": {
            "items": {
               "maxItems": 2,
               "minItems": 2,
               "prefixItems": [
                  {
                     "$ref": "#/$defs/StageTransition"
                  },
                  {
                     "type": "integer"
                  }
               ],
               "type": "array"
            },
            "type": "array"
         },
         "default": {},
         "title": "Graph",
         "type": "object"
      }
   },
   "$defs": {
      "Policy": {
         "additionalProperties": false,
         "description": "User-defined function that defines\nhow current Task parameters change according to metrics.",
         "properties": {
            "rule": {
               "description": "Callable with Serialization.",
               "title": "Rule",
               "type": "string"
            }
         },
         "required": [
            "rule"
         ],
         "title": "Policy",
         "type": "object"
      },
      "PolicyGraph": {
         "additionalProperties": false,
         "description": "Graph for Stage.",
         "properties": {
            "nodes": {
               "additionalProperties": {
                  "$ref": "#/$defs/Policy"
               },
               "default": {},
               "title": "Nodes",
               "type": "object"
            },
            "graph": {
               "additionalProperties": {
                  "items": {
                     "maxItems": 2,
                     "minItems": 2,
                     "prefixItems": [
                        {
                           "$ref": "#/$defs/PolicyTransition"
                        },
                        {
                           "type": "integer"
                        }
                     ],
                     "type": "array"
                  },
                  "type": "array"
               },
               "default": {},
               "title": "Graph",
               "type": "object"
            }
         },
         "title": "PolicyGraph",
         "type": "object"
      },
      "PolicyTransition": {
         "additionalProperties": false,
         "description": "User-defined function that defines\ncriteria for transitioning between policies based on metrics.",
         "properties": {
            "rule": {
               "description": "Callable with Serialization.",
               "title": "Rule",
               "type": "string"
            }
         },
         "required": [
            "rule"
         ],
         "title": "PolicyTransition",
         "type": "object"
      },
      "Stage": {
         "additionalProperties": false,
         "description": "Instance of a Task.\nTask Parameters may change according to rules defined in BehaviorGraph.\nStage manages a BehaviorGraph instance with a read/write API.",
         "properties": {
            "name": {
               "description": "Stage name.",
               "title": "Name",
               "type": "string"
            },
            "task": {
               "allOf": [
                  {
                     "$ref": "#/$defs/Task"
                  }
               ],
               "description": "Task in which this stage is based off of."
            },
            "graph": {
               "allOf": [
                  {
                     "$ref": "#/$defs/PolicyGraph"
                  }
               ],
               "default": {
                  "nodes": {},
                  "graph": {}
               }
            },
            "start_policies": {
               "default": [],
               "items": {
                  "$ref": "#/$defs/Policy"
               },
               "title": "Start Policies",
               "type": "array"
            }
         },
         "required": [
            "name",
            "task"
         ],
         "title": "Stage",
         "type": "object"
      },
      "StageTransition": {
         "additionalProperties": false,
         "description": "User-defined function that defines\ncriteria for transitioning stages based on metrics.",
         "properties": {
            "rule": {
               "description": "Callable with Serialization.",
               "title": "Rule",
               "type": "string"
            }
         },
         "required": [
            "rule"
         ],
         "title": "StageTransition",
         "type": "object"
      },
      "Task": {
         "additionalProperties": false,
         "description": "Base Task Primitive.\nHolds Task metadata and parameters.",
         "properties": {
            "name": {
               "description": "Name of the task.",
               "title": "Name",
               "type": "string"
            },
            "description": {
               "default": "",
               "description": "Description of the task.",
               "title": "Description",
               "type": "string"
            },
            "task_parameters": {
               "allOf": [
                  {
                     "$ref": "#/$defs/TaskParameters"
                  }
               ],
               "description": "Set of parameters associated with a subject task.\n    Subclass with Task Parameters."
            }
         },
         "required": [
            "name",
            "task_parameters"
         ],
         "title": "Task",
         "type": "object"
      },
      "TaskParameters": {
         "additionalProperties": true,
         "description": "Set of parameters associated with a subject task.\nSubclass with Task Parameters.",
         "properties": {},
         "title": "TaskParameters",
         "type": "object"
      }
   },
   "additionalProperties": false
}

Config:

extra: str = forbid
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

Fields:

pydantic model aind_behavior_curriculum.curriculum.StageTransition¶

Bases: AindBehaviorModel

User-defined function that defines criteria for transitioning stages based on metrics.

Show JSON schema

{
   "title": "StageTransition",
   "description": "User-defined function that defines\ncriteria for transitioning stages based on metrics.",
   "type": "object",
   "properties": {
      "rule": {
         "description": "Callable with Serialization.",
         "title": "Rule",
         "type": "string"
      }
   },
   "additionalProperties": false,
   "required": [
      "rule"
   ]
}

Config:

extra: str = forbid
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

Fields:

rule (aind_behavior_curriculum.curriculum.Rule)

Validators:

validate_rule » rule

field rule: Rule [Required]¶

Callable with Serialization.

Validated by:

validate_rule

validator validate_rule » rule¶

Stage Transition Signature: I: - metrics: Metrics object

O: - result: bool

aind_behavior_curriculum.curriculum_utils module¶

Useful Placeholders when making Curriculums

pydantic model aind_behavior_curriculum.curriculum_utils.Graduated¶

Bases: Task

Utility Final Task.

Show JSON schema

{
   "title": "Graduated",
   "description": "Utility Final Task.",
   "type": "object",
   "properties": {
      "name": {
         "const": "Graduated",
         "default": "Graduated",
         "enum": [
            "Graduated"
         ],
         "title": "Name",
         "type": "string"
      },
      "description": {
         "default": "",
         "description": "Description of the task.",
         "title": "Description",
         "type": "string"
      },
      "task_parameters": {
         "allOf": [
            {
               "$ref": "#/$defs/TaskParameters"
            }
         ],
         "default": {},
         "description": "Fill w/ Parameter Defaults"
      }
   },
   "$defs": {
      "TaskParameters": {
         "additionalProperties": true,
         "description": "Set of parameters associated with a subject task.\nSubclass with Task Parameters.",
         "properties": {},
         "title": "TaskParameters",
         "type": "object"
      }
   },
   "additionalProperties": false
}

Config:

extra: str = forbid
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

Fields:

name (Literal['Graduated'])
task_parameters (aind_behavior_curriculum.task.TaskParameters)

field name: Literal['Graduated'] = 'Graduated'¶

field task_parameters: TaskParameters = TaskParameters()¶: Fill w/ Parameter Defaults

aind_behavior_curriculum.curriculum_utils.create_empty_stage(s: Stage) → Stage¶: Prepares empty stage with tacit policy initalization. Convenient for initalizing many empty stages.

aind_behavior_curriculum.curriculum_utils.get_task_types()¶

Used for Curriculum StageGraph declaration. Ex: Tasks = get_task_types() class MyCurriculum(Curriculum):

name: Literal[“My Curriculum”] = “My Curriculum” graph: StageGraph[Tasks] = Field(default=StageGraph())

Explanation: Invokes Task.__subclasses__() in the module in which all Tasks have been defined.

aind_behavior_curriculum.curriculum_utils.init_stage_rule(metrics: Metrics, task_params: TaskParameters) → TaskParameters¶: Trivially pass the default

aind_behavior_curriculum.task module¶

Base Behavior Models

pydantic model aind_behavior_curriculum.task.Task¶

Bases: AindBehaviorModel

Base Task Primitive. Holds Task metadata and parameters.

Show JSON schema

{
   "title": "Task",
   "description": "Base Task Primitive.\nHolds Task metadata and parameters.",
   "type": "object",
   "properties": {
      "name": {
         "description": "Name of the task.",
         "title": "Name",
         "type": "string"
      },
      "description": {
         "default": "",
         "description": "Description of the task.",
         "title": "Description",
         "type": "string"
      },
      "task_parameters": {
         "allOf": [
            {
               "$ref": "#/$defs/TaskParameters"
            }
         ],
         "description": "Set of parameters associated with a subject task.\n    Subclass with Task Parameters."
      }
   },
   "$defs": {
      "TaskParameters": {
         "additionalProperties": true,
         "description": "Set of parameters associated with a subject task.\nSubclass with Task Parameters.",
         "properties": {},
         "title": "TaskParameters",
         "type": "object"
      }
   },
   "additionalProperties": false,
   "required": [
      "name",
      "task_parameters"
   ]
}

Config:

extra: str = forbid
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

Fields:

description (str)
name (str)
task_parameters (aind_behavior_curriculum.task.TaskParameters)

field description: str = ''¶: Description of the task.

field name: str [Required]¶: Name of the task.

field task_parameters: TaskParameters [Required]¶: Set of parameters associated with a subject task. Subclass with Task Parameters.

pydantic model aind_behavior_curriculum.task.TaskParameters¶

Bases: AindBehaviorModelExtra

Set of parameters associated with a subject task. Subclass with Task Parameters.

Show JSON schema

{
   "title": "TaskParameters",
   "description": "Set of parameters associated with a subject task.\nSubclass with Task Parameters.",
   "type": "object",
   "properties": {},
   "additionalProperties": true
}

Config:

extra: str = allow
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

aind_behavior_curriculum.trainer module¶

Core Trainer primitive.

pydantic model aind_behavior_curriculum.trainer.SubjectHistory¶

Bases: AindBehaviorModel

Record of subject locations in Curriculum. Histories can hold a ‘None’ object indicating user override off-curriculum. Pydantic model for de/serialization.

Show JSON schema

{
   "title": "SubjectHistory",
   "description": "Record of subject locations in Curriculum.\nHistories can hold a 'None' object indicating user\noverride off-curriculum.\nPydantic model for de/serialization.",
   "type": "object",
   "properties": {
      "stage_history": {
         "default": [],
         "items": {
            "anyOf": [
               {
                  "$ref": "#/$defs/Stage"
               },
               {
                  "type": "null"
               }
            ]
         },
         "title": "Stage History",
         "type": "array"
      },
      "policy_history": {
         "default": [],
         "items": {
            "anyOf": [
               {
                  "items": {
                     "$ref": "#/$defs/Policy"
                  },
                  "type": "array"
               },
               {
                  "type": "null"
               }
            ]
         },
         "title": "Policy History",
         "type": "array"
      }
   },
   "$defs": {
      "Policy": {
         "additionalProperties": false,
         "description": "User-defined function that defines\nhow current Task parameters change according to metrics.",
         "properties": {
            "rule": {
               "description": "Callable with Serialization.",
               "title": "Rule",
               "type": "string"
            }
         },
         "required": [
            "rule"
         ],
         "title": "Policy",
         "type": "object"
      },
      "PolicyGraph": {
         "additionalProperties": false,
         "description": "Graph for Stage.",
         "properties": {
            "nodes": {
               "additionalProperties": {
                  "$ref": "#/$defs/Policy"
               },
               "default": {},
               "title": "Nodes",
               "type": "object"
            },
            "graph": {
               "additionalProperties": {
                  "items": {
                     "maxItems": 2,
                     "minItems": 2,
                     "prefixItems": [
                        {
                           "$ref": "#/$defs/PolicyTransition"
                        },
                        {
                           "type": "integer"
                        }
                     ],
                     "type": "array"
                  },
                  "type": "array"
               },
               "default": {},
               "title": "Graph",
               "type": "object"
            }
         },
         "title": "PolicyGraph",
         "type": "object"
      },
      "PolicyTransition": {
         "additionalProperties": false,
         "description": "User-defined function that defines\ncriteria for transitioning between policies based on metrics.",
         "properties": {
            "rule": {
               "description": "Callable with Serialization.",
               "title": "Rule",
               "type": "string"
            }
         },
         "required": [
            "rule"
         ],
         "title": "PolicyTransition",
         "type": "object"
      },
      "Stage": {
         "additionalProperties": false,
         "description": "Instance of a Task.\nTask Parameters may change according to rules defined in BehaviorGraph.\nStage manages a BehaviorGraph instance with a read/write API.",
         "properties": {
            "name": {
               "description": "Stage name.",
               "title": "Name",
               "type": "string"
            },
            "task": {
               "allOf": [
                  {
                     "$ref": "#/$defs/Task"
                  }
               ],
               "description": "Task in which this stage is based off of."
            },
            "graph": {
               "allOf": [
                  {
                     "$ref": "#/$defs/PolicyGraph"
                  }
               ],
               "default": {
                  "nodes": {},
                  "graph": {}
               }
            },
            "start_policies": {
               "default": [],
               "items": {
                  "$ref": "#/$defs/Policy"
               },
               "title": "Start Policies",
               "type": "array"
            }
         },
         "required": [
            "name",
            "task"
         ],
         "title": "Stage",
         "type": "object"
      },
      "Task": {
         "additionalProperties": false,
         "description": "Base Task Primitive.\nHolds Task metadata and parameters.",
         "properties": {
            "name": {
               "description": "Name of the task.",
               "title": "Name",
               "type": "string"
            },
            "description": {
               "default": "",
               "description": "Description of the task.",
               "title": "Description",
               "type": "string"
            },
            "task_parameters": {
               "allOf": [
                  {
                     "$ref": "#/$defs/TaskParameters"
                  }
               ],
               "description": "Set of parameters associated with a subject task.\n    Subclass with Task Parameters."
            }
         },
         "required": [
            "name",
            "task_parameters"
         ],
         "title": "Task",
         "type": "object"
      },
      "TaskParameters": {
         "additionalProperties": true,
         "description": "Set of parameters associated with a subject task.\nSubclass with Task Parameters.",
         "properties": {},
         "title": "TaskParameters",
         "type": "object"
      }
   },
   "additionalProperties": false
}

Config:

extra: str = forbid
validate_assignment: bool = True
validate_defaults: bool = True
strict: bool = True
str_strip_whitespace: bool = True

Fields:

policy_history (list[Tuple[aind_behavior_curriculum.curriculum.Policy, ...] | None])
stage_history (list[aind_behavior_curriculum.curriculum.Stage | None])

field policy_history: list[Tuple[Policy, ...] | None] = []¶

field stage_history: list[Stage | None] = []¶

add_entry(stage: Stage | None, policies: Tuple[Policy, ...] | None) → None¶: Add to stage and policy history synchronously.

peek_last_entry() → tuple[Stage | None, Tuple[Policy, ...] | None]¶: Return most-recently added entry.

class aind_behavior_curriculum.trainer.Trainer¶

Bases: object

Pulls subject curriculum and history, and performs fundamental curriculum evaluation/update.

Intended usage: 1) Implement abstract methods 2) Call Trainer.register_subject() x N 3) Call Trainer.evaluate_subject() or

Trainer.override_subject_status() x N

eject_subject(s_id: int) → None¶: Send mouse off curriculum. Only way to get mouse back into system is with Trainer.override_subject_status(…)

evaluate_subjects() → None¶

Calls user-defined functions to automatically update subject stage along curriculum. The timestep between evaluate_subject calls is flexible– this function will skip subject to the latest stage/policy they are applicable for.

Evaluation checks for stage transitions before policy transitions.

If subject does not satisfy any transition criteria, this method creates a duplicate current (stage, policy) entry in stage history.

abstract load_data(subject_id: int) → tuple[Curriculum, SubjectHistory, Metrics]¶: User-defined. Loads 3 pieces of data in the following format: - subject Curriculum - subject History - subject Metrics

override_subject_status(s_id: int, override_stage: Stage, override_policies: Policy | list[Policy]) → None¶

Override subject (stage, policies) independent of evaluation. Stage and Policy objects may be accessed by calling Trainer.load_data and looking inside of the returned Curriculum.

(Soft Rejection– send mouse to Stage/Policy w/in Curriculum)

register_subject(subject_id: int, curriculum: Curriculum, start_stage: Stage, start_policies: Policy | list[Policy] | None = None) → None¶: Adds subject into the Trainer system. If start_policies is None, registration defaults to the Stage.start_policies.

abstract write_data(subject_id: int, curriculum: Curriculum, history: SubjectHistory) → None¶

User-defined. Exports 3 pieces of data to database. - subject Id - subject Curriculum - subject History

For Curriculums with no internal policies, insert tacit INIT_STAGE

Module contents¶

Init package

Modules in this file are public facing.