compositing

样长组合算法（Sample Compositing / Downhole Compositing）

沿钻孔轨迹按固定长度重新切分原始样品，通过长度加权平均法 计算每个新样段的品位，解决样品承载体积不一致的问题。

Background

钻孔取样时，样品长度常因岩性变化、矿化不均匀或人为因素而不等长。 如果直接用不等长样品去估算块体品位，长样品和短样品贡献的"权重" 在统计上并不公平。样长组合的目的，就是通过长度加权，让每个样品 数据"站"在同一个起跑线上。

Core Algorithm

1. 对每个钻孔，按"起始"深度排序样品
2. 从首个样品起始深度开始，以 composite_length 为步长生成等长组合区间
3. 对每个组合区间，找到与之重叠的原始样品，计算重叠长度
4. 用长度加权平均法计算组合区间的品位值
5. 处理边界"尾巴"（不足一个完整组合长度的末段）

Key Design Decisions

• 组合起止深度：默认为每个钻孔第一个样品的"起始"深度（而非孔口0深度）
• 边界处理：末尾不足一个组合长度的区间，根据覆盖率(combine_percent)决定保留或丢弃
• 间隙处理：样品间有深度间断时，该组合区间的品位为NaN（不强行内插）
• 3D坐标：可选计算组合样段中点的三维坐标（平衡切线法，用于后续块段估值）
• 沿孔深度：所有深度均为 Measured Depth（沿钻孔轨迹），倾角不影响长度加权计算

discard_low_coverage 策略说明

• False（默认）：保留所有组合区间，低覆盖率区间通过 "低覆盖率" 列标记
• True：从孔口方向找到最后一个满足覆盖率要求的区间，丢弃其之后的所有区间 （即"尾部截断"策略）。注意：此模式下，位于截断位置之前的中间间隙区间 仍会被保留（覆盖率=0，低覆盖率=True），请在使用时通过 "低覆盖率" 列过滤。 若需要丢弃所有不合格区间，请在结果上执行 df = df[~df['低覆盖率']]。

References

• 地质统计学基本假设：样品需有相同支撑（样长一致）
• 组合样长建议为块体尺寸的 1/4 到 1/3
• 平衡"平滑效应"与计算负担

Usage

from dimine_python_sdk.lib.prospecting.compositing import ( SampleCompositingParams, SampleCompositingProcessor, composite_samples, )

便捷函数（从 DrillDBManager）

result = composite_samples( db, composite_length=2.0, grade_fields=['TFe', 'TiO2'] ) composited_df = result['composited_df']

详细参数控制

params = SampleCompositingParams( composite_length=2.0, grade_fields=['TFe', 'TiO2'], combine_percent=0.75, include_3d_coords=True, ) processor = SampleCompositingProcessor(params) result = processor.process(db.sample, db.survey, db.collar)

SampleCompositingParams dataclass

样长组合参数

Attributes:

Name	Type	Description
composite_length	float	组合样长度（米），必须 > 0。 建议取值为块体尺寸的 1/4 到 1/3。
grade_fields	List[str]	需要做长度加权平均的品位字段名列表。 字段必须存在于样品表中，且为数值类型。
combine_percent	float	最小覆盖率（0-1），组合区间内实际样品覆盖长度 低于此比例时标记为低质量组合样。 默认 0.75（与 DIMINE C++ 接口默认值一致）。
start_mode	str	组合起始模式。 - 'first_sample': 从每个钻孔第一个样品起始深度开始（默认） - 'collar': 从孔口（深度0）开始 - 数字: 从指定的绝对深度开始（如 5.0）
discard_low_coverage	bool	尾部截断模式。 为 True 时，从孔口方向找到最后一个满足覆盖率要求的区间， 丢弃其之后的所有区间（即"尾部截断"）。 注意：截断点之前的中间低覆盖率区间仍会被保留。 为 False 时保留所有区间，通过 "低覆盖率" 列标记。
include_3d_coords	bool	是否计算组合样段中点的三维坐标。 需要提供 survey 和 collar 表。
min_samples_per_composite	int	每个组合样最少包含的原始样品数（警告阈值）。
auto_detect_grade_fields	bool	是否自动检测样品表中的数值型品位字段。 为 True 时，grade_fields 中未列出的数值列也会被组合。 自动排除标准列（工程号、样本编号、坐标等）。

