How to use the pm4py.objects.log.log.Trace function in pm4py

To help you get started, we’ve selected a few pm4py examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github nirdizati-research / predict-python / src / encoding / declare / declare.py View on Github external
def declare_encoding(log, labelling, encoding, additional_columns, cols=None): #TODO JONAS
    filter_t = True
    print("Filter_t", filter_t)
    templates = template_sizes.keys()

    constraint_threshold = 0.1
    candidate_threshold = 0.1

    #apply prefix
    log = [Trace(trace[:encoding.prefix_length], attributes=trace.attributes) for trace in log]

    # Read into suitable data structure
    transformed_log = xes_to_positional(log)
    labels = {trace.attributes['concept:name']: trace.attributes['label'] for trace in log}

    # Extract unique activities from log
    events_set = {event_label for tid in transformed_log for event_label in transformed_log[tid]}

    # Brute force all possible candidates
    if cols is None:
        candidates = [(event,) for event in events_set] + [(e1, e2) for e1 in events_set for e2 in events_set if e1 != e2]
    else:
        candidates = list({
            make_tuple(c.split(':')[1]) if len(c.split(':')) > 1 else c
            for c in cols
            if c not in ['label', 'trace_id']
github nirdizati-research / predict-python / src / runtime / replay.py View on Github external
simulating the log in the time passing
        :param trace_id:
        :param replay_job: job dictionary
        :param training_initial_job: job dictionary
        :return: List of requests
    """

    split = replay_job.split
    log = get_log(split.train_log)
    requests_list = list()
    eventlog = EventLog()
    trace = log[int(trace_id)]
    for key in log.attributes.keys():
        eventlog.attributes[key] = log.attributes[key]
    for index in range(len(trace)):
        new_trace = Trace(trace[0:index])
        for key in trace.attributes:
            new_trace.attributes[key] = trace.attributes[key]
        eventlog.append(new_trace)
    replay_job.case_id = trace_id
    replay_job.event_number = len(trace)
    replay_job.save()
    try:
        logger.error("Sending request for replay_prediction task.")
        r = requests.post(
            url="http://127.0.0.1:8000/runtime/replay_prediction/",
            data=export_log_as_string(eventlog),
            params={'jobId': replay_job.id, 'training_job': training_initial_job.id},
            headers={'Content-Type': 'text/plain', 'charset': 'UTF-8'}
        )
        requests_list.append(str(r))
    except Exception as e:
github nirdizati-research / predict-python / src / runtime / replay.py View on Github external
simulating the log in the time passing

        :param replay_job: job dictionary
        :param training_initial_job: job dictionary
        :return: List of requests
    """

    split = replay_job.split
    log = get_log(split.train_log)
    requests_list = list()

    eventlog = EventLog()
    for key in log.attributes.keys():
        eventlog.attributes[key] = log.attributes[key]
    for trace in log:
        new_trace = Trace(trace)
        for key in trace.attributes:
            new_trace.attributes[key] = trace.attributes[key]
        eventlog.append(new_trace)

    times = sorted(set([event['time:timestamp'] for trace in eventlog for event in trace]))

    for t in times[2::5]:
        filtered_eventlog = timestamp_filter.apply_events(eventlog, times[0].replace(tzinfo=None),
                                                          t.replace(tzinfo=None))
        trace_list = list()
        event_number = dict()
        for trace in filtered_eventlog:
            trace_list.append(trace.attributes['concept:name'])
            event_number[trace.attributes['concept:name']] = len(trace)
        replay_job.case_id = trace_list
        replay_job.event_number = event_number