Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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:
requests_list.append(str(e))
logger.warning(str(e))
return requests_list
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
replay_job.save()
try: #TODO check logger usage
logger.info("Sending request for replay_prediction task.")
r = requests.post(
url="http://server:8000/runtime/replay_prediction/",
data=export_log_as_string(filtered_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:
requests_list.append(str(e))
logger.warning(str(e))
training_log, test_log, additional_columns = get_train_test_log(replay_job.split)
training_df, _ = encode_label_logs(training_log, test_log, replay_job, additional_columns=additional_columns)
gold_values = dict(zip(training_df['trace_id'], training_df['label']))
parent_id = replay_job.id
final_job = duplicate_orm_row(replay_job)
final_job.parent_job = Job.objects.filter(pk=parent_id)[0]
final_job.gold_value = gold_values