Filters

class `BaseFilter`

class BaseFilter(BaseModel, ViewOperation):

A step that only drops events and objects, never changes them.

A filter narrows the log with :meth:Step.keep and leaves cleaning up to the one :class:Clean :func:filter_view puts after the last filter. So the log a filter reads need not be clean: a relation can point at an event or object an earlier filter dropped, and an entity can be in no relation. Join a relation to the events and objects to read only those still there.

Source

class BaseFilter(BaseModel, ViewOperation):
    """A step that only drops events and objects, never changes them.

    A filter narrows the log with :meth:`Step.keep` and leaves cleaning up to
    the one :class:`Clean` :func:`filter_view` puts after the last filter. So
    the log a filter reads need not be clean: a relation can point at an event
    or object an earlier filter dropped, and an entity can be in no relation.
    Join a relation to the events and objects to read only those still there.
    """

    @abstractmethod
    def build(self, step: Step) -> None:
        """Narrow ``step``'s events and/or objects with :meth:`Step.keep`."""

function `build`

def build(step: Step) -> None:

Narrow step’s events and/or objects with :meth:Step.keep.

Source

    @abstractmethod
    def build(self, step: Step) -> None:
        """Narrow ``step``'s events and/or objects with :meth:`Step.keep`."""

class `E2OCountFilter`

class E2OCountFilter(BaseFilter, RelationCountFilterConfig):

Keep events (or objects) by how many E2O relations of a given kind they have.

Counts the relations between activity source and object type target (optionally of one qualifier); direction picks whether the events (source) or the objects (target) are the ones counted.

Source

class E2OCountFilter(BaseFilter, RelationCountFilterConfig):
    """Keep events (or objects) by how many E2O relations of a given kind they have.

    Counts the relations between activity ``source`` and object type ``target``
    (optionally of one ``qualifier``); ``direction`` picks whether the events
    (source) or the objects (target) are the ones counted.
    """

    direction: Literal["source", "target"] = "source"

    def build(self, step: Step) -> None:
        eid, oid = ident(EID_COL), ident(OID_COL)
        counted_col = eid if self.direction == "source" else oid
        counted = f"""
            SELECT r.{counted_col} AS id FROM {step[E2O_TABLE]} r
            JOIN {step[EVENTS_TABLE]} e USING ({eid})
            JOIN {step[OBJECTS_TABLE]} o USING ({oid})
            WHERE e.{ident(ACTIVITY_COL)} = {literal(self.source)}
                AND o.{ident(OTYPE_COL)} = {literal(self.target)}
                {self._qualified(f"r.{ident(E2O_QUALIFIER)}")}
        """

        if self.direction == "source":
            step.keep(
                EVENTS_TABLE,
                _keep_counted(
                    step[EVENTS_TABLE],
                    EID_COL,
                    ACTIVITY_COL,
                    self.source,
                    counted,
                    self.range,
                ),
            )
        else:
            step.keep(
                OBJECTS_TABLE,
                _keep_counted(
                    step[OBJECTS_TABLE],
                    OID_COL,
                    OTYPE_COL,
                    self.target,
                    counted,
                    self.range,
                ),
            )

class `EventAttributeFilter`

class EventAttributeFilter(_AttributeFilter):

Keep the events whose attribute matches; activities without it are untouched.

Source

class EventAttributeFilter(_AttributeFilter):
    """Keep the events whose attribute matches; activities without it are untouched."""

    def build(self, step: Step) -> None:
        if self.attribute not in _attribute_names(step, EVENTS_TABLE):
            return

        step.keep(
            EVENTS_TABLE,
            self._keep(
                f"SELECT {ident(EID_COL)} AS id, {ident(ACTIVITY_COL)} AS type, "
                f"{ident(self.attribute)} AS value FROM {step[EVENTS_TABLE]}",
                EID_COL,
            ),
        )

class `EventTypeFilter`

class EventTypeFilter(BaseFilter):

Keep the events of the given activities.

Source

class EventTypeFilter(BaseFilter):
    """Keep the events of the given activities."""

    event_types: Annotated[
        list[str], Field(json_schema_extra={"fieldType": "event_type"})
    ]
    mode: Literal["exclude", "include"] = "exclude"

    def build(self, step: Step) -> None:
        step.keep(
            EVENTS_TABLE,
            f"SELECT {ident(EID_COL)} FROM {step[EVENTS_TABLE]} "
            f"WHERE {_selected(ACTIVITY_COL, self.event_types, self.mode)}",
        )

