Skip to content

Environment

Package defining environment related functionality.

MultiTaskAmbient

Bases: SVGAmbient

This class manages an SVG state and a collection of tasks that a user may want to interact with. It requires that a special agent the Avatar is provided which will act as an interface between the user and the environment. For details on what a Task is and how they are used see the MultiTaskEnvironment class documentation or read the icua wiki page on tasks (TODO provide a link to this).

Source code in icua\environment\multitask_ambient.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
class MultiTaskAmbient(SVGAmbient):
    """This class manages an SVG state and a collection of tasks that a user may want to interact with. It requires that a special agent the `Avatar` is provided which will act as an interface between the user and the environment. For details on what a `Task` is and how they are used see the `MultiTaskEnvironment` class documentation or read the `icua` wiki page on tasks (TODO provide a link to this)."""

    def __init__(
        self,
        avatar: Agent = None,
        agents: list[Agent] = None,
        svg_size: tuple[float, float] = None,
        svg_position: tuple[float, float] = None,
        logging_path: str = None,
        **kwargs,
    ):
        """Constructor.

        Args:
            avatar (Agent, optional): The users avatar. Defaults to None.
            agents (list[Agent], optional): list of initial agents. Defaults to None.
            svg_size (tuple[float, float], optional): size of the root SVG element, typically this should encompase the bounds of all tasks. Defaults to None, which will use the default `SVGAmbient.DEFAULT_SVG_SIZE`.
            svg_position (tuple[float, float], optional): position of the root SVG element in the UI window. Defaults to None, which which will use the default `SVGAmbient.DEFAULT_SVG_POSITION`.
            logging_path (str, optional): path that events will be logged to. Defaults to None  (see `MultiTaskAmbient` for details).
            kwargs (dict[str, Any]): Additional optional keyword arguments, see `SVGAmbient` for options.
        """
        # this needs to happen before any call to __update__
        self._event_logger = None
        self._initialise_logging(logging_path=logging_path)
        # initialise agents
        agents = agents if agents else []
        agents.append(avatar)
        super().__init__(
            agents,
            svg_size=svg_size,
            svg_position=svg_position,
            **kwargs,
        )
        self._avatar = avatar
        self._task_loader = TaskLoader()
        self._tasks: dict[str, Task] = dict()

    def _initialise_logging(self, logging_path: str = None):
        if logging_path:
            logging_file = EventLogger.default_log_path(
                path=logging_path, name="event_log_{datetime}.log"
            )
            # general event logging -- all calls to __update__ will trigger this logger
            self._event_logger = EventLogger(logging_file)

    def add_task(
        self,
        name: str,
        path: str | list[str],
        agent_actuators: list[Callable[[], Actuator]] | None = None,
        avatar_actuators: list[Callable[[], Actuator]] | None = None,
        enable: bool = False,
    ):
        """Add a new task, this will load all required files and prepare the task but will not start the task (unless `enable` is True). To start the task use `enable_task` or have an agent take the `EnableTask` action.

        Args:
            name (str): the unique name of the task.
            path (str | list[str]): path(s) to task files.
            agent_actuators (list[Callable[[], Actuator]] | None, optional): actuators with which to create agents from a schedule file, see `MultiTaskEnvironment` documentation for details. Defaults to None.
            avatar_actuators (list[Callable[[], Actuator]] | None, optional): actuators that will be added to the avatar upon enabling the task, see `MultiTaskEnvironment` documentation for details. Defaults to None.
            enable (bool, optional): whether to immediately enable the task. Defaults to False.
        """
        if name in self._tasks:
            raise ValueError(f"Task with name {name} already exists.")
        if agent_actuators is None:
            agent_actuators = []
        if avatar_actuators is None:
            avatar_actuators = []
        self._task_loader.register_task(name, path)
        self._tasks[name] = self._task_loader.load(
            name, avatar_actuators=avatar_actuators, agent_actuators=agent_actuators
        )
        if enable:
            self.enable_task(name)

    def enable_task(
        self, task_name: str, context: dict[str, Any] | None = None, insert_at: int = -1
    ):
        """Manually enable a task. Tasks may otherwise be enabled via an `EnableTask` action.

        Args:
            task_name (str): name of the task to enable.
            context (dict[str,Any]): context to use when enabling the task
            insert_at (int): at what position in the SVG tree to insert the task element.
        """
        event = EnableTask(
            source=self.id, task_name=task_name, context=context, insert_at=insert_at
        )
        result = self.__update__(event)
        if isinstance(result, ErrorActiveObservation):
            raise result.exception()

    def disable_task(self, task_name: str):
        """Manually disable a task. Tasks may otherwise be disabled via an `EnableTask` action.

        Args:
            task_name (str): name of the task to disable.
        """
        event = DisableTask(source=self.id, task_name=task_name)
        result = self.__update__(event)
        if isinstance(result, ErrorActiveObservation):
            raise result.exception()

    def rename_task(self, task_name: str, new_name: str) -> None:
        """Rename a task. If the task is enabled this will alter the `id` attribute of the task element (TODO).

        Args:
            task_name (str): current name of the task.
            new_name (str): new name of the task.
        """
        if task_name in self._tasks:
            if new_name in self._tasks:
                raise ValueError(
                    f"Failed to rename task: {task_name}, another take with name: {new_name} already exists."
                )
            task = self._tasks[new_name]
            self._tasks[task_name] = task
            task.task_name = new_name
            del self._tasks[task_name]
            # TODO update the name of the task element if it has been enabled!
        else:
            raise ValueError(f"Failed to rename task: {task_name} as it doesn't exist.")

    def __update__(self, action: Event) -> ActiveObservation | ErrorActiveObservation:  # noqa
        # always log the action immediately (before execution)
        if self._event_logger:
            self._event_logger.log(action)
        # execute the action here or in super()
        if isinstance(action, EnableTask):
            return self._enable_task(action)
        elif isinstance(action, DisableTask):
            return self._disable_task(action)
        elif isinstance(action, INERT_ACTIONS):
            # these actions have no effect but are important for experiment logging
            return None
        else:
            return super().__update__(action)

    def on_user_input_event(  # noqa
        self,
        action: EyeMotionEvent
        | EyeMotionEventRaw
        | MouseButtonEvent
        | MouseMotionEvent
        | KeyEvent
        | WindowCloseEvent
        | WindowOpenEvent
        | WindowFocusEvent
        | WindowMoveEvent
        | WindowResizeEvent
        | ScreenSizeEvent,
    ):
        return super().on_user_input_event(action)

    def is_task_enabled(self, task_name: str) -> bool:
        """Is the given task enabled? Specially, is the task element part of the state? This will search for an element with `id` equal to the name of the task.

        Note that if the `id` of the task element has been changed elsewherethen this will not give the expected result. This will not happen with correct use of `MultiTaskAmbient`.

        Args:
            task_name (str): name of the task to check.

        Returns:
            bool: True if the task is enabled (is part of the state), False otherwise.
        """
        try:
            result = self._state.select(
                Select.new(f"/svg:svg/*[@id='{task_name}']", ["id"])
            )
        except XPathElementsNotFound:
            return False
        return result is not None and len(result) > 0

    @wrap_observation
    def _disable_task(self, event: DisableTask):
        """Disables a task - this will be called when a `DisableTask` event is received."""
        raise NotImplementedError("TODO")  # TODO

    @wrap_observation
    def _enable_task(
        self, event: EnableTask
    ) -> ActiveObservation | ErrorActiveObservation:
        """Enables a task - this will be called when an `EnableTask` event is received."""
        task = self._tasks[event.task_name]
        # TODO this assumes that the id of the task is "task_name"!
        if self.is_task_enabled(event.task_name):
            raise ValueError(f"Task {event.task_name} is already enabled.")
        agent = task.get_agent()
        if agent:
            self.add_agent(agent)
        avatar = task.get_avatar(self._avatar)
        # this should not change... the task should only add relevant actuators
        assert self._avatar == avatar
        # default is to insert as the first child of the root
        # TODO we need to check what the task 'id' is to keep track of which tasks are enabled and active!
        return self.__update__(
            Insert(
                source=self.id,
                xpath="/svg:svg",
                element=task.get_xml(event.context),
                index=event.insert_at,
            )
        )

