Agent default

Module contains a default implementation for a guidance agent, see DefaultGuidanceAgent documentation for details.

ArrowGuidanceActuator

Bases: GuidanceActuator

A concrete implementation of GuidanceActuator that implements a guidance arrow.

The arrow can be displayed in three different ways depending on the value of arrow_mode: - "gaze" will display the arrow at the users gaze position, offset by arrow_offset. - "mouse" will display the arrow at the users mouse position, offset by arrow_offset. - "fixed" will display the arrow at a fixed position on the screen, this position is specified by arrow_offset.

Source code in icua\agent\actuator_guidance.py

class ArrowGuidanceActuator(GuidanceActuator):
    """A concrete implementation of `GuidanceActuator` that implements a guidance arrow.

    The arrow can be displayed in three different ways depending on the value of `arrow_mode`:
    - "gaze" will display the arrow at the users gaze position, offset by `arrow_offset`.
    - "mouse" will display the arrow at the users mouse position, offset by `arrow_offset`.
    - "fixed" will display the arrow at a fixed position on the screen, this position is specified by `arrow_offset`.
    """

    ARROW_MODES = Literal["gaze", "mouse", "fixed"]

    def __init__(
        self,
        arrow_mode: Literal["gaze", "mouse", "fixed"],
        arrow_scale: float = 1.0,
        arrow_fill_color: str = "none",
        arrow_stroke_color: str = "#ff0000",
        arrow_stroke_width: float = 4.0,
        arrow_offset: tuple[float, float] = (80, 80),
    ):
        """Constructor.

        Args:
            arrow_mode (Literal["gaze", "mouse", "fixed"]): modes for arrow display.
            arrow_scale (float, optional): scale of the arrow. Defaults to 1.0.
            arrow_fill_color (str, optional): fill colour of the arrow. Defaults to "none".
            arrow_stroke_color (str, optional): line colour of the arrow outline. Defaults to "#ff0000".
            arrow_stroke_width (float, optional): line width of the arrow outline. Defaults to 4.0.
            arrow_offset (tuple[float, float], optional): offset of the arrow from its set position. This is the position used if `arrow_mode == "fixed". Defaults to (80, 80).
        """
        super().__init__()
        self._arrow_mode = arrow_mode
        self._arrow_scale = arrow_scale
        self._arrow_fill_color = arrow_fill_color
        self._arrow_stroke_color = arrow_stroke_color
        self._arrow_stroke_width = arrow_stroke_width
        self._arrow_offset = arrow_offset

        if self._arrow_mode not in ArrowGuidanceActuator.ARROW_MODES.__args__:
            raise ValueError(
                f"Invalid argument: `arrow_mode` must be one of {ArrowGuidanceActuator.ARROW_MODES}"
            )
        self._guidance_arrow_id = "guidance_arrow"
        self._guidance_on = None
        self._gaze_position = None
        self._mouse_position = None

    def __attempt__(self):  # noqa
        if self._guidance_on is None:
            return []
        elif self._arrow_mode == "gaze":
            # eyetracking positions can be nan, dont update the position if they are?
            if isfinite(self._gaze_position[0]) and isfinite(self._gaze_position[1]):
                attrs = dict(
                    id=self._guidance_arrow_id,
                    x=self._gaze_position[0] + self._arrow_offset[0],
                    y=self._gaze_position[1] + self._arrow_offset[0],
                    point_to=self._guidance_on,
                )
                return [DrawArrowAction(xpath="/svg:svg", data=attrs)]
            else:
                LOGGER.warning("Ignoring NaN arrow position.")
            return []
        elif self._arrow_mode == "mouse":
            if self._mouse_position:
                attrs = dict(
                    id=self._guidance_arrow_id,
                    x=self._mouse_position[0] + self._arrow_offset[0],
                    y=self._mouse_position[1] + self._arrow_offset[0],
                    point_to=self._guidance_on,
                )
                return [DrawArrowAction(xpath="/svg:svg", data=attrs)]
            return []
        elif self._arrow_mode == "fixed":
            # TODO where should it be? the center of the screen?
            raise NotImplementedError("TODO")
        else:
            raise ValueError(
                f"Invalid argument: `arrow_mode` must be one of {ArrowGuidanceActuator.ARROW_MODES}"
            )

    @attempt([EyeMotionEvent])
    def set_gaze_position(self, action: EyeMotionEvent) -> None:
        """Sets the users current gaze position. This may be used as a position for arrow display."""
        self._gaze_position = action.position

    @attempt([MouseMotionEvent])
    def set_mouse_motion(self, action: MouseMotionEvent) -> None:
        """Sets the users current mouse position. This may be used as a position for arrow display."""
        self._mouse_position = action.position

    def on_add(self, agent: Agent) -> None:  # noqa
        super().on_add(agent)
        self.draw_guidance_arrow(
            self._guidance_arrow_id,
            0.0,
            0.0,
            fill=self._arrow_fill_color,
            stroke_width=self._arrow_stroke_width,
            stroke_color=self._arrow_stroke_color,
            opacity=0.0,
            scale=self._arrow_scale,
        )

    def on_remove(self, agent: Agent) -> None:  # noqa
        super().on_remove(agent)
        # TODO remove arrow element!

    @attempt()
    def draw_guidance_arrow(
        self,
        name: str,
        x: float,
        y: float,
        scale: float = 1.0,
        rotation: float = 0.0,
        fill: str = "none",
        opacity: float = 0.0,
        stroke_color: str = "#ff0000",
        stroke_width: float = 2.0,
        **kwargs,
    ) -> DrawArrowAction:
        """Attempt method that takes an action to draw a guidance arrow.

        Args:
            name (str): id.
            x (float): x position.
            y (float): y position.
            scale (float, optional): scale. Defaults to 1.0.
            rotation (float, optional): rotation. Defaults to 0.0.
            fill (str, optional): fill color. Defaults to "none".
            opacity (float, optional): opacity (0.0 means hidden). Defaults to 0.0.
            stroke_color (str, optional): color of the arrow border. Defaults to "#ff0000".
            stroke_width (float, optional): thickness of the arrow border. Defaults to 2.0.
            kwargs : (dict[str,Any]): additional optional keyword arguments.

        Returns:
            DrawArrowAction: action
        """
        return DrawArrowAction(
            xpath="/svg:svg",
            data=dict(
                id=name,
                x=x,
                y=y,
                scale=scale,
                rotation=rotation,
                fill=fill,
                opacity=opacity,
                stroke=stroke_color,
                stroke_width=stroke_width,
                **kwargs,
            ),
        )

    @attempt()
    def show_guidance(self, task: str) -> list[Action]:
        """Show guidance on the given task.

        Args:
            task (str): task to show guidance for.

        Returns:
            list[Action]: guidance actions
        """
        self._guidance_on = task
        actions = [
            ShowElementAction(xpath=f"//*[@id='{self._guidance_arrow_id}']"),
        ]
        return actions

    @attempt()
    def hide_guidance(self, task: str) -> list[Action]:
        """Hide guidance on the given task.

        Args:
            task (str): task to hide guidance for.

        Returns:
            list[Action]: guidance actions
        """
        self._guidance_on = None
        actions = [
            HideElementAction(xpath=f"//*[@id='{self._guidance_arrow_id}']"),
        ]
        return actions