class `EventTypeFrequencyFilter`

class EventTypeFrequencyFilter(BaseFilter):

Keep the events of activities with at least threshold of all events.

Shares are counted over the events the log keeps once cleaned.

Source

class EventTypeFrequencyFilter(BaseFilter):
    """Keep the events of activities with at least ``threshold`` of all events.

    Shares are counted over the events the log keeps once cleaned.
    """

    mode: Literal["include", "exclude"] = "include"
    threshold: float = Field(default=0.0, ge=0.0, le=1.0)

    def build(self, step: Step) -> None:
        step.keep(
            EVENTS_TABLE,
            _frequent(
                step[EVENTS_TABLE],
                EID_COL,
                ACTIVITY_COL,
                surviving_ids(step.tables)[0],
                self.threshold,
                self.mode,
            ),
        )

class `O2OCountFilter`

class O2OCountFilter(BaseFilter, RelationCountFilterConfig):

Keep objects by how many O2O relations of a given kind they have.

Counts the relations between object types source and target (optionally of one qualifier); direction picks whether the source or the target object is the one counted.

Source

class O2OCountFilter(BaseFilter, RelationCountFilterConfig):
    """Keep objects by how many O2O relations of a given kind they have.

    Counts the relations between object types ``source`` and ``target`` (optionally
    of one ``qualifier``); ``direction`` picks whether the source or the target
    object is the one counted.
    """

    direction: Literal["source", "target"] = "source"

    def build(self, step: Step) -> None:
        oid, otype = ident(OID_COL), ident(OTYPE_COL)
        is_source = self.direction == "source"
        counted_col = ident(O2O_SOURCE_ID if is_source else O2O_TARGET_ID)
        counted = f"""
            SELECT r.{counted_col} AS id FROM {step[O2O_TABLE]} r
            JOIN {step[OBJECTS_TABLE]} s ON s.{oid} = r.{ident(O2O_SOURCE_ID)}
            JOIN {step[OBJECTS_TABLE]} t ON t.{oid} = r.{ident(O2O_TARGET_ID)}
            WHERE s.{otype} = {literal(self.source)}
                AND t.{otype} = {literal(self.target)}
                {self._qualified(f"r.{ident(O2O_QUALIFIER)}")}
        """

        step.keep(
            OBJECTS_TABLE,
            _keep_counted(
                step[OBJECTS_TABLE],
                OID_COL,
                OTYPE_COL,
                self.source if is_source else self.target,
                counted,
                self.range,
            ),
        )

class `ObjectAttributeFilter`

class ObjectAttributeFilter(_AttributeFilter):

Keep the objects whose attribute matches; types without it are untouched.

An object attribute is read from its change rows, which is where every value it ever holds is stored. A dynamic attribute therefore has several, and the earliest is the one filtered on — the object’s value as it started out.

Source

class ObjectAttributeFilter(_AttributeFilter):
    """Keep the objects whose attribute matches; types without it are untouched.

    An object attribute is read from its change rows, which is where every value
    it ever holds is stored. A dynamic attribute therefore has several, and the
    earliest is the one filtered on -- the object's value as it started out.
    """

    def build(self, step: Step) -> None:
        if self.attribute not in _attribute_names(step, OBJECT_CHANGES_TABLE):
            return

        oid, ts = ident(OID_COL), ident(TIMESTAMP_COL)
        step.keep(
            OBJECTS_TABLE,
            self._keep(
                f"SELECT o.{oid} AS id, o.{ident(OTYPE_COL)} AS type, earliest.value "
                f"FROM {step[OBJECTS_TABLE]} o LEFT JOIN ("
                f"SELECT {oid}, arg_min({ident(self.attribute)}, {ts}) AS value "
                f"FROM {step[OBJECT_CHANGES_TABLE]} "
                f"WHERE {ident(OBJECT_CHANGED_FIELD)} = {literal(self.attribute)} "
                f"GROUP BY {oid}"
                f") earliest ON earliest.{oid} = o.{oid}",
                OID_COL,
            ),
        )

class `ObjectIdFilter`

class ObjectIdFilter(BaseFilter):

Keep the named objects.

Source

class ObjectIdFilter(BaseFilter):
    """Keep the named objects."""

    object_ids: list[str]
    mode: Literal["exclude", "include"] = "include"

    def build(self, step: Step) -> None:
        step.keep(
            OBJECTS_TABLE,
            f"SELECT {ident(OID_COL)} FROM {step[OBJECTS_TABLE]} "
            f"WHERE {_selected(OID_COL, self.object_ids, self.mode)}",
        )

