def model_failures()

in utils/ryzenai/notification_service.py [0:0]

• 38 lines of code
• 6 McCabe index (conditional complexity)

    def model_failures(self):
        # Load baseline data from a JSON file
        with open(tu.BASELINE_JSON, "r") as json_file:
            baseline_data = json.load(json_file)

        model_failure_sections = []

        text = (
            "The following section presents category-wise failures for models, illustrating "
            "the total number of DPU and CPU operators associated with each failure. If a failure is "
            "attributed to regression (indicated as 'Reg.' in the table), baseline values are provided "
            "in parentheses. For other failures, operator values are not displayed. Please refer to the "
            "post reply for additional details on these failures."
        )

        model_failure_sections.append(
            {"type": "header", "text": {"type": "plain_text", "text": "Category wise failures", "emoji": True}},
        )

        content = {"type": "section", "text": {"type": "plain_text", "text": text, "emoji": True}}
        model_failure_sections.append(content)

        for i, (key, result) in enumerate(self.model_results.items()):
            failures_info = []

            for failure in result["failures"]:
                # Extract information from failure details
                line = failure["line"]
                trace = failure["trace"]

                # Identify model_id based on the failure line
                model_id = self.extract_model_id(line)
                model_id = infer_model_id(model_id)

                # Get baseline values for the identified model_id
                baseline_ops = baseline_data.get(model_id.lower().replace("/", "_"), {})

                # Extract baseline values
                cpu_baseline_value = baseline_ops.get("cpu", 0)
                dpu_baseline_value = baseline_ops.get("dpu", 0)
                all_baseline_value = baseline_ops.get("all", 0)

                # Extract and compare values from the failure trace
                all_value_str, dpu_value_str, cpu_value_str, regressed = self.extract_operator_values(
                    trace, all_baseline_value, dpu_baseline_value, cpu_baseline_value
                )

                # Append information about the failure
                failures_info.append(
                    f"{all_value_str.rjust(9)} | {dpu_value_str.rjust(9)} | {cpu_value_str.rjust(9)} | {regressed.rjust(4)} | {model_id[:40]}"
                )

            if len(failures_info):
                # Prepare model failure sections
                model_failure_sections.extend(
                    self.prepare_model_failure_sections(i + 1, key, result["job_link"], failures_info)
                )

        return model_failure_sections