Filters

class BaseFilter

class BaseFilter(BaseModel, ABC):

A valid subset of an OCEL, expressed as the event/object ids to keep.

Source

class BaseFilter(BaseModel, ABC):
    """A valid subset of an OCEL, expressed as the event/object ids to keep."""

    @abstractmethod
    def keep(self, ocel: "OCEL") -> Keep:
        """Return the event/object ids to keep as a :class:`Keep`."""
        ...

function keep

def keep(ocel: 'OCEL') -> Keep:

Return the event/object ids to keep as a :class:Keep.

Source

    @abstractmethod
    def keep(self, ocel: "OCEL") -> Keep:
        """Return the event/object ids to keep as a :class:`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 keep(self, ocel) -> Keep:
        # The E2O table already carries the activity and the object type.
        matched = ocel.e2o.pl.filter(
            (pl.col(ACTIVITY_COL) == self.source) & (pl.col(OTYPE_COL) == self.target)
        )
        if self.qualifier is not None:
            matched = matched.filter(pl.col(E2O_QUALIFIER) == self.qualifier)

        if self.direction == "source":
            return Keep(
                events=_keep_counted(
                    entities=lambda: ocel.events.pl,
                    counts=matched.group_by(EID_COL).agg(pl.len().alias(_COUNT)),
                    matched=pl.col(ACTIVITY_COL) == self.source,
                    id_col=EID_COL,
                    count_range=self.range,
                )
            )

        return Keep(
            objects=_keep_counted(
                entities=lambda: ocel.objects.pl,
                counts=matched.group_by(OID_COL).agg(pl.len().alias(_COUNT)),
                matched=pl.col(OTYPE_COL) == self.target,
                id_col=OID_COL,
                count_range=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 keep(self, ocel) -> Keep:
        if self.attribute not in ocel.events.attribute_names:
            return Keep()

        return Keep(
            events=self._keep(
                ocel,
                f"SELECT {ident(EID_COL)} AS id, {ident(ACTIVITY_COL)} AS type, "
                f"{ident(self.attribute)} AS value FROM {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 keep(self, ocel) -> Keep:
        events = ocel.events.pl.filter(_selected(ACTIVITY_COL, self.event_types, self.mode))
        return Keep(events=events.select(EID_COL))

class EventTypeFrequencyFilter

class EventTypeFrequencyFilter(BaseFilter):

Keep the events of the most common activities, by cumulative frequency.

Source

class EventTypeFrequencyFilter(BaseFilter):
    """Keep the events of the most common activities, by cumulative frequency."""

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

    def keep(self, ocel) -> Keep:
        return Keep(
            events=ocel.events.pl.filter(
                _frequent(lambda: ocel.events.pl, ACTIVITY_COL, self.threshold, self.mode)
            ).select(EID_COL)
        )

class Keep

class Keep(NamedTuple):

The ids a filter keeps.

events / objects are single-column lazy frames of the surviving event / object ids. None on a side means “keep all” of that entity, which is not the same as an empty frame — that keeps none.

Source

class Keep(NamedTuple):
    """The ids a filter keeps.

    ``events`` / ``objects`` are single-column lazy frames of the surviving event /
    object ids. ``None`` on a side means "keep all" of that entity, which is not
    the same as an empty frame -- that keeps none.
    """

    events: pl.LazyFrame | None = None
    objects: pl.LazyFrame | None = None

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 keep(self, ocel) -> Keep:
        matched = ocel.o2o.typed_pl.filter(
            (pl.col(O2O_SOURCE_TYPE) == self.source) & (pl.col(O2O_TARGET_TYPE) == self.target)
        )
        if self.qualifier is not None:
            matched = matched.filter(pl.col(O2O_QUALIFIER) == self.qualifier)

        is_source = self.direction == "source"
        key_col = O2O_SOURCE_ID if is_source else O2O_TARGET_ID
        counted_type = self.source if is_source else self.target

        return Keep(
            objects=_keep_counted(
                entities=lambda: ocel.objects.pl,
                counts=matched.group_by(key_col)
                .agg(pl.len().alias(_COUNT))
                .rename({key_col: OID_COL}),
                matched=pl.col(OTYPE_COL) == counted_type,
                id_col=OID_COL,
                count_range=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 keep(self, ocel) -> Keep:
        if self.attribute not in ocel.objects.attribute_names:
            return Keep()

        oid, ts = ident(OID_COL), ident(TIMESTAMP_COL)
        return Keep(
            objects=self._keep(
                ocel,
                f"SELECT o.{oid} AS id, o.{ident(OTYPE_COL)} AS type, earliest.value "
                f"FROM {OBJECTS_TABLE} o LEFT JOIN ("
                f"SELECT {oid}, arg_min({ident(self.attribute)}, {ts}) AS value "
                f"FROM {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 keep(self, ocel) -> Keep:
        objects = ocel.objects.pl.filter(_selected(OID_COL, self.object_ids, self.mode))
        return Keep(objects=objects.select(OID_COL))

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 keep(self, ocel) -> Keep:
        objects = ocel.objects.pl.filter(_selected(OTYPE_COL, self.object_types, self.mode))
        return Keep(objects=objects.select(OID_COL))

class ObjectTypeFrequencyFilter

class ObjectTypeFrequencyFilter(BaseFilter):

Keep the objects of the most common types, by cumulative frequency.

Source

class ObjectTypeFrequencyFilter(BaseFilter):
    """Keep the objects of the most common types, by cumulative frequency."""

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

    def keep(self, ocel) -> Keep:
        return Keep(
            objects=ocel.objects.pl.filter(
                _frequent(lambda: ocel.objects.pl, OTYPE_COL, self.threshold, self.mode)
            ).select(OID_COL)
        )

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[Optional[str], Optional[str]]
    mode: Literal["exclude", "include"] = "include"

    def keep(self, ocel) -> Keep:
        start, end = (utc_bound(bound) for bound in self.time_range)

        within = pl.lit(True)
        if start is not None:
            within = within & (pl.col(TIMESTAMP_COL) >= pl.lit(start))
        if end is not None:
            within = within & (pl.col(TIMESTAMP_COL) <= pl.lit(end))
        if self.mode == "exclude":
            within = ~within

        return Keep(events=ocel.events.pl.filter(within).select(EID_COL))

function apply_filters

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

Return a new :class:OCEL holding the subset filters agree on.

Each filter names the ids it keeps and the sets are intersected, so a pipeline keeps what all of its filters keep. A filter that leaves a side None constrains only the other one.

Source

def apply_filters(ocel: "OCEL", filters: Sequence[BaseFilter]) -> "OCEL":
    """Return a new :class:`OCEL` holding the subset ``filters`` agree on.

    Each filter names the ids it keeps and the sets are intersected, so a pipeline
    keeps what *all* of its filters keep. A filter that leaves a side ``None``
    constrains only the other one.
    """
    from ocelescope.ocel.core.ocel import OCEL

    keeps: list[Keep] = [f.keep(ocel) for f in filters]

    kept_events, kept_objects = pl.collect_all(
        [
            _intersect(
                [k.events for k in keeps if k.events is not None],
                _all_ids(ocel, EVENTS_TABLE, EID_COL),
                EID_COL,
            ),
            _intersect(
                [k.objects for k in keeps if k.objects is not None],
                _all_ids(ocel, OBJECTS_TABLE, OID_COL),
                OID_COL,
            ),
        ]
    )

    clone = ocel._copy_database()
    try:
        clone.register("kept_events", kept_events)
        clone.register("kept_objects", kept_objects)
        try:
            clone.execute(
                f'DELETE FROM events WHERE "{EID_COL}" NOT IN (SELECT "{EID_COL}" FROM kept_events)'
            )
            clone.execute(
                f'DELETE FROM objects WHERE "{OID_COL}" '
                f'NOT IN (SELECT "{OID_COL}" FROM kept_objects)'
            )
        finally:
            clone.unregister("kept_events")
            clone.unregister("kept_objects")
    except Exception:
        clone.close()
        raise

    filtered = OCEL(
        clone,
    )
    filtered.clean()
    return filtered