__init__(avatar=None, agents=None, svg_size=None, svg_position=None, logging_path=None, **kwargs)

Constructor.

Parameters:

Name Type Description Default
avatar Agent

The users avatar. Defaults to None.

 None
agents list[Agent]

list of initial agents. Defaults to None.

 None
svg_size tuple[float, float]

size of the root SVG element, typically this should encompase the bounds of all tasks. Defaults to None, which will use the default SVGAmbient.DEFAULT_SVG_SIZE.

 None
svg_position tuple[float, float]

position of the root SVG element in the UI window. Defaults to None, which which will use the default SVGAmbient.DEFAULT_SVG_POSITION.

 None
logging_path str

path that events will be logged to. Defaults to None (see MultiTaskAmbient for details).

 None
kwargs dict[str, Any]

Additional optional keyword arguments, see SVGAmbient for options.

 {}
Source code in icua\environment\multitask_ambient.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(
    self,
    avatar: Agent = None,
    agents: list[Agent] = None,
    svg_size: tuple[float, float] = None,
    svg_position: tuple[float, float] = None,
    logging_path: str = None,
    **kwargs,
):
    """Constructor.

    Args:
        avatar (Agent, optional): The users avatar. Defaults to None.
        agents (list[Agent], optional): list of initial agents. Defaults to None.
        svg_size (tuple[float, float], optional): size of the root SVG element, typically this should encompase the bounds of all tasks. Defaults to None, which will use the default `SVGAmbient.DEFAULT_SVG_SIZE`.
        svg_position (tuple[float, float], optional): position of the root SVG element in the UI window. Defaults to None, which which will use the default `SVGAmbient.DEFAULT_SVG_POSITION`.
        logging_path (str, optional): path that events will be logged to. Defaults to None  (see `MultiTaskAmbient` for details).
        kwargs (dict[str, Any]): Additional optional keyword arguments, see `SVGAmbient` for options.
    """
    # this needs to happen before any call to __update__
    self._event_logger = None
    self._initialise_logging(logging_path=logging_path)
    # initialise agents
    agents = agents if agents else []
    agents.append(avatar)
    super().__init__(
        agents,
        svg_size=svg_size,
        svg_position=svg_position,
        **kwargs,
    )
    self._avatar = avatar
    self._task_loader = TaskLoader()
    self._tasks: dict[str, Task] = dict()

