Get task

Functions for extracting task events from an event log file.

get_resource_management_task_events(parser, events, norm=np.inf)

Extracts useful data for the resource management task from the event log.

Columns

• timestamp: float - the (logging) timestamp of the event
• frame: int - the frame number of the event, events with a frame number of 0 happen BEFORE the first frame is rendered to the user.
• user: bool - whether the event was triggered by the user or not.
• tank-a: float - tank a level
• tank-b: float - tank b level
• tank-c: float - tank c level
• tank-d: float - tank d level
• tank-e: float - tank e level
• tank-f: float - tank f level

Parameters:

Name	Type	Description	Default
parser	EventLogParser	parser used to parse the event log file.	required
events	list[tuple[float, Event]]	list of events that were parsed from the event log file.	required
norm	float | int	the norm to use for the distance metric, either "inf" for the max norm or an integer for the p-norm.	inf

Returns:

Type	Description
DataFrame	pd.DataFrame: dataframe with columns: ["timestamp", "frame", "user", *"tanks-{i}", *"pumps-{ij}"]

Source code in matbii\extras\analysis\get_task.py

def get_resource_management_task_events(
    parser: EventLogParser,
    events: list[tuple[float, Event]],
    norm: float | int = np.inf,
) -> pd.DataFrame:
    """Extracts useful data for the resource management task from the event log.

    Columns:
        - timestamp: float - the (logging) timestamp of the event
        - frame: int - the frame number of the event, events with a frame number of 0 happen BEFORE the first frame is rendered to the user.
        - user: bool - whether the event was triggered by the user or not.
        - tank-a: float - tank a level
        - tank-b: float - tank b level
        - tank-c: float - tank c level
        - tank-d: float - tank d level
        - tank-e: float - tank e level
        - tank-f: float - tank f level

    Args:
        parser (EventLogParser): parser used to parse the event log file.
        events (list[tuple[float, Event]]): list of events that were parsed from the event log file.
        norm (float | int, optional): the norm to use for the distance metric, either "inf" for the max norm or an integer for the p-norm.

    Returns:
        pd.DataFrame: dataframe with columns: ["timestamp", "frame", "user", *"tanks-{i}", *"pumps-{ij}"]
    """
    fevents = parser.filter_events(
        events,
        (
            Insert,
            Delete,
            Update,
            Replace,
            SetPumpAction,
            BurnFuelAction,
            PumpFuelAction,
            TogglePumpAction,
            TogglePumpFailureAction,
            KeyEvent,
            MouseButtonEvent,
            RenderEvent,
        ),
    )

    # # sort the events by their log timestamp
    fevents = parser.sort_by_timestamp(fevents)

    def _sense(state: XMLState):
        # sense_actions = ResourceManagementTaskAcceptabilitySensor().sense()
        # sense actions for tank states
        sense_actions = [
            Select(xpath=f"//*[@id='{id}']", attrs=["id", "data-level"])
            for id in tank_ids()
        ]
        result = sense(state, sense_actions)
        if result is None:
            result = dict()
        result_tanks = {k: v["data-level"] for k, v in result.items()}

        # actions for pump states
        sense_actions = [
            Select(xpath=f"//*[@id='{id}-button']", attrs=["id", "data-state"])
            for id in pump_ids()
        ]
        result = sense(state, sense_actions)
        # print(result)
        if result is None:
            result = dict()
        result_pumps = {k.rsplit("-", 1)[0]: v["data-state"] for k, v in result.items()}
        if len(result_pumps) == 0:
            return None
        return {**result_tanks, **result_pumps}

    df = _get_task_dataframe(fevents, _sense)
    if df is None:
        columns = [
            "timestamp",
            "frame",
            "user",
            *tank_ids(),
            *pump_ids(),
        ]
        return pd.DataFrame(columns=columns)

    # TODO combine rows if the timestamp matches
    return df

get_system_monitoring_task_events(parser, events)

Extracts useful data for the system monitoring task from the event log.

