File size: 3,690 Bytes
b400ace | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | import re
from copy import deepcopy
from fractions import Fraction
from deepmerge import Merger
from dotmap import DotMap
from src.constants import FIELD_LABEL_NUMBER_REGEX
from src.defaults import CONFIG_DEFAULTS, TEMPLATE_DEFAULTS
from src.schemas.constants import FIELD_STRING_REGEX_GROUPS
from src.utils.file import load_json
from src.utils.validations import (
validate_config_json,
validate_evaluation_json,
validate_template_json,
)
OVERRIDE_MERGER = Merger(
# pass in a list of tuples,with the
# strategies you are looking to apply
# to each type.
[
# (list, ["prepend"]),
(dict, ["merge"])
],
# next, choose the fallback strategies,
# applied to all other types:
["override"],
# finally, choose the strategies in
# the case where the types conflict:
["override"],
)
def get_concatenated_response(omr_response, template):
# Multi-column/multi-row questions which need to be concatenated
concatenated_response = {}
for field_label, concatenate_keys in template.custom_labels.items():
custom_label = "".join([omr_response[k] for k in concatenate_keys])
concatenated_response[field_label] = custom_label
for field_label in template.non_custom_labels:
concatenated_response[field_label] = omr_response[field_label]
return concatenated_response
def open_config_with_defaults(config_path):
user_tuning_config = load_json(config_path)
user_tuning_config = OVERRIDE_MERGER.merge(
deepcopy(CONFIG_DEFAULTS), user_tuning_config
)
validate_config_json(user_tuning_config, config_path)
# https://github.com/drgrib/dotmap/issues/74
return DotMap(user_tuning_config, _dynamic=False)
def open_template_with_defaults(template_path):
user_template = load_json(template_path)
user_template = OVERRIDE_MERGER.merge(deepcopy(TEMPLATE_DEFAULTS), user_template)
validate_template_json(user_template, template_path)
return user_template
def open_evaluation_with_validation(evaluation_path):
user_evaluation_config = load_json(evaluation_path)
validate_evaluation_json(user_evaluation_config, evaluation_path)
return user_evaluation_config
def parse_fields(key, fields):
parsed_fields = []
fields_set = set()
for field_string in fields:
fields_array = parse_field_string(field_string)
current_set = set(fields_array)
if not fields_set.isdisjoint(current_set):
raise Exception(
f"Given field string '{field_string}' has overlapping field(s) with other fields in '{key}': {fields}"
)
fields_set.update(current_set)
parsed_fields.extend(fields_array)
return parsed_fields
def parse_field_string(field_string):
if "." in field_string:
field_prefix, start, end = re.findall(FIELD_STRING_REGEX_GROUPS, field_string)[
0
]
start, end = int(start), int(end)
if start >= end:
raise Exception(
f"Invalid range in fields string: '{field_string}', start: {start} is not less than end: {end}"
)
return [
f"{field_prefix}{field_number}" for field_number in range(start, end + 1)
]
else:
return [field_string]
def custom_sort_output_columns(field_label):
label_prefix, label_suffix = re.findall(FIELD_LABEL_NUMBER_REGEX, field_label)[0]
return [label_prefix, int(label_suffix) if len(label_suffix) > 0 else 0]
def parse_float_or_fraction(result):
if type(result) == str and "/" in result:
result = float(Fraction(result))
else:
result = float(result)
return result
|