add_task(name, path, agent_actuators=None, avatar_actuators=None, enable=False)

Add a new task, this will load all required files and prepare the task but will not start the task (unless enable is True). To start the task use enable_task or have an agent take the EnableTask action.

Parameters:

Name Type Description Default
name str

the unique name of the task.

 required
path str | list[str]

path(s) to task files.

 required
agent_actuators list[Callable[[], Actuator]] | None

actuators with which to create agents from a schedule file, see MultiTaskEnvironment documentation for details. Defaults to None.

 None
avatar_actuators list[Callable[[], Actuator]] | None

actuators that will be added to the avatar upon enabling the task, see MultiTaskEnvironment documentation for details. Defaults to None.

 None
enable bool

whether to immediately enable the task. Defaults to False.

 False
Source code in icua\environment\multitask_ambient.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def add_task(
    self,
    name: str,
    path: str | list[str],
    agent_actuators: list[Callable[[], Actuator]] | None = None,
    avatar_actuators: list[Callable[[], Actuator]] | None = None,
    enable: bool = False,
):
    """Add a new task, this will load all required files and prepare the task but will not start the task (unless `enable` is True). To start the task use `enable_task` or have an agent take the `EnableTask` action.

    Args:
        name (str): the unique name of the task.
        path (str | list[str]): path(s) to task files.
        agent_actuators (list[Callable[[], Actuator]] | None, optional): actuators with which to create agents from a schedule file, see `MultiTaskEnvironment` documentation for details. Defaults to None.
        avatar_actuators (list[Callable[[], Actuator]] | None, optional): actuators that will be added to the avatar upon enabling the task, see `MultiTaskEnvironment` documentation for details. Defaults to None.
        enable (bool, optional): whether to immediately enable the task. Defaults to False.
    """
    if name in self._tasks:
        raise ValueError(f"Task with name {name} already exists.")
    if agent_actuators is None:
        agent_actuators = []
    if avatar_actuators is None:
        avatar_actuators = []
    self._task_loader.register_task(name, path)
    self._tasks[name] = self._task_loader.load(
        name, avatar_actuators=avatar_actuators, agent_actuators=agent_actuators
    )
    if enable:
        self.enable_task(name)

