def parse_ds_log_item()

in log_analyzer/ds_comm_log_analyzer.py [0:0]

• 47 lines of code
• 22 McCabe index (conditional complexity)

def parse_ds_log_item(line):
    index = line.lower().find(LOG_STARTER)
    if index == -1:
        return None
    item_list = line[index + len(LOG_STARTER) :].split("|")
    item = {}
    for raw_item in item_list:
        if "epoch" in raw_item:
            split_text = raw_item.split()
            numbers = [word for word in split_text if word.isdigit()]
            item["epoch_num"] = int(numbers[0])
            continue
        if "micro_step" in raw_item:
            split_text = raw_item.split()
            numbers = [word for word in split_text if word.replace(".", "").isdigit()]
            item["iter_time"] = float(numbers[0])
            continue
        if ":" not in raw_item:
            continue
        key, value = raw_item.split(":")
        key, value = clean_s(key), clean_s(value)
        if key == COMM_OP:
            item["comm_type"] = string2comm_type(value)
        elif key == MSG_SIZE or MSG_SIZE in key:
            item["msg_size"] = convert_msg_to_size(value)
        elif key == CALLER_FUNC:
            item["stage"] = value
        elif key == TIME_MS or TIME_MS in key:
            item["elapsed_time"] = float(value)
        if key == "group":
            group = eval(value)
            if len(group) == WORLD_SIZE:
                item["group"] = CommGroup.all
            elif len(group) == TP_SIZE:
                item["group"] = CommGroup.tp_group
            elif len(group) == DP_SIZE:
                item["group"] = CommGroup.dp_group
        elif "algbw" in key:
            item["algbw"] = float(value)
        elif "busbw" in key:
            item["busbw"] = float(value)
        else:
            try:
                item[key] = float(value)
            except:
                item[key] = value
    return item