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