Source code in dimine_python_sdk\lib\prospecting\compositing.py

@dataclass
class SampleCompositingParams:
    """样长组合参数

    Attributes:
        composite_length: 组合样长度（米），必须 > 0。
                          建议取值为块体尺寸的 1/4 到 1/3。
        grade_fields: 需要做长度加权平均的品位字段名列表。
                      字段必须存在于样品表中，且为数值类型。
        combine_percent: 最小覆盖率（0-1），组合区间内实际样品覆盖长度
                         低于此比例时标记为低质量组合样。
                         默认 0.75（与 DIMINE C++ 接口默认值一致）。
        start_mode: 组合起始模式。
                     - 'first_sample': 从每个钻孔第一个样品起始深度开始（默认）
                     - 'collar': 从孔口（深度0）开始
                     - 数字: 从指定的绝对深度开始（如 5.0）
        discard_low_coverage: 尾部截断模式。
                              为 True 时，从孔口方向找到最后一个满足覆盖率要求的区间，
                              丢弃其之后的所有区间（即"尾部截断"）。
                              注意：截断点之前的中间低覆盖率区间仍会被保留。
                              为 False 时保留所有区间，通过 "低覆盖率" 列标记。
        include_3d_coords: 是否计算组合样段中点的三维坐标。
                           需要提供 survey 和 collar 表。
        min_samples_per_composite: 每个组合样最少包含的原始样品数（警告阈值）。
        auto_detect_grade_fields: 是否自动检测样品表中的数值型品位字段。
                                  为 True 时，grade_fields 中未列出的数值列也会被组合。
                                  自动排除标准列（工程号、样本编号、坐标等）。
    """

    composite_length: float = 1.0
    grade_fields: List[str] = field(default_factory=list)
    combine_percent: float = 0.75
    start_mode: str = "first_sample"  # 'first_sample' | 'collar' | numeric string
    discard_low_coverage: bool = False
    include_3d_coords: bool = True
    min_samples_per_composite: int = 1
    auto_detect_grade_fields: bool = True

    _VALID_START_MODES = frozenset({"first_sample", "collar"})

    def __post_init__(self):
        if self.composite_length <= 0:
            raise ValueError(
                f"composite_length 必须 > 0，当前值: {self.composite_length}"
            )
        if not (0 <= self.combine_percent <= 1):
            raise ValueError(
                f"combine_percent 必须在 0-1 之间，当前值: {self.combine_percent}"
            )
        # 校验 start_mode：允许 'first_sample'、'collar'，或可转为 float 的字符串
        if (
            isinstance(self.start_mode, str)
            and self.start_mode not in self._VALID_START_MODES
        ):
            try:
                float(self.start_mode)
            except ValueError:
                raise ValueError(
                    f"start_mode 必须是 'first_sample'、'collar' 或数字，"
                    f"当前值: '{self.start_mode}'"
                )

SampleCompositingProcessor

样长组合处理器

对钻孔样品表执行样长组合操作。核心流程：

1. 数据预处理（NaN 清理、类型转换、数据校验）
2. 按钻孔（工程号）分组
3. 对每个钻孔独立执行长度加权组合
4. 汇总所有钻孔的组合结果
5. 生成统计报告

Usage

params = SampleCompositingParams( composite_length=2.0, grade_fields=['TFe', 'TiO2'], ) processor = SampleCompositingProcessor(params) result = processor.process(sample_df, survey_df, collar_df)

Source code in dimine_python_sdk\lib\prospecting\compositing.py