disable_task(task_name)

Manually disable a task. Tasks may otherwise be disabled via an EnableTask action.

Parameters:

Name Type Description Default
task_name str

name of the task to disable.

 required
Source code in icua\environment\multitask_ambient.py 
140
141
142
143
144
145
146
147
148
149
def disable_task(self, task_name: str):
    """Manually disable a task. Tasks may otherwise be disabled via an `EnableTask` action.

    Args:
        task_name (str): name of the task to disable.
    """
    event = DisableTask(source=self.id, task_name=task_name)
    result = self.__update__(event)
    if isinstance(result, ErrorActiveObservation):
        raise result.exception()

enable_task(task_name, context=None, insert_at=-1)

Manually enable a task. Tasks may otherwise be enabled via an EnableTask action.

Parameters:

Name Type Description Default
task_name str

name of the task to enable.

 required
context dict[str, Any]

context to use when enabling the task

 None
insert_at int

at what position in the SVG tree to insert the task element.

 -1
Source code in icua\environment\multitask_ambient.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def enable_task(
    self, task_name: str, context: dict[str, Any] | None = None, insert_at: int = -1
):
    """Manually enable a task. Tasks may otherwise be enabled via an `EnableTask` action.

    Args:
        task_name (str): name of the task to enable.
        context (dict[str,Any]): context to use when enabling the task
        insert_at (int): at what position in the SVG tree to insert the task element.
    """
    event = EnableTask(
        source=self.id, task_name=task_name, context=context, insert_at=insert_at
    )
    result = self.__update__(event)
    if isinstance(result, ErrorActiveObservation):
        raise result.exception()

is_task_enabled(task_name)

Is the given task enabled? Specially, is the task element part of the state? This will search for an element with id equal to the name of the task.

Note that if the id of the task element has been changed elsewherethen this will not give the expected result. This will not happen with correct use of MultiTaskAmbient.

Parameters:

Name Type Description Default
task_name str

name of the task to check.

 required

Returns:

Name Type Description
bool bool

True if the task is enabled (is part of the state), False otherwise.
Source code in icua\environment\multitask_ambient.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def is_task_enabled(self, task_name: str) -> bool:
    """Is the given task enabled? Specially, is the task element part of the state? This will search for an element with `id` equal to the name of the task.

    Note that if the `id` of the task element has been changed elsewherethen this will not give the expected result. This will not happen with correct use of `MultiTaskAmbient`.

    Args:
        task_name (str): name of the task to check.

    Returns:
        bool: True if the task is enabled (is part of the state), False otherwise.
    """
    try:
        result = self._state.select(
            Select.new(f"/svg:svg/*[@id='{task_name}']", ["id"])
        )
    except XPathElementsNotFound:
        return False
    return result is not None and len(result) > 0

rename_task(task_name, new_name)

Rename a task. If the task is enabled this will alter the id attribute of the task element (TODO).

Parameters:

Name Type Description Default
task_name str

current name of the task.

 required
new_name str

new name of the task.

 required
Source code in icua\environment\multitask_ambient.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def rename_task(self, task_name: str, new_name: str) -> None:
    """Rename a task. If the task is enabled this will alter the `id` attribute of the task element (TODO).

    Args:
        task_name (str): current name of the task.
        new_name (str): new name of the task.
    """
    if task_name in self._tasks:
        if new_name in self._tasks:
            raise ValueError(
                f"Failed to rename task: {task_name}, another take with name: {new_name} already exists."
            )
        task = self._tasks[new_name]
        self._tasks[task_name] = task
        task.task_name = new_name
        del self._tasks[task_name]
        # TODO update the name of the task element if it has been enabled!
    else:
        raise ValueError(f"Failed to rename task: {task_name} as it doesn't exist.")

MultiTaskEnvironment

Bases: Environment