class `ObjectTypeFilter`

class ObjectTypeFilter(BaseFilter):

Keep the objects of the given types.

Source

class ObjectTypeFilter(BaseFilter):
    """Keep the objects of the given types."""

    object_types: Annotated[
        list[str], Field(json_schema_extra={"fieldType": "object_type"})
    ]
    mode: Literal["exclude", "include"] = "exclude"

    def build(self, step: Step) -> None:
        step.keep(
            OBJECTS_TABLE,
            f"SELECT {ident(OID_COL)} FROM {step[OBJECTS_TABLE]} "
            f"WHERE {_selected(OTYPE_COL, self.object_types, self.mode)}",
        )

class `ObjectTypeFrequencyFilter`

class ObjectTypeFrequencyFilter(BaseFilter):

Keep the objects of types with at least threshold of all objects.

Shares are counted over the objects the log keeps once cleaned.

Source

class ObjectTypeFrequencyFilter(BaseFilter):
    """Keep the objects of types with at least ``threshold`` of all objects.

    Shares are counted over the objects the log keeps once cleaned.
    """

    mode: Literal["include", "exclude"] = "include"
    threshold: float = Field(default=0.0, ge=0.0, le=1.0)

    def build(self, step: Step) -> None:
        step.keep(
            OBJECTS_TABLE,
            _frequent(
                step[OBJECTS_TABLE],
                OID_COL,
                OTYPE_COL,
                surviving_ids(step.tables)[1],
                self.threshold,
                self.mode,
            ),
        )

class `TimeFrameFilter`

class TimeFrameFilter(BaseFilter):

Keep the events within a time range. Either end may be left open.

Source

class TimeFrameFilter(BaseFilter):
    """Keep the events within a time range. Either end may be left open."""

    time_range: tuple[str | None, str | None]
    mode: Literal["exclude", "include"] = "include"

    def build(self, step: Step) -> None:
        timestamp = ident(TIMESTAMP_COL)
        conditions = [
            f"{timestamp} {comparison} {timestamp_literal(bound)}"
            for bound, comparison in zip(
                (utc_bound(bound) for bound in self.time_range),
                (">=", "<="),
                strict=True,
            )
            if bound is not None
        ]

        within = " AND ".join(conditions) or "true"
        if self.mode == "exclude":
            within = f"NOT ({within})"

        step.keep(
            EVENTS_TABLE,
            f"SELECT {ident(EID_COL)} FROM {step[EVENTS_TABLE]} WHERE {within}",
        )

function `apply_filters`

def apply_filters(ocel: OCEL, filters: Sequence[BaseFilter]) -> OCEL:

Return a new in-memory :class:OCEL holding what filters leave.

Source

def apply_filters(ocel: OCEL, filters: Sequence[BaseFilter]) -> OCEL:
    """Return a new in-memory :class:`OCEL` holding what ``filters`` leave."""
    with filter_view(ocel, filters) as view:
        return view.to_ocel()

function `filter_view`

def filter_view(ocel: OCEL, filters: Sequence[BaseFilter]) -> OCELView:

ocel as filters leave it, in order, as a view the caller closes.

Each filter reads the log the ones before it left, uncleaned; one :class:Clean after the last makes the result a valid log.

Source

def filter_view(ocel: OCEL, filters: Sequence[BaseFilter]) -> OCELView:
    """``ocel`` as ``filters`` leave it, in order, as a view the caller closes.

    Each filter reads the log the ones before it left, uncleaned; one
    :class:`Clean` after the last makes the result a valid log.
    """
    view = OCELView(ocel)
    try:
        view.apply(*filters, Clean())
    except Exception:
        view.close()
        raise
    return view

Filters

class BaseFilter

function build

class E2OCountFilter

class EventAttributeFilter

class EventTypeFilter

class EventTypeFrequencyFilter

class O2OCountFilter

class ObjectAttributeFilter

class ObjectIdFilter

class ObjectTypeFilter

class ObjectTypeFrequencyFilter

class TimeFrameFilter

function apply_filters

function filter_view

class `BaseFilter`

function `build`

class `E2OCountFilter`

class `EventAttributeFilter`

class `EventTypeFilter`

class `EventTypeFrequencyFilter`

class `O2OCountFilter`

class `ObjectAttributeFilter`

class `ObjectIdFilter`

class `ObjectTypeFilter`

class `ObjectTypeFrequencyFilter`

class `TimeFrameFilter`

function `apply_filters`

function `filter_view`