class SampleCompositingProcessor:
    """样长组合处理器

    对钻孔样品表执行样长组合操作。核心流程：

        1. 数据预处理（NaN 清理、类型转换、数据校验）
        2. 按钻孔（工程号）分组
        3. 对每个钻孔独立执行长度加权组合
        4. 汇总所有钻孔的组合结果
        5. 生成统计报告

    Usage:
        params = SampleCompositingParams(
            composite_length=2.0,
            grade_fields=['TFe', 'TiO2'],
        )
        processor = SampleCompositingProcessor(params)
        result = processor.process(sample_df, survey_df, collar_df)
    """

    def __init__(self, params: Optional[SampleCompositingParams] = None, **kwargs):
        """
        Args:
            params: SampleCompositingParams 参数对象。
                    为 None 时从 **kwargs 构建。
            **kwargs: 直接传入 SampleCompositingParams 的参数。
        """
        if params is None:
            params = SampleCompositingParams(**kwargs)
        self.params = params

    # ------------------------------------------------------------------
    # 主入口
    # ------------------------------------------------------------------

    def process(
        self,
        sample_df: pd.DataFrame,
        survey_df: Optional[pd.DataFrame] = None,
        collar_df: Optional[pd.DataFrame] = None,
    ) -> Dict[str, Any]:
        """
        执行样长组合。

        Args:
            sample_df: 样品表 DataFrame（中文标准列名：工程号, 起始, 结束, ...）
            survey_df: 测斜表 DataFrame（可选，用于计算3D坐标）
            collar_df: 孔口表 DataFrame（可选，用于计算3D坐标）

        Returns:
            (composited_df, summary) 元组:
                composited_df — 组合后的样品 DataFrame
                summary      — 汇总统计 dict
        """
        p = self.params

        # --- 拷贝输入 ---
        samples = sample_df.copy()

        # --- 数据类型预处理 ---
        # 确保起始/结束为数值类型，转换失败的 NaN 提前清理
        for col in ["起始", "结束"]:
            if col in samples.columns:
                samples[col] = pd.to_numeric(samples[col], errors="coerce")

        # 提前统计并报告因转换失败而产生的 NaN
        nan_before = len(samples)
        samples = samples.dropna(subset=["起始", "结束"])
        nan_dropped = nan_before - len(samples)
        if nan_dropped > 0:
            _logger.warning(
                f"样品表中 {nan_dropped} 行因 '起始'/'结束' 为非数值而被丢弃。"
                f"原始行数: {nan_before}, 保留行数: {len(samples)}"
            )

        # 过滤 工程号 为空的行
        samples = samples.dropna(subset=["工程号"])
        samples["工程号"] = samples["工程号"].astype(str)

        # 空数据快速返回
        if len(samples) == 0:
            return pd.DataFrame(), {"原始样品总数": 0, "组合样品总数": 0}

        # --- 自动检测品位字段 ---
        grade_fields = list(p.grade_fields)
        if p.auto_detect_grade_fields:
            for col in samples.columns:
                if col not in _STANDARD_SAMPLE_COLS and col not in grade_fields:
                    if pd.api.types.is_numeric_dtype(samples[col]):
                        grade_fields.append(col)

        if not grade_fields:
            raise ValueError(
                "未指定品位字段，且样品表中未检测到数值型品位列。"
                "请通过 grade_fields 参数明确指定。"
            )

        # 验证品位字段存在
        missing_fields = [gf for gf in grade_fields if gf not in samples.columns]
        if missing_fields:
            raise ValueError(
                f"品位字段 {missing_fields} 在样品表中不存在。"
                f"可用列: {list(samples.columns)}"
            )

        # --- 构建孔口信息查找表 ---
        collar_lookup: Dict[str, Dict[str, float]] = {}
        if collar_df is not None:
            for _, row in collar_df.iterrows():
                hole_id = str(row.get("工程号", ""))
                collar_lookup[hole_id] = {
                    "横坐标": float(row.get("横坐标", 0)),
                    "纵坐标": float(row.get("纵坐标", 0)),
                    "高程": float(row.get("高程", 0)),
                }

        # --- 构建测斜数据查找表 ---
        survey_lookup: Dict[str, pd.DataFrame] = {}
        if survey_df is not None:
            for hole_id, group in survey_df.groupby("工程号"):
                survey_lookup[str(hole_id)] = group.copy()

        # --- 确定起始深度 ---
        start_depth_override = None
        if isinstance(p.start_mode, (int, float)):
            start_depth_override = float(p.start_mode)
        elif isinstance(p.start_mode, str):
            try:
                start_depth_override = float(p.start_mode)
            except ValueError:
                pass  # 不是数字字符串，保持为 None

        # --- 按钻孔分组处理 ---
        hole_ids = sorted(samples["工程号"].unique())
        all_composited: List[pd.DataFrame] = []
        all_stats: List[Dict[str, Any]] = []
        hole_data: Dict[str, pd.DataFrame] = {}
        all_validation_warnings: List[str] = []
        total_original = 0
        total_composited = 0

        for hole_id in hole_ids:
            hole_samples = (
                samples[samples["工程号"] == hole_id].sort_values("起始").copy()
            )
            if len(hole_samples) == 0:
                continue

            collar_info = collar_lookup.get(str(hole_id))
            survey_hole = survey_lookup.get(str(hole_id))

            comp_df, stats = _composite_single_hole(
                hole_id=str(hole_id),
                samples=hole_samples,
                composite_length=p.composite_length,
                grade_fields=grade_fields,
                combine_percent=p.combine_percent,
                start_mode=p.start_mode,
                start_depth_override=start_depth_override,
                discard_low_coverage=p.discard_low_coverage,
                collar_info=collar_info,
                survey_hole=survey_hole,
                include_3d=p.include_3d_coords,
            )

            total_original += stats.get("n_original", 0)
            total_composited += stats.get("n_composited", 0)
            all_stats.append(stats)

            if not comp_df.empty:
                all_composited.append(comp_df)
                hole_data[str(hole_id)] = comp_df

        # --- 汇总 ---
        if all_composited:
            composited_df = pd.concat(all_composited, ignore_index=True)
        else:
            cols = ["工程号", "样本编号", "起始", "结束", "样长"] + grade_fields
            if p.include_3d_coords:
                cols += ["X", "Y", "Z"]
            composited_df = pd.DataFrame(columns=cols)

        # --- 统计 ---
        per_hole_df = pd.DataFrame(all_stats) if all_stats else pd.DataFrame()

        summary: Dict[str, Any] = {
            "总钻孔数": len(hole_ids),
            "有效钻孔数": len([s for s in all_stats if s.get("n_composited", 0) > 0]),
            "原始样品总数": total_original,
            "组合样品总数": total_composited,
            "压缩比": (
                f"{total_original} → {total_composited}"
                if total_composited > 0
                else "N/A"
            ),
            "组合长度": p.composite_length,
            "最小覆盖率": p.combine_percent,
            "起始模式": p.start_mode,
            "品位字段": grade_fields,
        }

        if not per_hole_df.empty:
            low_cov_total = per_hole_df["low_coverage_count"].sum()
            summary["低覆盖率样本数"] = int(low_cov_total)
            if "mean_coverage" in per_hole_df.columns:
                summary["平均覆盖率"] = float(per_hole_df["mean_coverage"].mean())
            val_warn_total = per_hole_df["validation_warnings"].sum()
            if val_warn_total > 0:
                summary["数据质量警告数"] = int(val_warn_total)

        return composited_df, summary

