Discovery

class DiscoveryMethodMeta

class DiscoveryMethodMeta:

Source

@dataclass
class DiscoveryMethodMeta:
    name: str
    description: str | None
    func: Callable[..., Resource]
    resource_type: type[Resource]

function discovery_method

def discovery_method(name: str, description: str | None = None) -> Callable[[Callable[..., Resource]], Callable[..., Resource]]:

Mark a function as a discovery method.

Stamps __discovery_meta__ on the function. The backend scans for this attribute at startup, derives the parameter schema from the function signature, and exposes the method through the discovery API.

Source

def discovery_method(
    *,
    name: str,
    description: str | None = None,
) -> Callable[[Callable[..., Resource]], Callable[..., Resource]]:
    """Mark a function as a discovery method.

    Deprecated:
        Expose discovery algorithms as `@plugin_method`s on a `Plugin` instead.

    Stamps `__discovery_meta__` on the function. The backend scans for this
    attribute at startup, derives the parameter schema from the function
    signature, and exposes the method through the discovery API.
    """
    warnings.warn(
        "@discovery_method is deprecated; expose discovery algorithms as "
        "@plugin_method on a Plugin instead.",
        DeprecationWarning,
        stacklevel=2,
    )

    def decorator(func: Callable[..., Resource]) -> Callable[..., Resource]:
        hints = get_type_hints(func, include_extras=True)
        resource_type = _extract_resource_type(func.__name__, hints.get("return"))  # ty: ignore[unresolved-attribute]

        func.__discovery_meta__ = DiscoveryMethodMeta(  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
            name=name,
            description=description,
            func=func,
            resource_type=resource_type,
        )
        return func

    return decorator

function inductive_miner

def inductive_miner(ocel: OCEL, noise_threshold: Annotated[float, Field(ge=0, le=1, title='Noise Threshold', description='Fraction of infrequent behaviour to filter out (0 = no filtering, IMf variant).')] = 0.8, included_object_types: list[str] | None = None, included_activities: list[str] | None = None) -> PetriNet:

Source

def inductive_miner(
    ocel: OCEL,
    noise_threshold: Annotated[
        float,
        Field(
            ge=0,
            le=1,
            title="Noise Threshold",
            description="Fraction of infrequent behaviour to filter out (0 = no filtering, IMf variant).",
        ),
    ] = 0.8,
    included_object_types: list[str] | None = None,
    included_activities: list[str] | None = None,
) -> PetriNet:
    pm4py_ocel = _slim_pm4py_ocel(ocel, included_object_types, included_activities)
    # pm4py cannot mine an empty log, e.g. when nothing is selected.
    if pm4py_ocel.relations.empty:
        return PetriNet()

    ocpn = pm4py.discover_oc_petri_net(
        ocel=pm4py_ocel,
        noise_threshold=noise_threshold,
        disable_fallthroughs=False,
        disable_strict_sequence_cut=False,
    )
    return PetriNet.from_pm4py(ocpn)

function ocdfg_miner

def ocdfg_miner(ocel: OCEL, frequency_threshold: Annotated[float, Field(ge=0, le=1)] = 1, included_activities: list[str] | None = None, included_object_types: list[str] | None = None) -> DirectlyFollowsGraph:

Source

def ocdfg_miner(
    ocel: OCEL,
    frequency_threshold: Annotated[
        float,
        Field(
            ge=0,
            le=1,
        ),
    ] = 1,
    included_activities: list[str] | None = None,
    included_object_types: list[str] | None = None,
) -> DirectlyFollowsGraph:
    edges = [
        DFGEdge(
            object_type=object_type,
            source=source,
            target=target,
            count=count,
            object_count=object_count,
        )
        for object_type, source, target, count, object_count in ocel.sql(
            _dfg_query(
                included_activities=included_activities,
                included_object_types=included_object_types,
            )
        ).fetchall()
    ]

    activities = sorted(
        {name for edge in edges for name in (edge.source, edge.target) if name}
    )
    object_types = sorted({edge.object_type for edge in edges})
    dfg = DirectlyFollowsGraph(
        activities=[DFGActivity(name=name) for name in activities],
        object_types=[DFGObject(name=name) for name in object_types],
        edges=edges,
    )
    return dfg.filter_edges(frequency_threshold)