Environment implementation that supports "tasks". A task is a modular part of the environment which the user typically interacts with via their Avatar. Tasks will typially have an associated agent that updates/manages the tasks state. The users avatar should also provide some means to interact with tasks. When a new task is added, this may mean the avatar gains a new actuator and will often mean that a new agent is added to the environment.

TODO a thorough explaination of what tasks are and how to define/use them!

See

MultiTaskAmbient

Source code in icua\environment\multitask_environment.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
class MultiTaskEnvironment(Environment):
    """Environment implementation that supports "tasks". A task is a modular part of the environment which the user typically interacts with via their `Avatar`. Tasks will typially have an associated agent that updates/manages the tasks state. The users avatar should also provide some means to interact with tasks. When a new task is added, this may mean the avatar gains a new actuator and will often mean that a new agent is added to the environment.

    TODO a thorough explaination of what tasks are and how to define/use them!

    See:
        `MultiTaskAmbient`
    """

    def __init__(
        self,
        avatar: Agent,
        agents: list[Agent] = None,
        wait: float = 0.01,
        svg_size: tuple[float, float] = None,
        svg_position: tuple[float, float] = None,
        logging_path: str = None,
        terminate_after: float = -1,
        **kwargs: dict[str, Any],
    ):
        """Constructor.

        Args:
            avatar (Agent, optional): The users avatar. Defaults to None.
            agents (list[Agent], optional): list of initial agents. Defaults to None.
            wait (float, optional): time to wait between simulation cycles. Defaults to 0.01.
            svg_size (tuple[float, float], optional): size of the root SVG element. Defaults to None (see `MultiTaskAmbient` for details).
            svg_position (tuple[float, float], optional): position of the root SVG element. Defaults to None (see `MultiTaskAmbient` for details).
            logging_path (str, optional): path that events will be logged to. Defaults to None (see `MultiTaskAmbient` for details).
            terminate_after (float, optional): time after which to terminate the simulatio. Defaults to -1 (never terminate).
            kwargs (dict[str, Any]): Additional optional keyword arguments, see `MultiTaskAmbient` for options.
        """
        ambient = MultiTaskAmbient(
            avatar=avatar,
            agents=agents,
            logging_path=logging_path,
            svg_size=svg_size,
            svg_position=svg_position,
            **kwargs,
        )
        super().__init__(
            ambient=ambient,
            wait=wait,
            sync=True,
        )
        # time until termination
        self._terminate_after = terminate_after

    @property
    def ambient(self) -> MultiTaskAmbient:
        """Getter for the inner ambient, which is always a `MultiTaskAmbient`. IMPORTANT NOTE: remote ambients are not currently supported.

        Returns:
            MultiTaskAmbient: the ambient.
        """
        ambient = self._ambient._inner
        # TODO remote ambient currently not supported - this would be easy to do... (just need relevant methods exposed)
        assert isinstance(ambient, MultiTaskAmbient)
        return ambient

    def enable_task(
        self, task_name: str, context: dict[str, Any] = None, insert_at: int = -1
    ):
        """Manually enable a task. Tasks may otherwise be enabled via an `EnableTask` action.

        Args:
            task_name (str): name of the task to enable.
            context (dict[str,Any]): context to use when enabling the task
            insert_at (int): at what position in the SVG tree to insert the task element.
        """
        self.ambient.enable_task(task_name, context=context, insert_at=insert_at)

    def disable_task(self, task_name: str):
        """Manually disable a task. Tasks may otherwise be disabled via an `EnableTask` action.

        Args:
            task_name (str): name of the task to disable.
        """
        self.ambient.disable_task(task_name)

    def rename_task(self, task_name: str, new_name: str) -> None:
        """Rename a task. If the task is enabled this will alter the `id` attribute of the task element (TODO).

        Args:
            task_name (str): current name of the task.
            new_name (str): new name of the task.
        """
        return self.ambient.rename_task(task_name, new_name)

    def add_task(
        self,
        name: str,
        path: str,
        agent_actuators: list[Callable[[], Actuator]] = None,
        avatar_actuators: list[Callable[[], Actuator]] = None,
        enable: bool = False,
    ):
        """Add a new task, this will load all required files and prepare the task but will not start the task (unless `enable` is True). To start the task use `enable_task` or have an agent take the `EnableTask` action.

        Args:
            name (str): the unique name of the task.
            path (str | list[str]): path(s) to task files.
            agent_actuators (list[Callable[[], Actuator]] | None, optional): actuators with which to create agents from a schedule file, see `MultiTaskEnvironment` documentation for details. Defaults to None.
            avatar_actuators (list[Callable[[], Actuator]] | None, optional): actuators that will be added to the avatar upon enabling the task, see `MultiTaskEnvironment` documentation for details. Defaults to None.
            enable (bool, optional): whether to immediately enable the task. Defaults to False.
        """
        # attempt to log the files that will be used in the task
        self.ambient.add_task(
            name,
            path,
            agent_actuators=agent_actuators,
            avatar_actuators=avatar_actuators,
            enable=enable,
        )

    async def __initialise__(self, event_loop):  # noqa
        await self._ambient.__initialise__()

    def get_schedule(self) -> list[asyncio.Task]:  # noqa
        tasks = super().get_schedule()
        # whether to terminate after a given period of time
        if self._terminate_after > 0:

            async def _terminate_after(env: MultiTaskEnvironment):
                try:
                    await asyncio.sleep(env._terminate_after)
                    # TODO perhaps this should be an event that is logged?
                    LOGGER.debug(
                        f"Closing simulation: time limit ({env._terminate_after}s) reached"
                    )
                    await env.ambient.__terminate__()
                except asyncio.CancelledError:
                    pass
                except asyncio.TimeoutError:
                    pass

            tasks.append(asyncio.create_task(_terminate_after(self)))
        return tasks

