Objects

class ObjectsManager(BaseManager):
    """
    Manages object-level information within an OCEL instance.

    Provides access to:
    - the objects table
    - the object_changes table
    - object types and counts
    - object attribute names
    - per-object lookup helpers such as type-by-id

    Acts as a facade over the underlying PM4PY OCEL object.
    """

    @property
    def df(self) -> pd.DataFrame:
        """
        Return the object table from the underlying OCEL.

        Returns:
            DataFrame: A pandas DataFrame containing all objects and their static attributes.
        """
        return self._ocel.ocel.objects

    @property
    def changes(self) -> pd.DataFrame:
        """
        Return the dynamic object attribute change table.

        Returns:
            DataFrame: A pandas DataFrame containing all dynamic updates to object attributes.
        """
        return self._ocel.ocel.object_changes

    @property
    @instance_lru_cache()
    def types(self) -> list[str]:
        """
        Return the list of all object types present in the log.

        Returns:
            list[str]: Sorted list of unique object type names.
        """
        return list(sorted(self.df[OTYPE_COL].unique().tolist()))

    @property
    @instance_lru_cache()
    def counts(self) -> pd.Series:
        """
        Count how many objects exist for each object type.

        Returns:
            Series: A pandas Series indexed by object type with occurrence counts.
        """
        return self.df[OTYPE_COL].value_counts()

    @property
    @instance_lru_cache()
    def type_by_id(self) -> pd.Series:
        """
        Return a mapping from object ID to object type.

        Returns:
            Series: A pandas Series indexed by object ID, containing object types as values.
        """
        return cast(pd.Series, self.df[[OID_COL, OTYPE_COL]].set_index(OID_COL)[OTYPE_COL])

    def has_types(self, types: Iterable[str]) -> bool:
        """
        Check whether all provided object types exist in the OCEL.

        Args:
            types: Iterable of object type names to verify.

        Returns:
            bool: True if all types exist, False otherwise.
        """
        return all(ot in self.types for ot in types)

    @property
    def static_attribute_names(self) -> list[str]:
        """
        Return the names of all static object attributes.

        Static attributes are non-OCEL-prefixed columns in the objects
        table that contain at least one non-null value.

        Returns:
            list[str]: Sorted list of static object attribute names.
        """
        return sorted(
            [col for col in self.df.columns[self.df.count() > 0] if not col.startswith("ocel:")]
        )

    @property
    def dynamic_attribute_names(self) -> list[str]:
        """
        Return the names of all dynamic object attributes.

        Dynamic attributes are derived from the object_changes table,
        excluding OCEL system columns and internal counters.

        Returns:
            list[str]: Sorted list of dynamic object attribute names.
        """
        return sorted(
            [
                col
                for col in self.changes.columns[self.changes.count() > 0]
                if not col.startswith("ocel:") and col != "@@cumcount"
            ]
        )

    @property
    def attribute_names(self) -> list[str]:
        """
        Return all object attribute names.

        Combines both static and dynamic attributes into a unified list.

        Returns:
            list[str]: Sorted list of all object attribute names.
        """
        return sorted(set(self.static_attribute_names + self.dynamic_attribute_names))

    @property
    @instance_lru_cache()
    def attribute_summary(self) -> pd.DataFrame:
        """Return an attribute summary for objects, grouped by object type.

        RETURNS:
            A pandas DataFrame indexed by (ATTRIBUTE_COL, OTYPE_COL) containing the
            summary statistics produced by `get_summary`.
        """

        return self._ocel.attributes.get_object_summary()

    def object_attr_changes(
        self,
        object_types: Iterable[Any] | None = None,
        objects: Iterable[Any] | None = None,
        attributes: Iterable[Any] | None = None,
    ):
        """
        Return dynamic object attributes over time.

        Filters `object_changes` by object type and/or object id, forward-fills
        attribute values per object, and returns one row per `(object_id, timestamp)`.

        Args:
            object_types: Optional object types to include.
            objects: Optional object ids to include.
            attributes: Optional dynamic attribute names to include. If omitted, all
                dynamic attributes are returned.

        Returns:
            pandas.DataFrame: DataFrame indexed by `(ocel:oid, ocel:timestamp)` with
            the selected dynamic attribute columns and the object type column
            (`ocel:type`).
        """

        attr_cols = [
            attr_name
            for attr_name in self.dynamic_attribute_names
            if attributes is None or attr_name in attributes
        ]

        changes = self.changes

        mask = pd.Series(True, index=changes.index)

        if object_types is not None:
            mask &= changes[OTYPE_COL].isin(object_types)

        if objects is not None:
            mask &= changes[OID_COL].isin(objects)

        changes = cast(pd.DataFrame, changes.loc[mask])

        changes.sort_values([OID_COL, TIMESTAMP_COL])

        changes[attr_cols] = changes.groupby(OID_COL)[attr_cols].ffill()

        return (
            changes.assign(_nn=changes.notna().sum(axis=1))
            .reset_index()
            .sort_values([TIMESTAMP_COL, OID_COL] + ["_nn"])
            .drop_duplicates(subset=[TIMESTAMP_COL, OID_COL], keep="last")
            .set_index([OID_COL, TIMESTAMP_COL], drop=True)[
                attr_cols
                + [
                    OTYPE_COL,
                ]
            ]
        )