__init__(params=None, **kwargs)

Parameters:

Name	Type	Description	Default
params	Optional[SampleCompositingParams]	SampleCompositingParams 参数对象。 为 None 时从 **kwargs 构建。	None
**kwargs		直接传入 SampleCompositingParams 的参数。	{}

Source code in dimine_python_sdk\lib\prospecting\compositing.py

def __init__(self, params: Optional[SampleCompositingParams] = None, **kwargs):
    """
    Args:
        params: SampleCompositingParams 参数对象。
                为 None 时从 **kwargs 构建。
        **kwargs: 直接传入 SampleCompositingParams 的参数。
    """
    if params is None:
        params = SampleCompositingParams(**kwargs)
    self.params = params

process(sample_df, survey_df=None, collar_df=None)

执行样长组合。

Parameters:

Name	Type	Description	Default
sample_df	DataFrame	样品表 DataFrame（中文标准列名：工程号, 起始, 结束, ...）	required
survey_df	Optional[DataFrame]	测斜表 DataFrame（可选，用于计算3D坐标）	None
collar_df	Optional[DataFrame]	孔口表 DataFrame（可选，用于计算3D坐标）	None

Returns:

Type	Description
Dict[str, Any]	(composited_df, summary) 元组: composited_df — 组合后的样品 DataFrame summary — 汇总统计 dict

