Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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']
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:
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