ambient: MultiTaskAmbient property

Getter for the inner ambient, which is always a MultiTaskAmbient. IMPORTANT NOTE: remote ambients are not currently supported.

Returns:

Name Type Description
MultiTaskAmbient MultiTaskAmbient

the ambient.

__init__(avatar, agents=None, wait=0.01, svg_size=None, svg_position=None, logging_path=None, terminate_after=-1, **kwargs)

Constructor.

Parameters:

Name Type Description Default
avatar Agent

The users avatar. Defaults to None.

 required
agents list[Agent]

list of initial agents. Defaults to None.

 None
wait float

time to wait between simulation cycles. Defaults to 0.01.

 0.01
svg_size tuple[float, float]

size of the root SVG element. Defaults to None (see MultiTaskAmbient for details).

 None
svg_position tuple[float, float]

position of the root SVG element. Defaults to None (see MultiTaskAmbient for details).

 None
logging_path str

path that events will be logged to. Defaults to None (see MultiTaskAmbient for details).

 None
terminate_after float

time after which to terminate the simulatio. Defaults to -1 (never terminate).

 -1
kwargs dict[str, Any]

Additional optional keyword arguments, see MultiTaskAmbient for options.

 {}
Source code in icua\environment\multitask_environment.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def __init__(
    self,
    avatar: Agent,
    agents: list[Agent] = None,
    wait: float = 0.01,
    svg_size: tuple[float, float] = None,
    svg_position: tuple[float, float] = None,
    logging_path: str = None,
    terminate_after: float = -1,
    **kwargs: dict[str, Any],
):
    """Constructor.

    Args:
        avatar (Agent, optional): The users avatar. Defaults to None.
        agents (list[Agent], optional): list of initial agents. Defaults to None.
        wait (float, optional): time to wait between simulation cycles. Defaults to 0.01.
        svg_size (tuple[float, float], optional): size of the root SVG element. Defaults to None (see `MultiTaskAmbient` for details).
        svg_position (tuple[float, float], optional): position of the root SVG element. Defaults to None (see `MultiTaskAmbient` for details).
        logging_path (str, optional): path that events will be logged to. Defaults to None (see `MultiTaskAmbient` for details).
        terminate_after (float, optional): time after which to terminate the simulatio. Defaults to -1 (never terminate).
        kwargs (dict[str, Any]): Additional optional keyword arguments, see `MultiTaskAmbient` for options.
    """
    ambient = MultiTaskAmbient(
        avatar=avatar,
        agents=agents,
        logging_path=logging_path,
        svg_size=svg_size,
        svg_position=svg_position,
        **kwargs,
    )
    super().__init__(
        ambient=ambient,
        wait=wait,
        sync=True,
    )
    # time until termination
    self._terminate_after = terminate_after