Source code in dimine_python_sdk\lib\prospecting\compositing.py

def process(
    self,
    sample_df: pd.DataFrame,
    survey_df: Optional[pd.DataFrame] = None,
    collar_df: Optional[pd.DataFrame] = None,
) -> Dict[str, Any]:
    """
    执行样长组合。

    Args:
        sample_df: 样品表 DataFrame（中文标准列名：工程号, 起始, 结束, ...）
        survey_df: 测斜表 DataFrame（可选，用于计算3D坐标）
        collar_df: 孔口表 DataFrame（可选，用于计算3D坐标）

    Returns:
        (composited_df, summary) 元组:
            composited_df — 组合后的样品 DataFrame
            summary      — 汇总统计 dict
    """
    p = self.params

    # --- 拷贝输入 ---
    samples = sample_df.copy()

    # --- 数据类型预处理 ---
    # 确保起始/结束为数值类型，转换失败的 NaN 提前清理
    for col in ["起始", "结束"]:
        if col in samples.columns:
            samples[col] = pd.to_numeric(samples[col], errors="coerce")

    # 提前统计并报告因转换失败而产生的 NaN
    nan_before = len(samples)
    samples = samples.dropna(subset=["起始", "结束"])
    nan_dropped = nan_before - len(samples)
    if nan_dropped > 0:
        _logger.warning(
            f"样品表中 {nan_dropped} 行因 '起始'/'结束' 为非数值而被丢弃。"
            f"原始行数: {nan_before}, 保留行数: {len(samples)}"
        )

    # 过滤 工程号 为空的行
    samples = samples.dropna(subset=["工程号"])
    samples["工程号"] = samples["工程号"].astype(str)

    # 空数据快速返回
    if len(samples) == 0:
        return pd.DataFrame(), {"原始样品总数": 0, "组合样品总数": 0}

    # --- 自动检测品位字段 ---
    grade_fields = list(p.grade_fields)
    if p.auto_detect_grade_fields:
        for col in samples.columns:
            if col not in _STANDARD_SAMPLE_COLS and col not in grade_fields:
                if pd.api.types.is_numeric_dtype(samples[col]):
                    grade_fields.append(col)

    if not grade_fields:
        raise ValueError(
            "未指定品位字段，且样品表中未检测到数值型品位列。"
            "请通过 grade_fields 参数明确指定。"
        )

    # 验证品位字段存在
    missing_fields = [gf for gf in grade_fields if gf not in samples.columns]
    if missing_fields:
        raise ValueError(
            f"品位字段 {missing_fields} 在样品表中不存在。"
            f"可用列: {list(samples.columns)}"
        )

    # --- 构建孔口信息查找表 ---
    collar_lookup: Dict[str, Dict[str, float]] = {}
    if collar_df is not None:
        for _, row in collar_df.iterrows():
            hole_id = str(row.get("工程号", ""))
            collar_lookup[hole_id] = {
                "横坐标": float(row.get("横坐标", 0)),
                "纵坐标": float(row.get("纵坐标", 0)),
                "高程": float(row.get("高程", 0)),
            }

    # --- 构建测斜数据查找表 ---
    survey_lookup: Dict[str, pd.DataFrame] = {}
    if survey_df is not None:
        for hole_id, group in survey_df.groupby("工程号"):
            survey_lookup[str(hole_id)] = group.copy()

    # --- 确定起始深度 ---
    start_depth_override = None
    if isinstance(p.start_mode, (int, float)):
        start_depth_override = float(p.start_mode)
    elif isinstance(p.start_mode, str):
        try:
            start_depth_override = float(p.start_mode)
        except ValueError:
            pass  # 不是数字字符串，保持为 None

    # --- 按钻孔分组处理 ---
    hole_ids = sorted(samples["工程号"].unique())
    all_composited: List[pd.DataFrame] = []
    all_stats: List[Dict[str, Any]] = []
    hole_data: Dict[str, pd.DataFrame] = {}
    all_validation_warnings: List[str] = []
    total_original = 0
    total_composited = 0

    for hole_id in hole_ids:
        hole_samples = (
            samples[samples["工程号"] == hole_id].sort_values("起始").copy()
        )
        if len(hole_samples) == 0:
            continue

        collar_info = collar_lookup.get(str(hole_id))
        survey_hole = survey_lookup.get(str(hole_id))

        comp_df, stats = _composite_single_hole(
            hole_id=str(hole_id),
            samples=hole_samples,
            composite_length=p.composite_length,
            grade_fields=grade_fields,
            combine_percent=p.combine_percent,
            start_mode=p.start_mode,
            start_depth_override=start_depth_override,
            discard_low_coverage=p.discard_low_coverage,
            collar_info=collar_info,
            survey_hole=survey_hole,
            include_3d=p.include_3d_coords,
        )

        total_original += stats.get("n_original", 0)
        total_composited += stats.get("n_composited", 0)
        all_stats.append(stats)

        if not comp_df.empty:
            all_composited.append(comp_df)
            hole_data[str(hole_id)] = comp_df

    # --- 汇总 ---
    if all_composited:
        composited_df = pd.concat(all_composited, ignore_index=True)
    else:
        cols = ["工程号", "样本编号", "起始", "结束", "样长"] + grade_fields
        if p.include_3d_coords:
            cols += ["X", "Y", "Z"]
        composited_df = pd.DataFrame(columns=cols)

    # --- 统计 ---
    per_hole_df = pd.DataFrame(all_stats) if all_stats else pd.DataFrame()

    summary: Dict[str, Any] = {
        "总钻孔数": len(hole_ids),
        "有效钻孔数": len([s for s in all_stats if s.get("n_composited", 0) > 0]),
        "原始样品总数": total_original,
        "组合样品总数": total_composited,
        "压缩比": (
            f"{total_original} → {total_composited}"
            if total_composited > 0
            else "N/A"
        ),
        "组合长度": p.composite_length,
        "最小覆盖率": p.combine_percent,
        "起始模式": p.start_mode,
        "品位字段": grade_fields,
    }

    if not per_hole_df.empty:
        low_cov_total = per_hole_df["low_coverage_count"].sum()
        summary["低覆盖率样本数"] = int(low_cov_total)
        if "mean_coverage" in per_hole_df.columns:
            summary["平均覆盖率"] = float(per_hole_df["mean_coverage"].mean())
        val_warn_total = per_hole_df["validation_warnings"].sum()
        if val_warn_total > 0:
            summary["数据质量警告数"] = int(val_warn_total)

    return composited_df, summary

