def append()

in log_analyzer/log.py [0:0]

• 23 lines of code
• 5 McCabe index (conditional complexity)

    def append(self, log_item: Union[LogItem, Dict]):
        if isinstance(log_item, LogItem):
            self.workload.append(log_item)
            return
        if "stage" not in log_item:
            log_item["stage"] = log_item["operation"] if "operation" in log_item else ""
        if "comm_group" not in log_item:
            assert (
                log_item["comm_type"] == CommType.computation
            ), "comm_group is required for non-computation comm_type"
            log_item["comm_group"] = CommGroup.all
        self.workload.append(
            LogItem(
                comm_type=log_item["comm_type"],
                comm_group=log_item["comm_group"],
                comm_group_size=log_item["comm_group_size"],
                msg_size=log_item["msg_size"],
                stage=log_item["stage"],
                src=log_item.get("src", None),
                dst=log_item.get("dst", None),
                additional=log_item.get("additional", None),
            )
        )