__init__(arrow_mode, arrow_scale=1.0, arrow_fill_color='none', arrow_stroke_color='#ff0000', arrow_stroke_width=4.0, arrow_offset=(80, 80))

Constructor.

Parameters:

Name	Type	Description	Default
arrow_mode	Literal['gaze', 'mouse', 'fixed']	modes for arrow display.	required
arrow_scale	float	scale of the arrow. Defaults to 1.0.	1.0
arrow_fill_color	str	fill colour of the arrow. Defaults to "none".	'none'
arrow_stroke_color	str	line colour of the arrow outline. Defaults to "#ff0000".	'#ff0000'
arrow_stroke_width	float	line width of the arrow outline. Defaults to 4.0.	4.0
arrow_offset	tuple[float, float]	offset of the arrow from its set position. This is the position used if `arrow_mode == "fixed". Defaults to (80, 80).	(80, 80)

Source code in icua\agent\actuator_guidance.py

def __init__(
    self,
    arrow_mode: Literal["gaze", "mouse", "fixed"],
    arrow_scale: float = 1.0,
    arrow_fill_color: str = "none",
    arrow_stroke_color: str = "#ff0000",
    arrow_stroke_width: float = 4.0,
    arrow_offset: tuple[float, float] = (80, 80),
):
    """Constructor.

    Args:
        arrow_mode (Literal["gaze", "mouse", "fixed"]): modes for arrow display.
        arrow_scale (float, optional): scale of the arrow. Defaults to 1.0.
        arrow_fill_color (str, optional): fill colour of the arrow. Defaults to "none".
        arrow_stroke_color (str, optional): line colour of the arrow outline. Defaults to "#ff0000".
        arrow_stroke_width (float, optional): line width of the arrow outline. Defaults to 4.0.
        arrow_offset (tuple[float, float], optional): offset of the arrow from its set position. This is the position used if `arrow_mode == "fixed". Defaults to (80, 80).
    """
    super().__init__()
    self._arrow_mode = arrow_mode
    self._arrow_scale = arrow_scale
    self._arrow_fill_color = arrow_fill_color
    self._arrow_stroke_color = arrow_stroke_color
    self._arrow_stroke_width = arrow_stroke_width
    self._arrow_offset = arrow_offset

    if self._arrow_mode not in ArrowGuidanceActuator.ARROW_MODES.__args__:
        raise ValueError(
            f"Invalid argument: `arrow_mode` must be one of {ArrowGuidanceActuator.ARROW_MODES}"
        )
    self._guidance_arrow_id = "guidance_arrow"
    self._guidance_on = None
    self._gaze_position = None
    self._mouse_position = None

draw_guidance_arrow(name, x, y, scale=1.0, rotation=0.0, fill='none', opacity=0.0, stroke_color='#ff0000', stroke_width=2.0, **kwargs)

Attempt method that takes an action to draw a guidance arrow.

Parameters:

Name	Type	Description	Default
name	str	id.	required
x	float	x position.	required
y	float	y position.	required
scale	float	scale. Defaults to 1.0.	1.0
rotation	float	rotation. Defaults to 0.0.	0.0
fill	str	fill color. Defaults to "none".	'none'
opacity	float	opacity (0.0 means hidden). Defaults to 0.0.	0.0
stroke_color	str	color of the arrow border. Defaults to "#ff0000".	'#ff0000'
stroke_width	float	thickness of the arrow border. Defaults to 2.0.	2.0
kwargs		(dict[str,Any]): additional optional keyword arguments.	{}

Returns:

Name	Type	Description
DrawArrowAction	DrawArrowAction	action

Source code in icua\agent\actuator_guidance.py

@attempt()
def draw_guidance_arrow(
    self,
    name: str,
    x: float,
    y: float,
    scale: float = 1.0,
    rotation: float = 0.0,
    fill: str = "none",
    opacity: float = 0.0,
    stroke_color: str = "#ff0000",
    stroke_width: float = 2.0,
    **kwargs,
) -> DrawArrowAction:
    """Attempt method that takes an action to draw a guidance arrow.

    Args:
        name (str): id.
        x (float): x position.
        y (float): y position.
        scale (float, optional): scale. Defaults to 1.0.
        rotation (float, optional): rotation. Defaults to 0.0.
        fill (str, optional): fill color. Defaults to "none".
        opacity (float, optional): opacity (0.0 means hidden). Defaults to 0.0.
        stroke_color (str, optional): color of the arrow border. Defaults to "#ff0000".
        stroke_width (float, optional): thickness of the arrow border. Defaults to 2.0.
        kwargs : (dict[str,Any]): additional optional keyword arguments.

    Returns:
        DrawArrowAction: action
    """
    return DrawArrowAction(
        xpath="/svg:svg",
        data=dict(
            id=name,
            x=x,
            y=y,
            scale=scale,
            rotation=rotation,
            fill=fill,
            opacity=opacity,
            stroke=stroke_color,
            stroke_width=stroke_width,
            **kwargs,
        ),
    )

hide_guidance(task)

Hide guidance on the given task.

Parameters:

Name	Type	Description	Default
task	str	task to hide guidance for.	required

Returns:

Type	Description
list[Action]	list[Action]: guidance actions

Source code in icua\agent\actuator_guidance.py

@attempt()
def hide_guidance(self, task: str) -> list[Action]:
    """Hide guidance on the given task.

    Args:
        task (str): task to hide guidance for.

    Returns:
        list[Action]: guidance actions
    """
    self._guidance_on = None
    actions = [
        HideElementAction(xpath=f"//*[@id='{self._guidance_arrow_id}']"),
    ]
    return actions

set_gaze_position(action)

Sets the users current gaze position. This may be used as a position for arrow display.

Source code in icua\agent\actuator_guidance.py

@attempt([EyeMotionEvent])
def set_gaze_position(self, action: EyeMotionEvent) -> None:
    """Sets the users current gaze position. This may be used as a position for arrow display."""
    self._gaze_position = action.position

set_mouse_motion(action)

Sets the users current mouse position. This may be used as a position for arrow display.

Source code in icua\agent\actuator_guidance.py

@attempt([MouseMotionEvent])
def set_mouse_motion(self, action: MouseMotionEvent) -> None:
    """Sets the users current mouse position. This may be used as a position for arrow display."""
    self._mouse_position = action.position

show_guidance(task)

Show guidance on the given task.

Parameters:

Name	Type	Description	Default
task	str	task to show guidance for.	required

Returns:

Type	Description
list[Action]	list[Action]: guidance actions

Source code in icua\agent\actuator_guidance.py

@attempt()
def show_guidance(self, task: str) -> list[Action]:
    """Show guidance on the given task.

    Args:
        task (str): task to show guidance for.

    Returns:
        list[Action]: guidance actions
    """
    self._guidance_on = task
    actions = [
        ShowElementAction(xpath=f"//*[@id='{self._guidance_arrow_id}']"),
    ]
    return actions

BoxGuidanceActuator

Bases: GuidanceActuator

A concrete implementation of GuidanceActuator that implements box guidance. The box bounds a given task element serving to highlight it to the user, typically the task will be one that is not in an acceptible state.

Source code in icua\agent\actuator_guidance.py

class BoxGuidanceActuator(GuidanceActuator):
    """A concrete implementation of `GuidanceActuator` that implements box guidance. The box bounds a given task element serving to highlight it to the user, typically the task will be one that is not in an acceptible state."""

    def __init__(
        self,
        box_stroke_color: str = "#ff0000",
        box_stroke_width: float = 4.0,
    ):
        """Constructor.

        Args:
            box_stroke_color (str, optional): color of the box. Defaults to "#ff0000".
            box_stroke_width (float, optional): width of the box outline. Defaults to 4.0.
        """
        super().__init__()
        self._box_stroke_color = box_stroke_color
        self._box_stroke_width = box_stroke_width
        self._guidance_box_id_template = "guidance_box_%s"
        # ids of each of the guidance boxes that have been created
        # TODO they should be removed when `on_remove` is called!
        self._guidance_boxes = set()

    def on_remove(self, agent: Agent) -> None:  # noqa
        return super().on_remove(agent)  # TODO remove all guidance boxes!

    @attempt()
    def draw_guidance_box_on_element(
        self, element_id: str, **box_data: dict[str, Any]
    ) -> DrawBoxOnElementAction:
        """Attempt method that will draw a box around a given element.

        Args:
            element_id (str): the `id` of the element.
            box_data (dict[str,Any]): data associated with the box.

        Returns:
            DrawBoxOnElementAction: action
        """
        # TODO explicit parameters for box data
        box_data["id"] = self._guidance_box_id_template % element_id
        return DrawBoxOnElementAction(
            xpath=f"//*[@id='{element_id}']", box_data=box_data
        )

    @attempt()
    def show_guidance(self, task: str) -> list[Action]:
        """Show guidance on the given task.

        Args:
            task (str): task to show guidance for.

        Returns:
            list[Action]: guidance actions
        """
        self._guidance_on = task
        guidance_box_id = self._guidance_box_id_template % task

        actions = []

        if guidance_box_id not in self._guidance_boxes:
            # first time! insert the guidance box
            self._guidance_boxes.add(guidance_box_id)
            box_data = {
                "stroke-width": self._box_stroke_width,
                "stroke": self._box_stroke_color,
            }
            # draw the box but it is hidden (opacity=0)
            actions.append(
                self.draw_guidance_box_on_element(task, opacity=0.0, **box_data)
            )
        actions.append(ShowElementAction(xpath=f"//*[@id='{guidance_box_id}']"))
        return actions

    @attempt()
    def hide_guidance(self, task: str):
        """Hide guidance on the given task.

        Args:
            task (str): task to hide guidance for.

        Returns:
            list[Action]: guidance actions
        """
        self._guidance_on = None
        guidance_box_id = self._guidance_box_id_template % task

        actions = []

        if guidance_box_id not in self._guidance_boxes:
            # first time! insert the guidance box
            self._guidance_boxes.add(guidance_box_id)
            box_data = {
                "stroke-width": self._box_stroke_width,
                "stroke": self._box_stroke_color,
            }
            # draw the box but it is hidden (opacity=0)
            actions.append(
                self.draw_guidance_box_on_element(task, opacity=0.0, **box_data)
            )
        actions.append(HideElementAction(xpath=f"//*[@id='{guidance_box_id}']"))
        return actions

__init__(box_stroke_color='#ff0000', box_stroke_width=4.0)

Constructor.

Parameters:

Name	Type	Description	Default
box_stroke_color	str	color of the box. Defaults to "#ff0000".	'#ff0000'
box_stroke_width	float	width of the box outline. Defaults to 4.0.	4.0

Source code in icua\agent\actuator_guidance.py

def __init__(
    self,
    box_stroke_color: str = "#ff0000",
    box_stroke_width: float = 4.0,
):
    """Constructor.

    Args:
        box_stroke_color (str, optional): color of the box. Defaults to "#ff0000".
        box_stroke_width (float, optional): width of the box outline. Defaults to 4.0.
    """
    super().__init__()
    self._box_stroke_color = box_stroke_color
    self._box_stroke_width = box_stroke_width
    self._guidance_box_id_template = "guidance_box_%s"
    # ids of each of the guidance boxes that have been created
    # TODO they should be removed when `on_remove` is called!
    self._guidance_boxes = set()

draw_guidance_box_on_element(element_id, **box_data)

Attempt method that will draw a box around a given element.

Parameters:

Name	Type	Description	Default
element_id	str	the id of the element.	required
box_data	dict[str, Any]	data associated with the box.	{}

Returns:

Name	Type	Description
DrawBoxOnElementAction	DrawBoxOnElementAction	action

Source code in icua\agent\actuator_guidance.py

@attempt()
def draw_guidance_box_on_element(
    self, element_id: str, **box_data: dict[str, Any]
) -> DrawBoxOnElementAction:
    """Attempt method that will draw a box around a given element.

    Args:
        element_id (str): the `id` of the element.
        box_data (dict[str,Any]): data associated with the box.

    Returns:
        DrawBoxOnElementAction: action
    """
    # TODO explicit parameters for box data
    box_data["id"] = self._guidance_box_id_template % element_id
    return DrawBoxOnElementAction(
        xpath=f"//*[@id='{element_id}']", box_data=box_data
    )

hide_guidance(task)

Hide guidance on the given task.

Parameters:

Name	Type	Description	Default
task	str	task to hide guidance for.	required

Returns:

Type	Description
	list[Action]: guidance actions

Source code in icua\agent\actuator_guidance.py

@attempt()
def hide_guidance(self, task: str):
    """Hide guidance on the given task.

    Args:
        task (str): task to hide guidance for.

    Returns:
        list[Action]: guidance actions
    """
    self._guidance_on = None
    guidance_box_id = self._guidance_box_id_template % task

    actions = []

    if guidance_box_id not in self._guidance_boxes:
        # first time! insert the guidance box
        self._guidance_boxes.add(guidance_box_id)
        box_data = {
            "stroke-width": self._box_stroke_width,
            "stroke": self._box_stroke_color,
        }
        # draw the box but it is hidden (opacity=0)
        actions.append(
            self.draw_guidance_box_on_element(task, opacity=0.0, **box_data)
        )
    actions.append(HideElementAction(xpath=f"//*[@id='{guidance_box_id}']"))
    return actions

show_guidance(task)

Show guidance on the given task.

Parameters:

Name	Type	Description	Default
task	str	task to show guidance for.	required

Returns:

Type	Description
list[Action]	list[Action]: guidance actions

Source code in icua\agent\actuator_guidance.py

@attempt()
def show_guidance(self, task: str) -> list[Action]:
    """Show guidance on the given task.

    Args:
        task (str): task to show guidance for.

    Returns:
        list[Action]: guidance actions
    """
    self._guidance_on = task
    guidance_box_id = self._guidance_box_id_template % task

    actions = []

    if guidance_box_id not in self._guidance_boxes:
        # first time! insert the guidance box
        self._guidance_boxes.add(guidance_box_id)
        box_data = {
            "stroke-width": self._box_stroke_width,
            "stroke": self._box_stroke_color,
        }
        # draw the box but it is hidden (opacity=0)
        actions.append(
            self.draw_guidance_box_on_element(task, opacity=0.0, **box_data)
        )
    actions.append(ShowElementAction(xpath=f"//*[@id='{guidance_box_id}']"))
    return actions

DefaultGuidanceAgent

Bases: GuidanceAgent

Default implementation of a guidance agent for the matbii system.

Show guidance if

1. there is no guidance already active.
2. the task is unacceptable.
3. the user is not already attending on the task.
4. the grace period has elapsed.

Source code in matbii\guidance\agent_default.py

class DefaultGuidanceAgent(GuidanceAgent):
    """Default implementation of a guidance agent for the matbii system.

    Show guidance if:
        1. there is no guidance already active.
        2. the task is unacceptable.
        3. the user is not already attending on the task.
        4. the grace period has elapsed.
    """

    # used to break ties when multiple tasks could be highlighted. See `break_tie` method.
    BREAK_TIES = ("random", "longest")

    # condition 4. - whether to track the time since the last failure, or since the last guidance was shown
    GRACE_ON = ("guidance_task", "guidance_any", "failure", "attention")

    # method to use to determine which task the user is attending to
    ATTENTION_MODES = ("fixation", "gaze", "mouse")

    # if we havent had fresh eyetracking data for more than this, there may be something wrong... display a warning
    MISSING_GAZE_DATA_THRESHOLD = 0.1

    def __init__(
        self,
        sensors: list[Sensor],
        actuators: list[Actuator],
        break_ties: Literal["random", "longest"] = "random",
        grace_mode: Literal[
            "guidance_task", "guidance_any", "failure", "attention"
        ] = "failure",
        attention_mode: Literal["fixation", "gaze", "mouse"] = "fixation",
        grace_period: float = 3.0,
        counter_factual: bool = False,
        **kwargs: dict[str, Any],
    ):
        """Constructor.

        Args:
            sensors (list[Sensor]): list of sensors, this will typically be a list of `icua.agent.TaskAcceptabilitySensor`s. A `UserInputSensor` will always be added automatically.
            actuators (list[Actuator]): list of actuators, this will typically contain actuators that are capable of providing visual feedback to a user, see e.g. `icua.agent.GuidanceActuator` and its concrete implementations. A `CounterFactualGuidanceActuator` will be added by default.
            break_ties (Literal["random", "longest", "since"], optional): how to break ties if guidance may be shown on multiple tasks simultaneously. Defaults to "random". "random" will randomly break the tie, "longest" will choose the task that has been in failure for the longest, "since" will choose the task that has not had guidance shown for the longest time.
            grace_mode (Literal["guidance_task", "guidance_any", "failure", "attention"], optional): how to track the grace period. "guidance_task" will track the time since guidance was last shown on the task, "guidance_any" will track the time since guidance was last shown on any task, "failure" will track the time since the last failure on the task, "attention" will track the time since the user was last attending to a task. Defaults to "failure".
            attention_mode (Literal["fixation", "gaze", "mouse"], optional): method of determining where the user is attending. "fixation" will use the most recent gaze fixation, "gaze" will use the gaze position (including saccades), "mouse" will use the current mouse position. Defaults to "fixation".
            grace_period (float, optional): the time to wait (seconds) before guidance may be shown for a task after the last time guidance was shown on the task. Defaults to 3.0 seconds.
            counter_factual (bool, optional): whether guidance should be shown to the user, or whether it should just be logged.  This allows counter-factual experiments to be run, we can track when guidance would have been shown, and compare the when it was actually shown (in a different run). Defaults to False.
            kwargs (dict[str,Any]): Additional optional keyword arguments.
        """
        super().__init__(
            list(filter(None, sensors)),
            list(filter(None, actuators)),
            counter_factual=counter_factual,
            **kwargs,
        )
        self._attention_mode = attention_mode
        if self._attention_mode not in DefaultGuidanceAgent.ATTENTION_MODES:
            raise ValueError(
                f"Invalid argument: `attention_mode` {self._attention_mode} must be one of {DefaultGuidanceAgent.ATTENTION_MODES}"
            )

        self._grace_mode = grace_mode
        if self._grace_mode not in DefaultGuidanceAgent.GRACE_ON:
            raise ValueError(
                f"Invalid argument: `grace_mode` {self._grace_mode} must be one of {DefaultGuidanceAgent.GRACE_ON}"
            )

        self._break_ties = break_ties
        if self._break_ties not in DefaultGuidanceAgent.BREAK_TIES:
            raise ValueError(
                f"Invalid argument: `break_ties` {self._break_ties} must be one of {DefaultGuidanceAgent.BREAK_TIES}"
            )

        self._grace_period = grace_period  # TODO
        if self._grace_period < 0.0:
            raise ValueError(
                f"Invalid argument: `grace_period` {self._grace_period} must be > 0 "
            )

        # used to track the tasks that the user is currently attending to
        self._attending_tasks = set()

    def on_attending(self, attending_tasks: set[str]) -> None:
        """Called when the user's attention changes.

        Args:
            attending_tasks (set[str]): the new set of tasks that the user is currently attending to.
        """
        for task in attending_tasks:
            self.beliefs[task]["last_attended"] = self.get_cycle_start()
            self._log_info(task, "attending", True)
        if len(attending_tasks) == 0:
            self._log_info("none", "attending", True)

    def decide(self):  # noqa
        # update when the user was last attending to a task
        attending_tasks = self.get_attending_tasks()  # empty or a single task
        if attending_tasks != self._attending_tasks:
            self._attending_tasks = attending_tasks
            self.on_attending(attending_tasks)

        # make guidance decisions
        if self.guidance_on_tasks:
            # guidance is active, should it be?
            guidance_and_acceptable = self.guidance_on_tasks & self.acceptable_tasks
            if guidance_and_acceptable:
                # the task has guidance showing, but is now acceptable - hide guidance for this task
                for task in guidance_and_acceptable:
                    self.hide_guidance(task)
                return

            guidance_and_attending = self.guidance_on_tasks & self._attending_tasks
            if guidance_and_attending:
                # the task has guidance showing, but the user is attending to it - hide guidance for this task
                for task in guidance_and_attending:
                    self.hide_guidance(task)
                return

            # keep showing guidance
        else:
            # guidance is NOT active, should it be?
            # these tasks are candidates for showing guidance
            unacceptable = self.unacceptable_tasks
            # remove those tasks that the user is currently attending
            unacceptable -= self._attending_tasks
            # check the grace period
            unacceptable = self.grace_period_over(unacceptable)

            task = self.break_tie(unacceptable)
            if task:
                # we have selected a task to show guidance on
                self.show_guidance(task)

            # no task meets the criteria, the user is doing well!

    def grace_period_over(self, tasks: set[str]) -> set[str]:
        """Get the set of (unacceptable) tasks that have had their grace period elapse.

        Args:
            tasks (set[str]): the set of tasks to check.

        Returns:
            set[str]: the set of tasks that have had their grace period elapse.
        """
        if self._grace_mode == "guidance_task":
            # how long since guidance was last shown on the task?
            def _grace_met(t):
                tsgs = self.time_since_guidance_start(t)
                # check if grace period has elapsed, or if guidance has never been shown on the task (NaN value)
                return tsgs > self._grace_period or tsgs != tsgs

            return set(t for t in tasks if _grace_met(t))
        elif self._grace_mode == "guidance_any":
            # how long since guidance was last shown on any task?
            tsgs = self.time_since_guidance_start(None)
            if tsgs > self._grace_period or tsgs != tsgs:
                return tasks  # guidance has not been shown recently
            else:
                return set()  # a task has recently had guidance shown, we should not show guidance for any task yet
        elif self._grace_mode == "attention":

            def _grace_met(t):
                tsgs = self.time_since_last_attended(t)
                # check if grace period has elapsed, or if guidance has never been shown on the task (NaN value)
                return tsgs > self._grace_period or tsgs != tsgs

            # print({t: (_grace_met(t), self.time_since_last_attended(t)) for t in tasks})
            # how long since the user was last attending to a task?
            return set(t for t in tasks if _grace_met(t))
        elif self._grace_mode == "failure":
            # how long has this task been in failure?
            return set(
                t
                for t in tasks
                if self.time_since_failure_start(t) > self._grace_period
            )
        else:
            raise ValueError(
                f"Unknown grace mode: {self._grace_mode}, must be one of {DefaultGuidanceAgent.GRACE_ON}"
            )

    def time_since_last_attended(self, task: str) -> float:
        """Get the time since the user was last attending to a task."""
        return self.get_cycle_start() - self.beliefs[task].get(
            "last_attended", float("nan")
        )

    def get_attending_tasks(self) -> set[str]:
        """Get the set of tasks that the user is currently attending to - this is typically only one task.

        Returns:
            set[str]: set of tasks that the user is currently attending to (empty if no tasks are being attended to).
        """
        if self._attention_mode == "fixation":
            result = self.attending_task_fixation()
        elif self._attention_mode == "gaze":
            result = self.attending_task_gaze()
        elif self._attention_mode == "mouse":
            result = self.attending_task_mouse()
        else:
            raise ValueError(f"Unknown attention mode: {self._attention_mode}")
        if result is None:
            return set()
        else:
            return set([result])

    def attending_task_mouse(self) -> str | None:
        """Use mouse position to determine which task the user is attending to."""
        return self.mouse_target

    def attending_task_gaze(self) -> str | None:
        """Use gaze position (including saccades) to determine which task the user is attending to."""
        return self.gaze_target

    def attending_task_fixation(self) -> str | None:
        """Use fixation to determine which task the user is attending to."""
        #print(self.fixation_target)
        return self.fixation_target

    def break_tie(
        self,
        tasks: Iterable[str],
        method: Literal["random", "longest"] = "random",
    ) -> str | None:
        """Break a tie on tasks that all met the criteria for displaying guiance.

        The tie is broken using the `method` argument:
        - "random" will randomly break the tie.
        - "longest" will choose the task that has been in failure for the longest.

        Args:
            tasks (Iterable[str]): tasks to break the tie, one of which will be returned.
            method (Literal["random", "longest"], optional): how to break ties if guidance may be shown on multiple tasks simultaneously. Defaults to "random".

        Returns:
            str: the chosen task, or None if there are no tasks to choose from.
        """
        if len(tasks) == 0:
            return None
        if method == "random":
            # randomly break the tie
            return random.choice(list(tasks))
        elif method == "longest":
            # choose the task longest in failure
            return max(
                [(task, self.time_since_failure_start(task)) for task in tasks],
                key=lambda x: x[1],
            )[0]
        else:
            raise ValueError(
                f"Unknown guidance tie break method: {method}, must be one of {DefaultGuidanceAgent.BREAK_TIES}"
            )

    @observe([EyeMotionEvent, MouseMotionEvent])
    def _on_motion_event(self, event: MouseMotionEvent | EyeMotionEvent):
        """It may be useful to the actuators to get these events. It is a trick to forward sensory input to the agents actuators. The `ArrowGuidanceActuator` is an example that requires this information to display the array at the gaze/mouse position."""
        # TODO we may need to guard against actuators executing these actions...?
        # manually attempt the event, we could specify which actuators need this information...?
        self.attempt(event)

    def on_acceptable(self, task: str):  # noqa
        self._log_info(task, "acceptable", True)
        return super().on_acceptable(task)

    def on_unacceptable(self, task: str):  # noqa
        self._log_info(task, "acceptable", False)
        return super().on_unacceptable(task)

    def _log_info(self, task, z, ok):
        """Log info related to a task to the console."""
        info = "task %20s %20s %s" % (z, task, ["✘", "✔"][int(ok)])
        LOGGER.info(info)

__init__(sensors, actuators, break_ties='random', grace_mode='failure', attention_mode='fixation', grace_period=3.0, counter_factual=False, **kwargs)

Constructor.

Parameters:

Name	Type	Description	Default
sensors	list[Sensor]	list of sensors, this will typically be a list of icua.agent.TaskAcceptabilitySensors. A UserInputSensor will always be added automatically.	required
actuators	list[Actuator]	list of actuators, this will typically contain actuators that are capable of providing visual feedback to a user, see e.g. icua.agent.GuidanceActuator and its concrete implementations. A CounterFactualGuidanceActuator will be added by default.	required
break_ties	Literal['random', 'longest', 'since']	how to break ties if guidance may be shown on multiple tasks simultaneously. Defaults to "random". "random" will randomly break the tie, "longest" will choose the task that has been in failure for the longest, "since" will choose the task that has not had guidance shown for the longest time.	'random'
grace_mode	Literal['guidance_task', 'guidance_any', 'failure', 'attention']	how to track the grace period. "guidance_task" will track the time since guidance was last shown on the task, "guidance_any" will track the time since guidance was last shown on any task, "failure" will track the time since the last failure on the task, "attention" will track the time since the user was last attending to a task. Defaults to "failure".	'failure'
attention_mode	Literal['fixation', 'gaze', 'mouse']	method of determining where the user is attending. "fixation" will use the most recent gaze fixation, "gaze" will use the gaze position (including saccades), "mouse" will use the current mouse position. Defaults to "fixation".	'fixation'
grace_period	float	the time to wait (seconds) before guidance may be shown for a task after the last time guidance was shown on the task. Defaults to 3.0 seconds.	3.0
counter_factual	bool	whether guidance should be shown to the user, or whether it should just be logged. This allows counter-factual experiments to be run, we can track when guidance would have been shown, and compare the when it was actually shown (in a different run). Defaults to False.	False
kwargs	dict[str, Any]	Additional optional keyword arguments.	{}

Source code in matbii\guidance\agent_default.py

def __init__(
    self,
    sensors: list[Sensor],
    actuators: list[Actuator],
    break_ties: Literal["random", "longest"] = "random",
    grace_mode: Literal[
        "guidance_task", "guidance_any", "failure", "attention"
    ] = "failure",
    attention_mode: Literal["fixation", "gaze", "mouse"] = "fixation",
    grace_period: float = 3.0,
    counter_factual: bool = False,
    **kwargs: dict[str, Any],
):
    """Constructor.

    Args:
        sensors (list[Sensor]): list of sensors, this will typically be a list of `icua.agent.TaskAcceptabilitySensor`s. A `UserInputSensor` will always be added automatically.
        actuators (list[Actuator]): list of actuators, this will typically contain actuators that are capable of providing visual feedback to a user, see e.g. `icua.agent.GuidanceActuator` and its concrete implementations. A `CounterFactualGuidanceActuator` will be added by default.
        break_ties (Literal["random", "longest", "since"], optional): how to break ties if guidance may be shown on multiple tasks simultaneously. Defaults to "random". "random" will randomly break the tie, "longest" will choose the task that has been in failure for the longest, "since" will choose the task that has not had guidance shown for the longest time.
        grace_mode (Literal["guidance_task", "guidance_any", "failure", "attention"], optional): how to track the grace period. "guidance_task" will track the time since guidance was last shown on the task, "guidance_any" will track the time since guidance was last shown on any task, "failure" will track the time since the last failure on the task, "attention" will track the time since the user was last attending to a task. Defaults to "failure".
        attention_mode (Literal["fixation", "gaze", "mouse"], optional): method of determining where the user is attending. "fixation" will use the most recent gaze fixation, "gaze" will use the gaze position (including saccades), "mouse" will use the current mouse position. Defaults to "fixation".
        grace_period (float, optional): the time to wait (seconds) before guidance may be shown for a task after the last time guidance was shown on the task. Defaults to 3.0 seconds.
        counter_factual (bool, optional): whether guidance should be shown to the user, or whether it should just be logged.  This allows counter-factual experiments to be run, we can track when guidance would have been shown, and compare the when it was actually shown (in a different run). Defaults to False.
        kwargs (dict[str,Any]): Additional optional keyword arguments.
    """
    super().__init__(
        list(filter(None, sensors)),
        list(filter(None, actuators)),
        counter_factual=counter_factual,
        **kwargs,
    )
    self._attention_mode = attention_mode
    if self._attention_mode not in DefaultGuidanceAgent.ATTENTION_MODES:
        raise ValueError(
            f"Invalid argument: `attention_mode` {self._attention_mode} must be one of {DefaultGuidanceAgent.ATTENTION_MODES}"
        )

    self._grace_mode = grace_mode
    if self._grace_mode not in DefaultGuidanceAgent.GRACE_ON:
        raise ValueError(
            f"Invalid argument: `grace_mode` {self._grace_mode} must be one of {DefaultGuidanceAgent.GRACE_ON}"
        )

    self._break_ties = break_ties
    if self._break_ties not in DefaultGuidanceAgent.BREAK_TIES:
        raise ValueError(
            f"Invalid argument: `break_ties` {self._break_ties} must be one of {DefaultGuidanceAgent.BREAK_TIES}"
        )

    self._grace_period = grace_period  # TODO
    if self._grace_period < 0.0:
        raise ValueError(
            f"Invalid argument: `grace_period` {self._grace_period} must be > 0 "
        )

    # used to track the tasks that the user is currently attending to
    self._attending_tasks = set()

attending_task_fixation()

Use fixation to determine which task the user is attending to.

Source code in matbii\guidance\agent_default.py

def attending_task_fixation(self) -> str | None:
    """Use fixation to determine which task the user is attending to."""
    #print(self.fixation_target)
    return self.fixation_target

attending_task_gaze()

Use gaze position (including saccades) to determine which task the user is attending to.

Source code in matbii\guidance\agent_default.py

def attending_task_gaze(self) -> str | None:
    """Use gaze position (including saccades) to determine which task the user is attending to."""
    return self.gaze_target

attending_task_mouse()

Use mouse position to determine which task the user is attending to.

Source code in matbii\guidance\agent_default.py

def attending_task_mouse(self) -> str | None:
    """Use mouse position to determine which task the user is attending to."""
    return self.mouse_target

break_tie(tasks, method='random')

Break a tie on tasks that all met the criteria for displaying guiance.

The tie is broken using the method argument: - "random" will randomly break the tie. - "longest" will choose the task that has been in failure for the longest.

Parameters:

Name	Type	Description	Default
tasks	Iterable[str]	tasks to break the tie, one of which will be returned.	required
method	Literal['random', 'longest']	how to break ties if guidance may be shown on multiple tasks simultaneously. Defaults to "random".	'random'

Returns:

Name	Type	Description
str	str | None	the chosen task, or None if there are no tasks to choose from.

Source code in matbii\guidance\agent_default.py

def break_tie(
    self,
    tasks: Iterable[str],
    method: Literal["random", "longest"] = "random",
) -> str | None:
    """Break a tie on tasks that all met the criteria for displaying guiance.

    The tie is broken using the `method` argument:
    - "random" will randomly break the tie.
    - "longest" will choose the task that has been in failure for the longest.

    Args:
        tasks (Iterable[str]): tasks to break the tie, one of which will be returned.
        method (Literal["random", "longest"], optional): how to break ties if guidance may be shown on multiple tasks simultaneously. Defaults to "random".

    Returns:
        str: the chosen task, or None if there are no tasks to choose from.
    """
    if len(tasks) == 0:
        return None
    if method == "random":
        # randomly break the tie
        return random.choice(list(tasks))
    elif method == "longest":
        # choose the task longest in failure
        return max(
            [(task, self.time_since_failure_start(task)) for task in tasks],
            key=lambda x: x[1],
        )[0]
    else:
        raise ValueError(
            f"Unknown guidance tie break method: {method}, must be one of {DefaultGuidanceAgent.BREAK_TIES}"
        )

get_attending_tasks()

Get the set of tasks that the user is currently attending to - this is typically only one task.

Returns:

Type	Description
set[str]	set[str]: set of tasks that the user is currently attending to (empty if no tasks are being attended to).

Source code in matbii\guidance\agent_default.py

def get_attending_tasks(self) -> set[str]:
    """Get the set of tasks that the user is currently attending to - this is typically only one task.

    Returns:
        set[str]: set of tasks that the user is currently attending to (empty if no tasks are being attended to).
    """
    if self._attention_mode == "fixation":
        result = self.attending_task_fixation()
    elif self._attention_mode == "gaze":
        result = self.attending_task_gaze()
    elif self._attention_mode == "mouse":
        result = self.attending_task_mouse()
    else:
        raise ValueError(f"Unknown attention mode: {self._attention_mode}")
    if result is None:
        return set()
    else:
        return set([result])

grace_period_over(tasks)

Get the set of (unacceptable) tasks that have had their grace period elapse.

Parameters:

Name	Type	Description	Default
tasks	set[str]	the set of tasks to check.	required

Returns:

Type	Description
set[str]	set[str]: the set of tasks that have had their grace period elapse.

Source code in matbii\guidance\agent_default.py

def grace_period_over(self, tasks: set[str]) -> set[str]:
    """Get the set of (unacceptable) tasks that have had their grace period elapse.

    Args:
        tasks (set[str]): the set of tasks to check.

    Returns:
        set[str]: the set of tasks that have had their grace period elapse.
    """
    if self._grace_mode == "guidance_task":
        # how long since guidance was last shown on the task?
        def _grace_met(t):
            tsgs = self.time_since_guidance_start(t)
            # check if grace period has elapsed, or if guidance has never been shown on the task (NaN value)
            return tsgs > self._grace_period or tsgs != tsgs

        return set(t for t in tasks if _grace_met(t))
    elif self._grace_mode == "guidance_any":
        # how long since guidance was last shown on any task?
        tsgs = self.time_since_guidance_start(None)
        if tsgs > self._grace_period or tsgs != tsgs:
            return tasks  # guidance has not been shown recently
        else:
            return set()  # a task has recently had guidance shown, we should not show guidance for any task yet
    elif self._grace_mode == "attention":

        def _grace_met(t):
            tsgs = self.time_since_last_attended(t)
            # check if grace period has elapsed, or if guidance has never been shown on the task (NaN value)
            return tsgs > self._grace_period or tsgs != tsgs

        # print({t: (_grace_met(t), self.time_since_last_attended(t)) for t in tasks})
        # how long since the user was last attending to a task?
        return set(t for t in tasks if _grace_met(t))
    elif self._grace_mode == "failure":
        # how long has this task been in failure?
        return set(
            t
            for t in tasks
            if self.time_since_failure_start(t) > self._grace_period
        )
    else:
        raise ValueError(
            f"Unknown grace mode: {self._grace_mode}, must be one of {DefaultGuidanceAgent.GRACE_ON}"
        )

on_attending(attending_tasks)

Called when the user's attention changes.

Parameters:

Name	Type	Description	Default
attending_tasks	set[str]	the new set of tasks that the user is currently attending to.	required

Source code in matbii\guidance\agent_default.py

def on_attending(self, attending_tasks: set[str]) -> None:
    """Called when the user's attention changes.

    Args:
        attending_tasks (set[str]): the new set of tasks that the user is currently attending to.
    """
    for task in attending_tasks:
        self.beliefs[task]["last_attended"] = self.get_cycle_start()
        self._log_info(task, "attending", True)
    if len(attending_tasks) == 0:
        self._log_info("none", "attending", True)

time_since_last_attended(task)

Get the time since the user was last attending to a task.

Source code in matbii\guidance\agent_default.py

def time_since_last_attended(self, task: str) -> float:
    """Get the time since the user was last attending to a task."""
    return self.get_cycle_start() - self.beliefs[task].get(
        "last_attended", float("nan")
    )