Columns

• timestamp: float - the (logging) timestamp of the event
• frame: int - the frame number of the event, events with a frame number of 0 happen BEFORE the first frame is rendered to the user.
• user: bool - whether the event was triggered by the user (True) or not (False).
• light-1: int - the state of light-1
• light-2: int - the state of light-2
• slider-1: int - the state of slider-1
• slider-2: int - the state of slider-2
• slider-3: int - the state of slider-3
• slider-4: int - the state of slider-4

NOTE: the number of sliders/lights is assumed to be 4/2 respectively, any alternative settings will not work here - this may be fixed in future versions.

Parameters:

Name	Type	Description	Default
parser	EventLogParser	parser used to parse the event log file.	required
events	list[tuple[float, Event]]	list of events that were parsed from the event log file.	required

Returns:

Type	Description
DataFrame	pd.DataFrame: dataframe with columns: ["timestamp", "frame", "user", "light-1", "light-2", "slider-1", "slider-2", "slider-3", "slider-4"]

Source code in matbii\extras\analysis\get_task.py

def get_system_monitoring_task_events(
    parser: EventLogParser,
    events: list[tuple[float, Event]],
) -> pd.DataFrame:
    """Extracts useful data for the system monitoring task from the event log.

    Columns:
        - timestamp: float - the (logging) timestamp of the event
        - frame: int - the frame number of the event, events with a frame number of 0 happen BEFORE the first frame is rendered to the user.
        - user: bool - whether the event was triggered by the user (True) or not (False).
        - light-1: int - the state of light-1
        - light-2: int - the state of light-2
        - slider-1: int - the state of slider-1
        - slider-2: int - the state of slider-2
        - slider-3: int - the state of slider-3
        - slider-4: int - the state of slider-4

    NOTE: the number of sliders/lights is assumed to be 4/2 respectively, any alternative settings will not work here - this may be fixed in future versions.

    Args:
        parser (EventLogParser): parser used to parse the event log file.
        events (list[tuple[float, Event]]): list of events that were parsed from the event log file.

    Returns:
        pd.DataFrame: dataframe with columns: ["timestamp", "frame", "user", "light-1", "light-2", "slider-1", "slider-2", "slider-3", "slider-4"]
    """
    # these are raw events that may be triggered internally by matbii
    # insert events insert the tasks into the xml state at the start of the simulation
    fevents = parser.filter_events(
        events,
        (
            Insert,
            Delete,
            Replace,
            Update,
            SetLightAction,
            ToggleLightAction,
            SetSliderAction,
            KeyEvent,
            MouseButtonEvent,
            RenderEvent,
        ),
    )
    # sort the events by their timestamp
    fevents = parser.sort_by_timestamp(fevents)

    def _rename_column(k: str):
        return "-".join(k.split("-")[:2])

    def _sense(state: XMLState):
        # sense actions to get relevant data from the state
        sense_actions = SystemMonitoringTaskAcceptabilitySensor().sense()
        result = sense(state, sense_actions)
        if result is None:
            return None
        return {
            _rename_column(k): v["data-state"]
            for k, v in result.items()
            if "data-state" in v
        }

    df = _get_task_dataframe(fevents, _sense)
    if df is None:
        columns = [
            "timestamp",
            "frame",
            "user",
            "light-1",
            "light-2",
            "slider-1",
            "slider-2",
            "slider-3",
            "slider-4",
        ]
        return pd.DataFrame(columns=columns)
    return df

get_tracking_task_events(parser, events, norm=np.inf)

Extracts useful data for the tracking task from the event log.

Columns

• timestamp: float - the (logging) timestamp of the event
• frame: int - the frame number of the event, events with a frame number of 0 happen BEFORE the first frame is rendered to the user.
• user: bool - whether the event was triggered by the user or not.
• x: float - the x coordinate of the tracking target
• y: float - the y coordinate of the tracking target
• distance: float - the distance to the center of the task (according to the given norm).