add_task(name, path, agent_actuators=None, avatar_actuators=None, enable=False)

Add a new task, this will load all required files and prepare the task but will not start the task (unless enable is True). To start the task use enable_task or have an agent take the EnableTask action.

Parameters:

Name Type Description Default
name str

the unique name of the task.

 required
path str | list[str]

path(s) to task files.

 required
agent_actuators list[Callable[[], Actuator]] | None

actuators with which to create agents from a schedule file, see MultiTaskEnvironment documentation for details. Defaults to None.

 None
avatar_actuators list[Callable[[], Actuator]] | None

actuators that will be added to the avatar upon enabling the task, see MultiTaskEnvironment documentation for details. Defaults to None.

 None
enable bool

whether to immediately enable the task. Defaults to False.

 False
Source code in icua\environment\multitask_environment.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def add_task(
    self,
    name: str,
    path: str,
    agent_actuators: list[Callable[[], Actuator]] = None,
    avatar_actuators: list[Callable[[], Actuator]] = None,
    enable: bool = False,
):
    """Add a new task, this will load all required files and prepare the task but will not start the task (unless `enable` is True). To start the task use `enable_task` or have an agent take the `EnableTask` action.

    Args:
        name (str): the unique name of the task.
        path (str | list[str]): path(s) to task files.
        agent_actuators (list[Callable[[], Actuator]] | None, optional): actuators with which to create agents from a schedule file, see `MultiTaskEnvironment` documentation for details. Defaults to None.
        avatar_actuators (list[Callable[[], Actuator]] | None, optional): actuators that will be added to the avatar upon enabling the task, see `MultiTaskEnvironment` documentation for details. Defaults to None.
        enable (bool, optional): whether to immediately enable the task. Defaults to False.
    """
    # attempt to log the files that will be used in the task
    self.ambient.add_task(
        name,
        path,
        agent_actuators=agent_actuators,
        avatar_actuators=avatar_actuators,
        enable=enable,
    )

disable_task(task_name)

Manually disable a task. Tasks may otherwise be disabled via an EnableTask action.

Parameters:

Name Type Description Default
task_name str

name of the task to disable.

 required
Source code in icua\environment\multitask_environment.py 
85
86
87
88
89
90
91
def disable_task(self, task_name: str):
    """Manually disable a task. Tasks may otherwise be disabled via an `EnableTask` action.

    Args:
        task_name (str): name of the task to disable.
    """
    self.ambient.disable_task(task_name)

enable_task(task_name, context=None, insert_at=-1)

Manually enable a task. Tasks may otherwise be enabled via an EnableTask action.

Parameters:

Name Type Description Default
task_name str

name of the task to enable.

 required
context dict[str, Any]

context to use when enabling the task

 None
insert_at int

at what position in the SVG tree to insert the task element.

 -1
Source code in icua\environment\multitask_environment.py 
73
74
75
76
77
78
79
80
81
82
83
def enable_task(
    self, task_name: str, context: dict[str, Any] = None, insert_at: int = -1
):
    """Manually enable a task. Tasks may otherwise be enabled via an `EnableTask` action.

    Args:
        task_name (str): name of the task to enable.
        context (dict[str,Any]): context to use when enabling the task
        insert_at (int): at what position in the SVG tree to insert the task element.
    """
    self.ambient.enable_task(task_name, context=context, insert_at=insert_at)

rename_task(task_name, new_name)

Rename a task. If the task is enabled this will alter the id attribute of the task element (TODO).

Parameters:

Name Type Description Default
task_name str

current name of the task.

 required
new_name str

new name of the task.

 required
Source code in icua\environment\multitask_environment.py 
 93
 94
 95
 96
 97
 98
 99
100
def rename_task(self, task_name: str, new_name: str) -> None:
    """Rename a task. If the task is enabled this will alter the `id` attribute of the task element (TODO).

    Args:
        task_name (str): current name of the task.
        new_name (str): new name of the task.
    """
    return self.ambient.rename_task(task_name, new_name)