composite_samples(db_or_samples, composite_length=2.0, grade_fields=None, combine_percent=0.75, discard_low_coverage=False, start_mode='first_sample', include_3d_coords=True, auto_detect_grade_fields=True, survey_df=None, collar_df=None)

样长组合便捷函数。

支持两种输入模式： 1. 从 DrillDBManager 直接传入： result = composite_samples(db, composite_length=2.0, grade_fields=['TFe']) 2. 从独立的 DataFrame 传入： result = composite_samples( sample_df, composite_length=2.0, grade_fields=['TFe', 'TiO2'], survey_df=survey_df, collar_df=collar_df )

discard_low_coverage 策略： - False（默认）：保留所有组合区间，通过 "低覆盖率" 列标记 - True：尾部截断——找到最后一个合格区间，丢弃其之后的所有区间。 如需丢弃所有不合格区间（包括中间的），请在结果上执行： df = df[~df['低覆盖率']]

Parameters:

Name	Type	Description	Default
db_or_samples		DrillDBManager 实例 或 样品表 pd.DataFrame	required
composite_length	float	组合样长度（米），默认 2.0	2.0
grade_fields	Optional[List[str]]	品位字段列表。为 None 时自动检测数值列	None
combine_percent	float	最小覆盖率（0-1），默认 0.75	0.75
discard_low_coverage	bool	尾部截断模式，默认 False	False
start_mode	str	起始模式，默认 'first_sample'	'first_sample'
include_3d_coords	bool	是否计算3D坐标	True
auto_detect_grade_fields	bool	是否自动检测品位字段	True
survey_df	Optional[DataFrame]	测斜表（仅当 db_or_samples 为 DataFrame 时需要）	None
collar_df	Optional[DataFrame]	孔口表（仅当 db_or_samples 为 DataFrame 时需要）	None