Parameters:

Name	Type	Description	Default
parser	EventLogParser	parser used to parse the event log file.	required
events	list[tuple[float, Event]]	list of events that were parsed from the event log file.	required
norm	float | int	the norm to use for the distance metric, either "inf" for the max norm or an integer for the p-norm.	inf

Returns:

Type	Description
DataFrame	pd.DataFrame: dataframe with columns: ["timestamp", "frame", "user", "x", "y", "distance"]

Source code in matbii\extras\analysis\get_task.py

def get_tracking_task_events(
    parser: EventLogParser,
    events: list[tuple[float, Event]],
    norm: float | int = np.inf,
) -> pd.DataFrame:
    """Extracts useful data for the tracking task from the event log.

    Columns:
        - timestamp: float - the (logging) timestamp of the event
        - frame: int - the frame number of the event, events with a frame number of 0 happen BEFORE the first frame is rendered to the user.
        - user: bool - whether the event was triggered by the user or not.
        - x: float - the x coordinate of the tracking target
        - y: float - the y coordinate of the tracking target
        - distance: float - the distance to the center of the task (according to the given `norm`).

    Args:
        parser (EventLogParser): parser used to parse the event log file.
        events (list[tuple[float, Event]]): list of events that were parsed from the event log file.
        norm (float | int, optional): the norm to use for the distance metric, either "inf" for the max norm or an integer for the p-norm.

    Returns:
        pd.DataFrame: dataframe with columns: ["timestamp", "frame", "user", "x", "y", "distance"]
    """
    fn_norm = partial(np.linalg.norm, ord=norm)
    # collect all the relevant events for this task
    fevents = parser.filter_events(
        events,
        (
            Insert,
            Delete,
            Update,
            Replace,
            TargetMoveAction,
            KeyEvent,
            MouseButtonEvent,
            RenderEvent,
        ),
    )

    # # sort the events by their log timestamp
    fevents = parser.sort_by_timestamp(fevents)

    def _sense(state: XMLState):
        sense_actions = TrackingTaskAcceptabilitySensor().sense()
        result = sense(state, sense_actions)
        if result is None:
            return None
        # TODO compute distance to the center of the task
        bw, bh = (
            result[TrackingTaskAcceptabilitySensor._BOX_ID]["width"],
            result[TrackingTaskAcceptabilitySensor._BOX_ID]["height"],
        )
        bx, by = (
            result[TrackingTaskAcceptabilitySensor._BOX_ID]["x"] + bw / 2,
            result[TrackingTaskAcceptabilitySensor._BOX_ID]["y"] + bh / 2,
        )
        tw, th = (
            result[TrackingTaskAcceptabilitySensor._TARGET_ID]["width"],
            result[TrackingTaskAcceptabilitySensor._TARGET_ID]["height"],
        )
        tx, ty = (
            result[TrackingTaskAcceptabilitySensor._TARGET_ID]["x"] + tw / 2,
            result[TrackingTaskAcceptabilitySensor._TARGET_ID]["y"] + th / 2,
        )
        return dict(
            x=tx,
            y=ty,
            distance=fn_norm((tx - bx, ty - by)),
        )

    df = _get_task_dataframe(fevents, _sense)
    if df is None:
        columns = [
            "timestamp",
            "frame",
            "user",
            "x",
            "y",
            "distance",
        ]
        return pd.DataFrame(columns=columns)
    return df

sense(state, sense_actions)

Sense data from the state using the provided sense actions.

Source code in matbii\extras\analysis\get_task.py

def sense(state: XMLState, sense_actions: list[Select]):
    """Sense data from the state using the provided sense actions."""

    def _sense(state: XMLState, sense_actions: list[Select]):
        for action in sense_actions:
            yield action.__execute__(state)

    data = dict()
    try:
        for obs in _sense(state, sense_actions):
            for value in obs:
                id = value.pop("id")
                data[id] = value
    except Exception:
        return None
    return data