Diff Coverage

        return False
    if search_space.pp_range and strategy.pp_degree not in search_space.pp_range:
        return False
    if search_space.cp_range and strategy.cp_degree not in search_space.cp_range:
        return False
    if search_space.micro_batch_num_range and strategy.micro_batch_num not in search_space.micro_batch_num_range:
        return False
    return True


def generate_candidates_sapp(

    _patch_evaluator_set_config(parallelize)
    parallelize.instance.enable_debug = False
    ccfg = parallelize.config.ccfg
    if "vp" not in ccfg.__dict__ or ccfg.__dict__.get("vp", 0) < 1:
        ccfg.vp = 1
    parallelize.bound_space()
    space = parallelize.generate_search_space(
        folder="", threads_num=None
    )

    )
    if not os.path.isdir(nd_output_dir):
        return collected

    for pattern in ["results.pdf", "debug_*.csv"]:
        for src in sorted(glob.glob(os.path.join(nd_output_dir, pattern))):
            dst = os.path.join(output_dir, os.path.basename(src))
            shutil.copy2(src, dst)
            collected.append(dst)

    return collected


def generate_candidates(
    hardware: HardwareConfig,

                continue
            remaining = n // (tp * pp)
            for cp in cp_values:
                if remaining % cp != 0:
                    continue
                dp = remaining // cp
                if dp < 1:
                    continue
                if dp_values and dp not in dp_values:
                    continue

                mbn_values = _enumerate_mbn_values(gbs, dp, pp, search_space)

                    continue

                mbn_values = _enumerate_mbn_values(gbs, dp, pp, search_space)
                if not mbn_values and not search_space.micro_batch_num_range:
                    mbn_values = [1]

                for mbn in mbn_values:
                    strategy = ParallelStrategy(
                        dp_degree=dp,

            filtered.append(s)
            continue

        if s.tp_degree > kv_heads:
            s.mark_infeasible(
                f"tp_degree={s.tp_degree} > num_kv_heads={kv_heads}"
            )
            filtered.append(s)
            continue

        if s.pp_degree > 1 and model.num_layers % s.pp_degree != 0:
            s.mark_infeasible(
                f"num_layers={model.num_layers} not divisible by "

            continue

        total = s.dp_degree * s.tp_degree * s.pp_degree * s.cp_degree
        if total != n:
            s.mark_infeasible(
                f"dp*tp*pp*cp={total} != num_devices={n}"
            )
            filtered.append(s)
            continue

        if s.pp_degree > 1 and s.micro_batch_num < s.pp_degree:
            s.mark_infeasible(
                f"micro_batch_num={s.micro_batch_num} < pp_degree={s.pp_degree}"

        return strategies

    for s in strategies:
        if not s.is_feasible:
            continue
        if s.memory_cost.total > constraints.memory_limit_mb:
            s.mark_infeasible(
                f"memory={s.memory_cost.total:.0f}MB > "
                f"limit={constraints.memory_limit_mb:.0f}MB"

        Filtered list of strategies.
    """
    for s in strategies:
        if not s.is_feasible:
            continue
        if s.dp_degree < constraints.min_dp_degree:
            s.mark_infeasible(
                f"dp_degree={s.dp_degree} < min_dp_degree={constraints.min_dp_degree}"
            )

        return strategies

    for s in strategies:
        if not s.is_feasible:
            continue
        if s.performance_cost.total > constraints.max_step_time_ms:
            s.mark_infeasible(
                f"step_time={s.performance_cost.total:.1f}ms > "
                f"max_step_time={constraints.max_step_time_ms:.1f}ms"

            extra["dimensions"] = dimensions
        par = Parallelize("hyper_v2", yaml_path, machine, **extra)
        ccfg = par.config.ccfg
        if not hasattr(ccfg, "vp") or ccfg.vp == 0:
            ccfg.vp = 1
        par._yaml_path = yaml_path  # pylint: disable=protected-access
        return par

    def to_sapp_yaml(self) -> str:

        mem_gb = int(h.memory_per_device_mb / 1024)

        ac_mode = "full"
        if hasattr(self, "_recompute_mode"):
            ac_mode = self._recompute_mode

        doc = {
            "model": {
                "name": m.model_name,

    new_pp = strategy.pp_degree
    new_vpp = strategy.vpp_degree
    n_lay = ccfg.n_lay
    if getattr(ccfg, "emb_out_in_offset", False):
        n_lay += 2
    if getattr(ccfg, "is_mtp_in_offset", False):
        n_lay += getattr(ccfg, "n_mtp", 0)
    adapter = BalancingAdapter(
        layers=n_lay,
        offset=ccfg.offset,
        recompute=ccfg.full_rec,

        )

        _apply_strategy_to_ccfg(ccfg, strategy)
        device_type = hardware.to_sapp_machine().device
        sapp_score = sapp_estimate_performance(ccfg, device_type=device_type)

    pp = strategy.pp_degree
    mbn = strategy.micro_batch_num

    total_flops = n_lay * flops_per_layer / pp

    tflops = hardware.tflops_per_device
    if tflops <= 0:
        tflops = 300.0
    compute_ms = total_flops / (tflops * 1e9) * 1000

    comm_ms = 0.0
    if tp > 1:

    stages = []
    start = 0
    for i in range(pp):
        if offset and i < len(offset):
            num_layers = layers_per_stage + int(offset[i])
        else:
            num_layers = layers_per_stage
        end = start + num_layers
        stage_mem = strategy.memory_cost.total / pp

        ValueError: If the JSON file is missing ``layers_description``.
    """
    # pylint: disable=import-outside-toplevel
    if profiling_json_path:
        from hyper_parallel.auto_parallel.sapp_ppb.utils.layer import (
            generate_layers_list,
        )
        folder = os.path.dirname(os.path.abspath(profiling_json_path))
        base = os.path.splitext(os.path.basename(profiling_json_path))[0]
        return generate_layers_list(folder, base)

    from hyper_parallel.auto_parallel.sapp_nd.memory_estimation.estimate_v2 import (
        EvaluatorV2,
    )

    model_name = getattr(ccfg, "model_name", "dense-llm")

    layer_list = ppb_input.get("layers_description", [])
    if not layer_list:
        layer_list = ppb_input.get("layers_description_new", [])
    if not layer_list:
        return layers

    for layer_dict in layer_list:
        ltype_str = layer_dict.get("type", "BODY")
        ltype = {

    new_stage_info = []
    for i in range(min(n_stages, len(mem_act_flat))):
        n_lay = strategy.extra.get("layers_per_stage", strategy.extra.get("num_layers", 0) // n_stages)
        if isinstance(n_lay, list) and i < len(n_lay):
            nl = n_lay[i]
        else:
            nl = strategy.memory_cost.total / n_stages if n_stages > 0 else 0
        end = start + int(nl) if isinstance(nl, (int, float)) else start
        new_stage_info.append(StageInfo(

    stage_recompute = [0] * num_stages
    for layer_name in body_names:
        rec_map = variables.get(layer_name)
        if rec_map is None:
            continue
        for rec in Recompute.TYPE:
            counts = rec_map[rec]
            for i in range(num_interleave):
                for s in range(num_stages):

            number_of_micro_batch=num_micro_batch,
            max_memory=max_memory,
            layers=layers,
        )
    except Exception:
        pipeline = _solve_pipeline(
            model_name=model_name,
            num_stages=num_stages,
            num_micro_batch=num_micro_batch,
            max_memory=max_memory,

            num_micro_batch=num_micro_batch,
            max_memory=max_memory,
            layers=layers,
        )
        best_dist = pipeline.get_result()

    pipeline = _solve_pipeline(
        model_name=model_name,
        num_stages=num_stages,

        strategy, ccfg, profiling_json_path=profiling_json_path,
    )

    if not layers:
        raise RuntimeError(
            f"PPB produced no layers for strategy {strategy.key}"
        )

    max_memory = int(memory_limit_mb) if memory_limit_mb else int(strategy.memory_cost.total)

    _apply_strategy_to_ccfg(ccfg, strategy)

    try:
        layers = _build_layers_from_strategy(strategy, ccfg)
    except Exception as exc:
        logger.warning("PPB layer build failed for %s plot: %s", strategy.key, exc)
        return None

    if not layers:
        return None

    max_memory = int(memory_limit_mb) if memory_limit_mb else int(strategy.memory_cost.total)
    vpp = strategy.vpp_degree if strategy.vpp_degree > 0 else 1

        plot_path = os.path.join(output_dir, f"ppb_pipeline_{strategy.key}.svg")
        pipeline.simulate(show=False, file_name=plot_path)
        if os.path.exists(plot_path):
            return plot_path
    except Exception as exc:
        logger.warning("PPB pipeline plot failed for %s: %s", strategy.key, exc)

    return None

            f"cp{self.cp_degree}",
            f"mbn{self.micro_batch_num}",
        ]
        if self.vpp_degree > 1:
            parts.append(f"vpp{self.vpp_degree}")
        if self.ep_degree > 1:
            parts.append(f"ep{self.ep_degree}")
        if self.op_degree > 1:
            parts.append(f"op{self.op_degree}")

    def _init_sapp(self) -> None:
        """Lazily initialise SAPP-ND Parallelize instance and extract ccfg."""
        if self._parallelize is not None:
            return
        self._parallelize = self.config.create_sapp_parallelize()
        self._ccfg = self._parallelize.config.ccfg

    def tune(self) -> TunerResult:

                try:
                    nd_plot_files.extend(
                        generate_nd_memory_plots(s, self._ccfg, out)
                    )
                except Exception as exc:
                    logger.warning("ND memory plot failed: %s", exc)

        chart_files = self._visualizer.visualize(
            result.top_k_strategies,
            result.all_candidates,

            constraints: Constraint configuration.
        """
        gbs = constraints.global_batch_size
        if gbs <= 0:
            gbs = self._infer_gbs(strategies, hardware)

        for s in strategies:
            if not s.is_feasible:
                continue
            try:
                s.memory_cost = estimate_memory(s, model, gbs, ccfg=self._ccfg)
                s.performance_cost = estimate_performance(s, model, hardware, gbs, ccfg=self._ccfg)
                s.stage_info = compute_stage_info(s, model, gbs)
            except Exception as exc:
                s.mark_not_supported(f"estimation failed: {exc}")
                logger.warning("Estimation failed for %s: %s", s.key, exc)

    def _infer_gbs(
        self,
        strategies: List[ParallelStrategy],

        yaml_path = getattr(self._parallelize, "_yaml_path", None)
        if yaml_path and os.path.exists(yaml_path):
            try:
                os.unlink(yaml_path)
            except OSError:
                pass

    @staticmethod
    def _build_ppb_result(strategy: ParallelStrategy) -> Optional[PPBResult]:
        """Extract PPB result from a strategy's extra metadata.

                # pylint: disable=import-outside-toplevel
                import matplotlib
                matplotlib.use("Agg")
                self._matplotlib_available = True
            except ImportError:
                self._matplotlib_available = False
                logger.warning(
                    "matplotlib not available; falling back to text charts"
                )
        return self._matplotlib_available

        for src in (nd_plot_files or []):
            if src and os.path.exists(src):
                dst = os.path.join(out, os.path.basename(src))
                if os.path.abspath(src) != os.path.abspath(dst):
                    shutil.copy2(src, dst)
                files.append(dst)

        for src in (ppb_plot_files or []):
            if src and os.path.exists(src):

        for src in (ppb_plot_files or []):
            if src and os.path.exists(src):
                dst = os.path.join(out, os.path.basename(src))
                if os.path.abspath(src) != os.path.abspath(dst):
                    shutil.copy2(src, dst)
                files.append(dst)

        return [f for f in files if f]

        Returns:
            Output file path, or None if no strategies.
        """
        if not strategies:
            return None

        labels = [s.key for s in strategies]
        step_times = [s.performance_cost.total for s in strategies]
        path = os.path.join(out, "step_time_comparison.txt")

                img_path = os.path.join(out, "step_time_comparison.png")
                fig.savefig(img_path, dpi=150)
                plt.close(fig)
                return img_path
            except Exception as exc:
                logger.warning("matplotlib chart failed: %s", exc)

        return path

    def _memory_chart(
        self,
        strategies: List[ParallelStrategy],

        Returns:
            Output file path, or None if no strategies.
        """
        if not strategies:
            return None

        labels = [s.key for s in strategies]
        memories = [s.memory_cost.total for s in strategies]
        path = os.path.join(out, "memory_comparison.txt")

                img_path = os.path.join(out, "memory_comparison.png")
                fig.savefig(img_path, dpi=150)
                plt.close(fig)
                return img_path
            except Exception as exc:
                logger.warning("matplotlib chart failed: %s", exc)

        return path

    def _stage_distribution_chart(
        self,
        strategies: List[ParallelStrategy],

                n_strats = len(pp_strategies)
                fig, axes = plt.subplots(1, n_strats, figsize=(6 * n_strats, 4))
                if n_strats == 1:
                    axes = [axes]
                for ax, s in zip(axes, pp_strategies):
                    stages = s.stage_info
                    stage_ids = [st.stage_id for st in stages]
                    mems = [st.memory_mb for st in stages]

                img_path = os.path.join(out, "stage_distribution.png")
                fig.savefig(img_path, dpi=150)
                plt.close(fig)
                return img_path
            except Exception as exc:
                logger.warning("matplotlib chart failed: %s", exc)

        return path

    def _feasibility_chart(
        self,
        all_candidates: List[ParallelStrategy],

                f.write(f"Feasible:         {feasible} ({100 * feasible / total:.1f}%)\n")
                f.write(f"Infeasible:       {infeasible} ({100 * infeasible / total:.1f}%)\n")
                f.write(f"Not supported:    {not_supported} ({100 * not_supported / total:.1f}%)\n")
            else:
                f.write("Feasible:         0\n")
                f.write("Infeasible:       0\n")
                f.write("Not supported:    0\n")

        if self._check_matplotlib() and total > 0:
            try:
                # pylint: disable=import-outside-toplevel

                img_path = os.path.join(out, "feasibility_distribution.png")
                fig.savefig(img_path, dpi=150)
                plt.close(fig)
                return img_path
            except Exception as exc:
                logger.warning("matplotlib chart failed: %s", exc)

        return path

    def _filter_reasons_chart(
        self,
        all_candidates: List[ParallelStrategy],

        """
        reasons: Dict[str, int] = {}
        for s in all_candidates:
            if s.filter_reason:
                category = re.split(r"[=<>]", s.filter_reason)[0].strip()
                reasons[category] = reasons.get(category, 0) + 1

        path = os.path.join(out, "filter_reasons.txt")
        with open(path, "w", encoding="utf-8") as f:
            f.write("Filter Reasons Statistics\n")

        with open(path, "w", encoding="utf-8") as f:
            f.write("Filter Reasons Statistics\n")
            f.write("=" * 40 + "\n")
            for reason, count in sorted(reasons.items(), key=lambda x: -x[1]):
                f.write(f"  {reason}: {count}\n")

        if self._check_matplotlib() and reasons:
            try:
                # pylint: disable=import-outside-toplevel
                import matplotlib.pyplot as plt
                fig, ax = plt.subplots(figsize=(8, max(4, len(reasons) * 0.5)))
                sorted_reasons = sorted(reasons.items(), key=lambda x: -x[1])
                labels = [r[0][:40] for r in sorted_reasons]
                counts = [r[1] for r in sorted_reasons]
                ax.barh(range(len(labels)), counts, align="center")
                ax.set_yticks(range(len(labels)))
                ax.set_yticklabels(labels)
                ax.set_xlabel("Count")
                ax.set_title("Filter Reasons Distribution")
                fig.tight_layout()
                img_path = os.path.join(out, "filter_reasons.png")
                fig.savefig(img_path, dpi=150)
                plt.close(fig)
                return img_path
            except Exception as exc:
                logger.warning("matplotlib chart failed: %s", exc)

        return path

    def _dimension_trend_chart(

                img_path = os.path.join(out, "dimension_trends.png")
                fig.savefig(img_path, dpi=150)
                plt.close(fig)
                return img_path
            except Exception as exc:
                logger.warning("matplotlib chart failed: %s", exc)

        return path