MT1 TA Testing solutions¶
Ex1¶
In [ ]:
def filter_errors(errors, error_codes, exclude_descriptions):
return [i for i in errors if error_codes[i['error_code']] not in exclude_descriptions]
In [ ]:
def filter_errors(errors, error_codes, exclude_descriptions):
###
filtered_errors = []
for e in errors:
if error_codes[e['error_code']] not in exclude_descriptions:
filtered_errors.append(e)
return filtered_errors
In [ ]:
exclude_codes = []
for descr in exclude_descriptions:
for code, descr2 in error_codes.items():
if descr == descr2:
exclude_codes.append(code)
output = []
for elem in errors:
if elem['error_code'] not in exclude_codes:
output.append(elem)
return output
In [ ]:
def filter_errors(errors, error_codes, exclude_descriptions):
dist_errors = [k for k, desc in error_codes.items() if desc not in exclude_descriptions]
res = []
for e in errors:
if e['error_code'] in dist_errors:
res.append(e)
return res
In [ ]:
def filter_errors(errors, error_codes, exclude_descriptions):
new_list = []
for error in errors:
if error_codes[error['error_code']] not in exclude_descriptions:
new_list.append(error)
return new_list
In [ ]:
def filter_errors(errors, error_codes, exclude_descriptions):
filtered_errors = []
for error in errors:
description = error_codes[error['error_code']]
if description not in exclude_descriptions:
filtered_errors.append(error)
return filtered_errors
Ex2¶
In [ ]:
def filter_messages_by_conversation(messages, filtered_errors):
ids = [i['conversation_id'] for i in filtered_errors]
return {k:v for k,v in messages.items() if k in ids}
In [ ]:
def filter_messages_by_conversation(messages, filtered_errors):
###
conversation_ids = [e['conversation_id'] for e in filtered_errors]
filtered_messages = {}
for k, v in messages.items():
if k in conversation_ids:
filtered_messages[k] = v
return filtered_messages
###
In [ ]:
errors_ids = [x['conversation_id'] for x in filtered_errors]
output = {}
for message, value in messages.items():
if message in errors_ids:
output[message] = value
return output
In [ ]:
def filter_messages_by_conversation(messages, filtered_errors):
filtered_conv_id = [e['conversation_id'] for e in filtered_errors]
filtered_messages = {k:v for k, v in messages.items() if k in filtered_conv_id}
return filtered_messages
In [ ]:
def filter_messages_by_conversation(messages, filtered_errors):
err_conv = []
for d in filtered_errors:
err_conv.append(d['conversation_id'])
mess = {}
for i in messages.keys():
if i in err_conv:
mess[i] = messages[i]
return mess
In [ ]:
def filter_messages_by_conversation(messages, filtered_errors):
conversation_ids = {
error['conversation_id']
for error in filtered_errors
}
filtered_messages = {
conversation_id: conversation
for conversation_id, conversation in messages.items()
if conversation_id in conversation_ids
}
return filtered_messages
Ex3¶
In [ ]:
def DEBUG_flatten_messages(filtered_messages):
# FIXME
flattened = []
all_keys = set()
for conversation_id, msgs in filtered_messages.items():
for message_entry in msgs:
for message_id, details in message_entry.items():
all_keys.update(details.keys())
headers = ['conversation_id', 'message_id'] + sorted(all_keys)
flattened.append(headers)
row = []
for conversation_id, msgs in filtered_messages.items():
for message_entry in msgs:
for message_id, details in message_entry.items():
row = [conversation_id, message_id]
row += [details.get(key) for key in sorted(all_keys)]
flattened.append(row)
return flattened
In [ ]:
def DEBUG_flatten_messages(filtered_messages):
# FIXME
flattened = []
all_keys = set()
for conversation_id, msgs in filtered_messages.items():
for message_entry in msgs:
for message_id, details in message_entry.items():
all_keys.update(details.keys())
sorted_keys = sorted(all_keys)
headers = ['conversation_id', 'message_id'] + sorted_keys
flattened.append(headers)
for conversation_id, msgs in filtered_messages.items():
for message_entry in msgs:
for message_id, details in message_entry.items():
row = [conversation_id, message_id]
row += [details.get(key) for key in sorted_keys]
flattened.append(row)
return flattened
In [ ]:
def DEBUG_flatten_messages(filtered_messages):
# FIXME
flattened = []
all_keys = set()
for conversation_id, msgs in filtered_messages.items():
for message_entry in msgs:
for message_id, details in message_entry.items():
all_keys.update(details.keys())
headers = ['conversation_id', 'message_id'] + sorted(all_keys)
flattened.append(headers)
row = []
all_keys = sorted(list(all_keys))
for conversation_id, msgs in filtered_messages.items():
for message_entry in msgs:
for message_id, details in message_entry.items():
row = [conversation_id, message_id]
row += [details.get(key) for key in all_keys]
flattened.append(row)
return flattened
In [ ]:
def DEBUG_flatten_messages(filtered_messages):
flattened = []
all_keys = set()
for conversation_id, msgs in filtered_messages.items():
for message_entry in msgs:
for message_id, details in message_entry.items():
all_keys.update(details.keys())
headers = ['conversation_id', 'message_id'] + sorted(all_keys)
flattened.append(headers)
row = []
for conversation_id, msgs in filtered_messages.items():
for message_entry in msgs:
row = []
for message_id, details in message_entry.items():
row = [conversation_id, message_id]
for key in sorted(all_keys):
row.append(details.get(key, None))
sorted(all_keys)]
flattened.append(row)
return flattened
In [ ]:
def DEBUG_flatten_messages(filtered_messages):
# FIXME
flattened = []
all_keys = set()
for conversation_id, msgs in filtered_messages.items():
for message_entry in msgs:
for message_id, details in message_entry.items():
all_keys.update(details.keys())
headers = ['conversation_id', 'message_id'] + sorted(all_keys)
sorted_all_keys = sorted(all_keys)
flattened.append(headers)
for conversation_id, msgs in filtered_messages.items():
for message_entry in msgs:
for message_id, details in message_entry.items():
row = []
row.append(conversation_id)
row.append(message_id)
for key in sorted_all_keys:
if key in details:
row.append(details[key])
else:
row.append(None)
flattened.append(row)
return flattened
In [ ]:
def DEBUG_flatten_messages(filtered_messages):
flattened = []
all_keys = set()
for conversation_id, msgs in filtered_messages.items():
for message_entry in msgs:
for message_id, details in message_entry.items():
all_keys.update(details.keys())
sorted_keys = sorted(all_keys)
headers = ['conversation_id', 'message_id'] + sorted_keys
flattened.append(headers)
for conversation_id, msgs in filtered_messages.items():
for message_entry in msgs:
for message_id, details in message_entry.items():
row = [conversation_id, message_id]
row += [
details.get(key)
for key in sorted_keys
]
flattened.append(row)
return flattened
Ex4¶
In [ ]:
def clean_conversation_strings(flattened_messages):
cleaned_messages = [row.copy() for row in flattened_messages]
headers = cleaned_messages[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
for row in cleaned_messages[1:]:
if len(row) > message_index and row[message_index]:
row[message_index] = ''.join([i for i in row[message_index] if ord(i)<128])
return cleaned_messages
In [ ]:
def clean_conversation_strings(flattened_messages):
cleaned_messages = [row.copy() for row in flattened_messages]
headers = cleaned_messages[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
for row in cleaned_messages[1:]:
if len(row) > message_index and row[message_index]:
###
row[message_index] = re.sub(r'[^\x00-\x7F]', '', row[message_index])
###
return cleaned_messages
In [ ]:
def clean_conversation_strings(flattened_messages):
cleaned_messages = [row.copy() for row in flattened_messages]
headers = cleaned_messages[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
for row in cleaned_messages[1:]:
if len(row) > message_index and row[message_index]:
###
### YOUR CODE HERE
###
message = row[3]
cleaned = ""
for char in message:
if ord(char) < 128:
cleaned += char
row[3] = cleaned
return cleaned_messages
In [ ]:
def clean_conversation_strings(flattened_messages):
import re
cleaned_messages = [row.copy() for row in flattened_messages]
headers = cleaned_messages[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
for row in cleaned_messages[1:]:
if len(row) > message_index and row[message_index]:
print(row[-2])
row[-2] = re.sub(r'[^\x00-\x7F]', '', row[-2])
return cleaned_messages
In [ ]:
def clean_conversation_strings(flattened_messages):
cleaned_messages = [row.copy() for row in flattened_messages]
headers = cleaned_messages[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
for row in cleaned_messages[1:]:
if len(row) > message_index and row[message_index]:
###
text = row.pop(message_index)
cleaned_text = text.encode("ascii", errors="ignore").decode()
row.insert(message_index, cleaned_text)
###
return cleaned_messages
In [ ]:
def clean_conversation_strings(flattened_messages):
cleaned_messages = [row.copy() for row in flattened_messages]
headers = cleaned_messages[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
for row in cleaned_messages[1:]:
if len(row) > message_index and row[message_index]:
row[message_index] = re.sub(
r'[^\x00-\x7F]',
'',
row[message_index]
)
return cleaned_messages
Ex5¶
In [ ]:
def DEBUG_remove_stop_words_and_punctuation(cleaned_messages, stop_words):
# FIXME
cleansed_messages = [row.copy() for row in cleaned_messages]
headers = cleansed_messages[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
translator = str.maketrans('', '', string.punctuation)
for row in cleansed_messages[1:]:
if len(row) > message_index and row[message_index]:
message = row[message_index].translate(translator)
words = message.split()
filtered_words = [word for word in words if word.lower() not in stop_words]
row[message_index] = ' '.join(filtered_words)
return cleansed_messages
In [ ]:
def DEBUG_remove_stop_words_and_punctuation(cleaned_messages, stop_words):
# FIXME
cleansed_messages = [row.copy() for row in cleaned_messages]
headers = cleansed_messages[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
translator = str.maketrans('', '', string.punctuation)
for row in cleansed_messages[1:]:
if len(row) > message_index and row[message_index]:
no_punctuation = row[message_index].translate(translator)
words = no_punctuation.split()
filtered_words = [word for word in words if word.lower() not in stop_words]
row[message_index] = ' '.join(filtered_words)
return cleansed_messages
In [ ]:
def DEBUG_remove_stop_words_and_punctuation(cleaned_messages, stop_words):
# FIXME
cleansed_messages = [row.copy() for row in cleaned_messages]
headers = cleansed_messages[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
translator = str.maketrans('', '', string.punctuation)
for row in cleansed_messages[1:]:
if len(row) > message_index and row[message_index]:
msg = row[message_index].translate(translator)
words = msg.split()
filtered_words = [word for word in words if word.lower() not in stop_words]
row[message_index] = ' '.join(filtered_words)
return cleansed_messages
In [ ]:
def DEBUG_remove_stop_words_and_punctuation(cleaned_messages, stop_words):
cleansed_messages = [row.copy() for row in cleaned_messages]
headers = cleansed_messages[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
translator = str.maketrans('', '', string.punctuation)
for row in cleansed_messages[1:]:
if len(row) > message_index and row[message_index]:
words = row[message_index].split()
filtered_words = [word for word in words if word.translate(translator).lower() not in stop_words]
filtered_words = [word.translate(translator) for word in filtered_words]
row[message_index] = ' '.join([f for f in filtered_words if f.lower() not in stop_words])
return cleansed_messages
In [ ]:
def DEBUG_remove_stop_words_and_punctuation(cleaned_messages, stop_words):
# FIXME
cleansed_messages = [row.copy() for row in cleaned_messages]
headers = cleansed_messages[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
translator = str.maketrans('', '', string.punctuation)
for row in cleansed_messages[1:]:
if len(row) > message_index and row[message_index]:
words = row[message_index].split()
# print(words)
filtered_words = [word.translate(translator) for word in words]
filtered_words = [word for word in filtered_words if word.lower() not in stop_words]
row[message_index] = ' '.join(filtered_words)
return cleansed_messages
In [ ]:
def DEBUG_remove_stop_words_and_punctuation(cleaned_messages, stop_words):
cleansed_messages = [row.copy() for row in cleaned_messages]
headers = cleansed_messages[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
translator = str.maketrans('', '', string.punctuation)
for row in cleansed_messages[1:]:
if len(row) > message_index and row[message_index]:
message_without_punctuation = (
row[message_index].translate(translator)
)
words = message_without_punctuation.split()
filtered_words = [
word
for word in words
if word.lower() not in stop_words
]
row[message_index] = ' '.join(filtered_words)
return cleansed_messages
Ex6¶
In [ ]:
def sentiment_analysis(nostopwords_messages):
messages_with_sentiment = [row.copy() for row in nostopwords_messages]
sia = CustomSentimentIntensityAnalyzer()
messages_with_sentiment[0].append('sentiment')
headers = messages_with_sentiment[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
for i in messages_with_sentiment[1:]:
if i[message_index]:
score = sia.polarity_scores(i[message_index])
if score['compound'] >= 0:
i.append('positive')
else:
i.append('negative')
return messages_with_sentiment
In [ ]:
###
for row in messages_with_sentiment[1:]:
if len(row) > message_index and row[message_index]:
scores = sia.polarity_scores(row[message_index])
row.append('positive' if scores['compound'] >= 0 else 'negative')
###
In [ ]:
def sentiment_analysis(nostopwords_messages):
messages_with_sentiment = [row.copy() for row in nostopwords_messages]
sia = CustomSentimentIntensityAnalyzer()
messages_with_sentiment[0].append('sentiment')
headers = messages_with_sentiment[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
###
### YOUR CODE HERE
###
for message in messages_with_sentiment[1:]:
# print(message)
if message == "":
continue
score = sia.polarity_scores(message[3])
if score['compound'] >= 0:
message.append("positive")
else:
message.append("negative")
return messages_with_sentiment
In [ ]:
def sentiment_analysis(nostopwords_messages):
messages_with_sentiment = [row.copy() for row in nostopwords_messages]
sia = CustomSentimentIntensityAnalyzer()
messages_with_sentiment[0].append('sentiment')
headers = messages_with_sentiment[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
for i, row in enumerate(messages_with_sentiment):
if i > 0:
# print(row[-2])
cur_msg = row[-2]
if cur_msg:
score = 'positive' if sia.polarity_scores(cur_msg)['compound'] >= 0 else 'negative'
else:
score = None
row.append(score)
return messages_with_sentiment
In [ ]:
def sentiment_analysis(nostopwords_messages):
messages_with_sentiment = [row.copy() for row in nostopwords_messages]
sia = CustomSentimentIntensityAnalyzer()
messages_with_sentiment[0].append('sentiment')
headers = messages_with_sentiment[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
###
for row in messages_with_sentiment[1:]:
if row[message_index]:
comp = sia.polarity_scores(row[message_index])['compound']
if comp < 0:
row.append('negative')
else:
row.append('positive')
###
return messages_with_sentiment
In [ ]:
def sentiment_analysis(nostopwords_messages):
messages_with_sentiment = [
row.copy()
for row in nostopwords_messages
]
sia = CustomSentimentIntensityAnalyzer()
messages_with_sentiment[0].append('sentiment')
headers = messages_with_sentiment[0]
if 'message' not in headers:
raise ValueError("No 'message' column found in the headers.")
message_index = headers.index('message')
for row in messages_with_sentiment[1:]:
if len(row) > message_index and row[message_index]:
compound = sia.polarity_scores(
row[message_index]
)['compound']
if compound >= 0:
row.append('positive')
else:
row.append('negative')
return messages_with_sentiment
Ex7¶
In [ ]:
def DEBUG_detect_daily_outliers(filtered_errors):
# FIXME
date_counts = {}
for error in filtered_errors:
error_dt = datetime.strptime(error['datetime'], '%Y-%m-%d %H:%M:%S')
error_date_str = error_dt.strftime('%Y-%m-%d')
date_counts[error_date_str] = date_counts.get(error_date_str, 0) + 1
counts = sorted(date_counts.values())
n = len(counts)
Q1 = counts[(n-1) // 4]
Q3 = counts[(3 * (n-1)) // 4]
IQR = Q3 - Q1
upper_bound = Q3 + 1.5 * IQR
return [date for date, count in date_counts.items() if count > upper_bound]
In [ ]:
def DEBUG_detect_daily_outliers(filtered_errors):
# FIXME
date_counts = {}
for error in filtered_errors:
error_dt = datetime.strptime(error['datetime'], '%Y-%m-%d %H:%M:%S')
error_date_str = error_dt.strftime('%Y-%m-%d')
date_counts[error_date_str] = date_counts.get(error_date_str, 0) + 1
counts = sorted(date_counts.values())
n = len(counts)
Q1 = counts[(n - 1) // 4]
Q3 = counts[(3 * (n - 1)) // 4]
IQR = Q3 - Q1
upper_bound = Q3 + 1.5 * IQR
return [date for date, count in date_counts.items() if count > upper_bound]
In [ ]:
def DEBUG_detect_daily_outliers(filtered_errors):
# FIXME
date_counts = {}
for error in filtered_errors:
error_dt = datetime.strptime(error['datetime'], '%Y-%m-%d %H:%M:%S')
error_date_str = error_dt.strftime('%Y-%m-%d')
date_counts[error_date_str] = date_counts.get(error_date_str, 0) + 1
counts = sorted(date_counts.values())
n = len(counts)
Q1 = counts[(n-1) // 4]
Q3 = counts[(3 * (n-1)) // 4]
IQR = Q3 - Q1
upper_bound = Q3 + 1.5 * IQR
return [date for date, count in date_counts.items() if count > upper_bound]
In [ ]:
def DEBUG_detect_daily_outliers(filtered_errors):
date_counts = {}
for error in filtered_errors:
error_dt = datetime.strptime(error['datetime'], '%Y-%m-%d %H:%M:%S')
error_date_str = error_dt.strftime('%Y-%m-%d')
date_counts[error_date_str] = date_counts.get(error_date_str, 0) + 1
counts = sorted(date_counts.values())
n = len(counts)
Q1 = counts[(n - 1) // 4]
Q3 = counts[(3 * (n - 1)) // 4]
IQR = Q3 - Q1
upper_bound = Q3 + 1.5 * IQR
return [date for date, count in date_counts.items() if count > upper_bound]
In [ ]:
def DEBUG_detect_daily_outliers(filtered_errors):
# FIXME
date_counts = {}
for error in filtered_errors:
error_dt = datetime.strptime(error['datetime'], '%Y-%m-%d %H:%M:%S')
error_date_str = error_dt.strftime('%Y-%m-%d')
date_counts[error_date_str] = date_counts.get(error_date_str, 0) + 1
counts = sorted(date_counts.values())
n = len(counts)
Q1 = counts[(n-1) // 4]
Q3 = counts[(3 * (n-1)) // 4]
IQR = Q3 - Q1
upper_bound = Q3 + 1.5 * IQR
return [date for date, count in date_counts.items() if count > upper_bound]
In [ ]:
def DEBUG_detect_daily_outliers(filtered_errors):
date_counts = {}
for error in filtered_errors:
error_dt = datetime.strptime(
error['datetime'],
'%Y-%m-%d %H:%M:%S'
)
error_date_str = error_dt.strftime('%Y-%m-%d')
date_counts[error_date_str] = (
date_counts.get(error_date_str, 0) + 1
)
counts = sorted(date_counts.values())
if not counts:
return []
n = len(counts)
Q1 = counts[(n - 1) // 4]
Q3 = counts[(3 * (n - 1)) // 4]
IQR = Q3 - Q1
upper_bound = Q3 + 1.5 * IQR
return [
date
for date, count in date_counts.items()
if count > upper_bound
]
Ex8¶
In [ ]:
def filter_schedule_by_outliers(outliers, schedule):
return [i for i in schedule if i['date'] in outliers]
In [ ]:
def filter_schedule_by_outliers(outliers, schedule):
###
filtered_schedule = []
for event in schedule:
if event['date'] in outliers:
filtered_schedule.append(event)
return filtered_schedule
###
In [ ]:
def filter_schedule_by_outliers(outliers, schedule):
###
### YOUR CODE HERE
###
output = []
for elem in schedule:
if elem['date'] in outliers:
output.append(elem)
return output
In [ ]:
def summarize_outliers_by_label(outliers, schedule):
from collections import Counter
output = []
labels = [s['label'] for s in schedule]
count_by_label = Counter(labels)
relevant_records = [s['label'] for s in schedule if s['date'] in outliers]
outlier_count_by_label = Counter(relevant_records)
for k, v in count_by_label.items():
total_days = v
outlier_days = outlier_count_by_label.get(k, 0)
outlier_rate = round(outlier_days / total_days, 3)
output.append((k,outlier_days, total_days, outlier_rate))
output = sorted(output, key= lambda x: (-x[3], x[0]))
return output
In [ ]:
def summarize_outliers_by_label(outliers, schedule):
out_la = {}
t_la_date_count = {}
for d in schedule:
t_la_date_count[d['label']] = t_la_date_count.get(d['label'], 0) + 1
if d['date'] in outliers:
out_la[d['label']] = out_la.get(d['label'], 0) + 1
# print(t_la_date_count)
# print(out_la)
res = []
for k, v in t_la_date_count.items():
a = k
if k in out_la:
b = out_la[k]
else:
b = 0
c = v
d = round(b / c, 3)
res.append((a, b, c, d))
res.sort(key=lambda x: (-x[3], x[0]))
# print(res)
return res
In [ ]:
def summarize_outliers_by_label(outliers, schedule):
outlier_set = set(outliers)
label_counts = {}
for entry in schedule:
label = entry['label']
date = entry['date']
if label not in label_counts:
label_counts[label] = {
'outlier_days': 0,
'total_days': 0
}
label_counts[label]['total_days'] += 1
if date in outlier_set:
label_counts[label]['outlier_days'] += 1
result = []
for label, counts in label_counts.items():
outlier_days = counts['outlier_days']
total_days = counts['total_days']
outlier_rate = round(
outlier_days / total_days,
3
)
result.append(
(
label,
outlier_days,
total_days,
outlier_rate
)
)
result.sort(
key=lambda row: (-row[3], row[0])
)
return result