Returns:

Type	Description
Dict[str, Any]	(composited_df, summary) 元组

Source code in dimine_python_sdk\lib\prospecting\compositing.py

def composite_samples(
    db_or_samples,
    composite_length: float = 2.0,
    grade_fields: Optional[List[str]] = None,
    combine_percent: float = 0.75,
    discard_low_coverage: bool = False,
    start_mode: str = "first_sample",
    include_3d_coords: bool = True,
    auto_detect_grade_fields: bool = True,
    survey_df: Optional[pd.DataFrame] = None,
    collar_df: Optional[pd.DataFrame] = None,
) -> Dict[str, Any]:
    """
    样长组合便捷函数。

    支持两种输入模式：
        1. 从 DrillDBManager 直接传入：
           result = composite_samples(db, composite_length=2.0, grade_fields=['TFe'])
        2. 从独立的 DataFrame 传入：
           result = composite_samples(
               sample_df, composite_length=2.0, grade_fields=['TFe', 'TiO2'],
               survey_df=survey_df, collar_df=collar_df
           )

    discard_low_coverage 策略：
        - False（默认）：保留所有组合区间，通过 "低覆盖率" 列标记
        - True：尾部截断——找到最后一个合格区间，丢弃其之后的所有区间。
          如需丢弃所有不合格区间（包括中间的），请在结果上执行：
          df = df[~df['低覆盖率']]

    Args:
        db_or_samples: DrillDBManager 实例 或 样品表 pd.DataFrame
        composite_length: 组合样长度（米），默认 2.0
        grade_fields: 品位字段列表。为 None 时自动检测数值列
        combine_percent: 最小覆盖率（0-1），默认 0.75
        discard_low_coverage: 尾部截断模式，默认 False
        start_mode: 起始模式，默认 'first_sample'
        include_3d_coords: 是否计算3D坐标
        auto_detect_grade_fields: 是否自动检测品位字段
        survey_df: 测斜表（仅当 db_or_samples 为 DataFrame 时需要）
        collar_df: 孔口表（仅当 db_or_samples 为 DataFrame 时需要）

    Returns:
        (composited_df, summary) 元组
    """
    from dimine_python_sdk.lib.prospecting.drill_db import DrillDBManager

    if isinstance(db_or_samples, DrillDBManager):
        db = db_or_samples
        sample_df = db.sample.copy()
        if survey_df is None:
            survey_df = db.survey.copy()
        if collar_df is None:
            collar_df = db.collar.copy()
    elif isinstance(db_or_samples, pd.DataFrame):
        sample_df = db_or_samples.copy()
    else:
        raise TypeError(
            f"db_or_samples 必须是 DrillDBManager 或 pd.DataFrame，"
            f"当前类型: {type(db_or_samples)}"
        )

    params = SampleCompositingParams(
        composite_length=composite_length,
        grade_fields=grade_fields or [],
        combine_percent=combine_percent,
        start_mode=start_mode,
        discard_low_coverage=discard_low_coverage,
        include_3d_coords=include_3d_coords,
        auto_detect_grade_fields=auto_detect_grade_fields,
    )

    processor = SampleCompositingProcessor(params)
    return processor.process(sample_df, survey_df, collar_df)