conn_client

ConnClient dataclass

Bases: BaseConnClient

Source code in dimine_python_sdk\conn\conn_client.py

class ConnClient(BaseConnClient):

    async def get_files(self) -> List[FileInfo]:
        """获取文件对象树"""
        res = await self._send_command(CommandType.GET_FILE_TREE)
        if res.status == ResponseStatus.FAIL:
            raise Exception(res.error)
        return [FileInfo(**item, related=True) for item in res.payload]

    async def close_file(self, file_id: str):
        """关闭文件"""
        res = await self._send_command(CommandType.FILE_CLOSE, {"file": file_id})
        if res.status == ResponseStatus.FAIL:
            raise Exception(res.error)

    async def create_file(self, file_path: str) -> None:
        """在 Dimine 中新建文件

        Args:
            file_path: 新建文件的完整路径

        Raises:
            Exception: 创建失败时抛出异常
        """
        res = await self._send_command(CommandType.FILE_CREATE, {"file_path": file_path})
        if res.status == ResponseStatus.FAIL:
            raise Exception(res.error)

    async def open_file(self, file_path: str| Path) -> FileInfo:
        """打开文件"""
        if isinstance(file_path, str):  # 如果是 str 对象，则转换为 Path
            file_path = Path(file_path)
            if not file_path.exists():  # 如果文件不存在，报错
                raise Exception(f"file {file_path} not exists")
        res = await self._send_command(CommandType.FILE_OPEN, {"file_path": str(file_path)})
        if res.status == ResponseStatus.FAIL:
            raise Exception(res.error)
        if res.payload:
            return FileInfo(id=res.payload[0].get("file_id"), name=res.payload[0].get("file_id"), related=True)
        else:
            raise Exception("file already opened!")


    async def export_file(self, file_id: str, file_path: str| Path, mkdir: bool = True):
        """导出文件

        Args:
            file_id: 文件ID或文件名称
            file_path: 导出目标路径
            mkdir: 是否自动创建目录
        Raises:
            Exception: 导出失败时抛出异常
        """
        if isinstance(file_path, str):  # 如果是 str 对象，则转换为 Path
            file_path = Path(file_path)
            if not file_path.parent.exists():  # 如果目录不存在，报错
                if mkdir:
                    file_path.parent.mkdir(parents=True, exist_ok=True)
                else:
                    raise Exception(f"path {file_path.parent} not exists")
        res = await self._send_command(CommandType.FILE_EXPORT, {"file": file_id, "file_path": str(file_path)})
        if res.status == ResponseStatus.FAIL:
            raise Exception(res.error)

# 模型相关方法
    async def create_geometry(
        self,
        geometries: List[GEOMETRY],
        *,
        file_transfer: bool | Literal["auto"] = "auto",
        file_transfer_threshold: int = 10000,
    ) -> None:
        """创建模型

        Args:
            geometries: 模型列表，每个模型包含以下字段:
                - file: 文件名称
                - layer: 图层名称
                - feature: 要素列名称
                - entity_type: "point | line | polyline | shell"
                - geometry: 几何信息 (coordinates)
                - properties: 属性信息列表
                - color: 颜色 [R, G, B]
            file_transfer: 是否通过本地临时 DMF 文件传输大模型。
                "auto"（默认）时根据 ``file_transfer_threshold`` 自动判断；
                ``True`` 时全部走文件路径；``False`` 时全部走原 WebSocket 坐标路径。
            file_transfer_threshold: 顶点/面片数阈值，默认 10000。

        Raises:
            Exception: 创建失败时抛出异常
        """
        if not geometries:
            return

        small_geometries = []
        large_geometries = []
        for geom in geometries:
            if self._should_use_file_transfer(geom, file_transfer, file_transfer_threshold):
                large_geometries.append(geom)
            else:
                small_geometries.append(geom)

        # 小模型保持原有批量 WebSocket 发送逻辑
        if small_geometries:
            payload = [self._format_geometry_payload(g) for g in small_geometries]
            response = await self._send_command(CommandType.CREATE_GEOMETRY, payload)
            if response.status == ResponseStatus.FAIL:
                raise Exception(response.error)
            for i, item in enumerate(response.payload or []):
                small_geometries[i].id = item.get("id")
                small_geometries[i].related = True

        # 大模型逐个生成临时 DMF 文件并通过 file_path 发送
        if large_geometries:
            tmp_dir = Path(tempfile.mkdtemp(prefix="dimine_create_geometry_"))
            try:
                await self._create_geometry_via_file(large_geometries, tmp_dir)
            finally:
                shutil.rmtree(tmp_dir, ignore_errors=True)

    def _should_use_file_transfer(
        self,
        geom: GEOMETRY,
        file_transfer: bool | Literal["auto"],
        threshold: int,
    ) -> bool:
        """判断单个几何体是否应走本地文件传输路径。"""
        if file_transfer is True:
            return True
        if file_transfer is False:
            return False

        # 默认 "auto"：按顶点/面片数阈值判断
        if isinstance(geom, Point):
            return False
        if isinstance(geom, Line):
            return np.asarray(geom.geometry).shape[0] > threshold
        if isinstance(geom, Shell):
            tin = geom.geometry
            return (
                np.asarray(tin.points).shape[0] > threshold
                or np.asarray(tin.faces).shape[0] > threshold
            )
        return False

    def _format_geometry_payload(self, geom: GEOMETRY, *, include_geometry: bool = True) -> Dict[str, Any]:
        """将几何体格式化为 ``geometry.create`` 消息 payload。"""
        payload: Dict[str, Any] = {
            "file": geom.file,
            "layer": geom.layer,
            "feature": geom.feature or "0",
            "entity_type": geom.type,
            "color": geom.color,
            "properties": [
                {"name": prop.name, "value": prop.value}
                for prop in geom.properties
            ],
        }
        if include_geometry:
            if geom.type == "shell":
                payload["geometry"] = {
                    "indexs": geom.geometry.faces.tolist(),
                    "points": geom.geometry.points.tolist(),
                }
            else:
                payload["geometry"] = geom.geometry.tolist()
        return payload

    def _entity_type_for_file_transfer(self, geom: GEOMETRY) -> str:
        """为文件传输路径计算与 DMF 实体一致的 entity_type。"""
        if isinstance(geom, Point):
            return "point"
        if isinstance(geom, Line):
            coords = np.asarray(geom.geometry)
            return "line" if coords.shape[0] == 2 else "polyline"
        if isinstance(geom, Shell):
            return "shell"
        return geom.type or "line"

    def _insert_geometry_into_dmf(self, dmf, geom: GEOMETRY) -> None:
        """将几何体写入 DMF 内存模型（图层名为 ``"0"``）。"""
        name = geom.feature or "0"
        if isinstance(geom, Point):
            dmf.insert_point("0", geom.geometry, name=name)
        elif isinstance(geom, Line):
            coords = np.asarray(geom.geometry, dtype=np.float64)
            if coords.shape[0] == 2:
                dmf.insert_line("0", coords[0], coords[1], name=name)
            else:
                dmf.insert_polyline("0", coords, name=name)
        elif isinstance(geom, Shell):
            dmf.insert_shell(
                "0",
                geom.geometry.points,
                geom.geometry.faces,
                name=name,
            )

    def _get_dmf_file_class(self):
        """懒加载 ``DmfFile``，便于测试注入。"""
        try:
            from dimine_python_sdk.lib.io.dmf_file import DmfFile
        except ImportError as exc:
            raise RuntimeError(
                "文件传输方式创建几何需要本地 DmPyBindInterface 扩展，但当前环境无法导入 DmfFile"
            ) from exc
        return DmfFile

    async def _create_geometry_via_file(
        self,
        geometries: List[GEOMETRY],
        tmp_dir: Path,
    ) -> None:
        """将大模型逐个写入临时 DMF 并通过 ``file_path`` 发送创建命令。"""
        DmfFile = self._get_dmf_file_class()

        for idx, geom in enumerate(geometries):
            dmf = DmfFile()
            dmf.add_layer("0")
            self._insert_geometry_into_dmf(dmf, geom)

            temp_path = tmp_dir / f"geom_{idx}.dmf"
            dmf.save(str(temp_path))

            payload = self._format_geometry_payload(geom, include_geometry=False)
            payload["entity_type"] = self._entity_type_for_file_transfer(geom)
            payload["file_path"] = str(temp_path)

            response = await self._send_command(CommandType.CREATE_GEOMETRY, payload)
            if response.status == ResponseStatus.FAIL:
                raise Exception(response.error)

            returned = response.payload[0] if isinstance(response.payload, list) else response.payload
            if isinstance(returned, dict):
                geom.id = returned.get("id")
                geom.related = True

    async def get_geometry(self, file_id: str, layer_id: str, entity_types: List[str]=[], features: Optional[List[str]] = None, with_geometry: bool = False) -> List[GEOMETRY]:
        """获取模型

        Args:
            file_id: 文件ID
            layer_id: 图层名称
            entity_types: 获取哪些类型的模型
            features: 获取哪些要素相关的模型
            with_geometry: 是否返回模型坐标信息，默认 False
        """
        # 先获取要素信息，
        all_features = await self.get_features(file_id)

        def _get_property_type(feature_name:str, name: str)-> str:
            """获取属性类型"""
            for feature in all_features:
                if feature.name == feature_name:
                    for property in feature.properties:
                        if property.name == name:
                            return property.type

        payload = {
            "file": file_id,
            "layer": layer_id,
            "entity_type": entity_types,
            "with_geometry": with_geometry,
        }
        if features:
            payload["feature"] = features

        response = await self._send_command(CommandType.GET_GEOMETRY, payload)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)

        def _format_value(type:str, value: Any):
            if value == "" and type not in ("string", "char"):
                return None
            elif type == "double":
                return float(value)
            elif type in ("string", "char"):
                return str(value)
            elif type in ["int", "long", "short", "byte"]:
                return int(value)
            elif type == "float":
                return float(value)
            elif type == "color":
                return int(value)
            elif type == "date":
                return str(value)
            elif type == "binary":
                return value
            else:
                raise Exception(f"unknown property type: {type}")

        def _handle_property(feature_name:str, properties: List[Dict[str, Any]]):
            for property in properties:
                property["type"] = _get_property_type(feature_name, property.get("name"))
                property["value"] = _format_value(property.get("type"), property.get("value"))

        geometries = []
        for item in response.payload:
            _handle_property(item.get("feature"), item.get("properties", []))
            if isinstance(item.get("geometry", []), list):
                tdata = np.asarray(item.get("geometry", [])).astype(np.float64)
                if len(tdata.shape) == 1:
                    item["geometry"] = tdata
                    geometries.append(Point(**item, related=True))
                elif len(tdata.shape) == 2:
                    geometries.append(Line(**item, related=True))
            elif isinstance(item.get("geometry", []), dict):
                item["geometry"]["faces"] = item["geometry"].pop("indexs") if "indexs" in item["geometry"] else []
                geometries.append(Shell(**item, related=True))
        for geom in geometries:
            geom.file = file_id
            geom.layer = layer_id
        return geometries

    async def update_geometry(self, geometries: List[GEOMETRY]) -> None:
        """更新模型

        Args:
            geometries: 模型列表，每个模型包含以下字段:
                - id: 模型ID (句柄ID)
                - layer: 图层名称
                - entity_type: "point | line | polyline | shell"
                - geometry: 模型几何信息 (可选)
                - properties: 模型属性信息 (可选)
                - color: 颜色 [R, G, B] (可选)
                - feature: 模型关联要素列表 (可选)

        Raises:
            Exception: 更新失败时抛出异常
        """
        # 格式化模型数据
        formatted_geometries = []
        for geom in geometries:
            formatted_geom = {
                "id": geom.id,
                "file": geom.file,
                "layer": geom.layer,
                "feature": geom.feature,
                "entity_type": geom.type,
                "geometry": geom.geometry.tolist() if geom.type!="shell" else {"indexs": geom.geometry.faces.tolist(), "points": geom.geometry.points.tolist()},
                "color": geom.color,
            }

            # 处理属性信息
            formatted_geom["properties"] = [
                    {"name": prop.name, "value": prop.value}
                    for prop in geom.properties
                ]
            formatted_geometries.append(formatted_geom)

        response = await self._send_command(CommandType.UPDATE_GEOMETRY, formatted_geometries)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)

    async def delete_geometry(self, delete_items: List[Dict[str, str]]) -> None:
        """删除模型

        Args:
            delete_items: 删除项列表，每个项包含:
                - id: 模型ID (句柄ID)
                - file: 文件ID (从哪个文件删除模型)

        Raises:
            Exception: 删除失败时抛出异常
        """
        response = await self._send_command(CommandType.DELETE_GEOMETRY, delete_items)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)



    async def update_or_create_geometry(
        self,
        geometries: List[GEOMETRY] | GEOMETRY,
        *,
        file_transfer: bool | Literal["auto"] = "auto",
        file_transfer_threshold: int = 10000,
    ) -> None:
        """创建或更新多个模型, 如果指定了file_id和layer_id, 则将模型添加到指定的文件及图层中"""
        if isinstance(geometries, (Point, Line, Shell)):
            geometries = [geometries]
        update_geometries = [g for g in geometries if g.related]
        create_geometries = [g for g in geometries if not g.related]
        if update_geometries:
            await self.update_geometry(update_geometries)
        if create_geometries:
            await self.create_geometry(
                create_geometries,
                file_transfer=file_transfer,
                file_transfer_threshold=file_transfer_threshold,
            )


    # 图层相关方法
    async def create_layer(self, file_id: str, name: str) -> LayerInfo:
        """创建图层

        Args:
            file_id: 文件ID或文件名称
            name: 图层名称

        Returns:
            创建的图层信息

        Raises:
            Exception: 创建失败时抛出异常
        """
        payload = {
            "file": file_id,
            "name": name
        }
        response = await self._send_command(CommandType.CREATE_LAYER, payload)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)
        return LayerInfo(id=name, name=name, related=True)


    async def get_layers(self, file_id: str) -> List[LayerInfo]:
        """获取图层"""
        response = await self._send_command(CommandType.GET_LAYERS, {"file": file_id})
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)
        return [LayerInfo(**item, related=True) for item in response.payload]

    async def delete_layer(self, file_id: str, layer_name: str) -> None:
        """删除图层

        Args:
            file_id: 文件ID或文件名称
            layer_name: 图层名称

        Raises:
            Exception: 删除失败时抛出异常
        """
        payload = {
            "file": file_id,
            "name": layer_name
        }
        response = await self._send_command(CommandType.DELETE_LAYER, payload)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)

    # 要素相关方法
    async def create_feature(self, file_id: str, feature_name: str,
                            feature_define: List[Dict[str, str]]) -> None:
        """添加要素

        Args:
            file_id: 文件ID、文件名称
            feature_name: 要素名称
            feature_define: 属性定义列表, 每个元素包含:
                - name: 属性名称
                - type: 属性类型 ("byte", "short", "int", "long", "float", "double", "char", "string", "color", "date", "binary")

        Raises:
            Exception: 创建失败时抛出异常
        """
        payload = {
            "file": file_id,
            "feature_name": feature_name,
            "feature_define": feature_define
        }
        response = await self._send_command(CommandType.CREATE_FEATURE, payload)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)


    async def get_features(self, file_id: str) -> List[FeatureInfo]:
        """获取要素

        Args:
            file_id: 文件ID

        Returns:
            要素信息列表

        Raises:
            Exception: 获取失败时抛出异常
        """
        response = await self._send_command(CommandType.GET_FEATURES, {"file": file_id})
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)
        return [FeatureInfo(**item, related=True) for item in response.payload]

    async def update_feature(self, file_id: str, feature_name: str,
                            feature_define: List[Dict[str, str]]) -> None:
        """更新要素

        Args:
            file_id: 文件名称或文件ID
            feature_name: 要素名称
            feature_define: 属性定义列表, 每个元素包含:
                - name: 属性名称
                - type: 属性类型 ("byte", "short", "int", "long", "float", "double", "char", "string", "color", "date", "binary")

        Raises:
            Exception: 更新失败时抛出异常
        """
        payload = {
            "file": file_id,
            "feature_name": feature_name,
            "feature_define": feature_define
        }
        response = await self._send_command(CommandType.UPDATE_FEATURE, payload)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)

    async def delete_feature(self, file: str, feature_name: str) -> None:
        """删除要素

        Args:
            file: 文件名称
            feature_name: 要素名称

        Raises:
            Exception: 删除失败时抛出异常
        """
        payload = {
            "file": file,
            "feature_name": feature_name
        }
        response = await self._send_command(CommandType.DELETE_FEATURE, payload)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)



    async def select_geometry(self, geometrys: List[Dict[str, str]],
                               mode: SelectMode | str = SelectMode.REPLACE) -> List[Dict[str, str]]:
        """选中/取消选中模型

        Args:
            geometrys: 要操作的模型列表，每个元素包含:
                - id: 模型ID (句柄ID)
                - file: 所属文件ID
            mode: 选择模式，可选 replace(替换当前选择) | add(追加) | remove(移除)

        Returns:
            执行后当前选中的模型列表

        Raises:
            Exception: 操作失败时抛出异常
        """
        payload = {
            "geometrys": geometrys,
            "mode": str(mode.value if isinstance(mode, SelectMode) else mode)
        }
        response = await self._send_command(CommandType.GEOMETRY_SELECT, payload)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)
        payload = response.payload or {}
        return payload.get("selected_geometrys", [])

    async def set_geometry_visibility(self, geometrys: List[Dict[str, str]],
                                       action: VisibilityAction | str = VisibilityAction.SHOW) -> None:
        """设置模型显示/隐藏状态

        Args:
            geometrys: 要操作的模型列表，每个元素包含:
                - id: 模型ID (句柄ID)
                - file: 所属文件ID
            action: 操作类型，可选 show(显示) | hide(隐藏) | exclusive_show(排他性显示)

        Raises:
            Exception: 操作失败时抛出异常
        """
        payload = {
            "geometrys": geometrys,
            "action": str(action)
        }
        response = await self._send_command(CommandType.GEOMETRY_VISIBILITY, payload)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)

    async def set_layer_visibility(self, file_id: str, layer_name: str,
                                    action: VisibilityAction | str = VisibilityAction.SHOW) -> None:
        """设置图层显示/隐藏状态

        Args:
            file_id: 文件ID或文件名称
            layer_name: 图层名称
            action: 操作类型，可选 show(显示) | hide(隐藏) | exclusive_show(排他性显示)

        Raises:
            Exception: 操作失败时抛出异常
        """
        payload = {
            "file": file_id,
            "layer": layer_name,
            "action": str(action)
        }
        response = await self._send_command(CommandType.LAYER_VISIBILITY, payload)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)

    async def set_file_visibility(self, file_id: str,
                                   action: VisibilityAction | str = VisibilityAction.SHOW) -> None:
        """设置文件显示/隐藏状态

        Args:
            file_id: 文件ID或文件名称
            action: 操作类型，可选 show(显示) | hide(隐藏) | exclusive_show(排他性显示)

        Raises:
            Exception: 操作失败时抛出异常
        """
        payload = {
            "file": file_id,
            "action": str(action)
        }
        response = await self._send_command(CommandType.FILE_VISIBILITY, payload)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)

    async def set_camera(self, center_point: List[float] | np.ndarray,
                          normal_vector: List[float] | np.ndarray) -> None:
        """设置相机视角

        Args:
            center_point: 相机中心点坐标 [x, y, z]
            normal_vector: 相机法向量 [x, y, z]

        Raises:
            Exception: 操作失败时抛出异常
        """
        payload = {
            "center_point": list(center_point),
            "normal_vector": list(normal_vector)
        }
        response = await self._send_command(CommandType.CAMERA_SET, payload)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)

    async def take_screenshot(self) -> str:
        """截图

        Returns:
            生成的图片完整路径

        Raises:
            Exception: 操作失败时抛出异常
        """
        import os
        from pathlib import Path

        response = await self._send_command(CommandType.CAMERA_SCREENSHOT, {})
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)
        image = response.payload.get("image", "")
        if not image:
            raise Exception("未生成图片")
        if not os.path.isabs(image):
            dimine_home = os.environ.get("DIMINE_HOME")
            if dimine_home:
                image = str(Path(dimine_home) / image)
        return image

    async def add_text(self, texts: List[TextInfo]) -> None:
        """添加文字

        Args:
            texts: 文字信息列表

        Raises:
            Exception: 添加失败时抛出异常
        """
        payload = []
        for t in texts:
            payload.append({
                "file": t.file,
                "layer": t.layer,
                "text": t.text,
                "position": t.position,
                "color": t.color,
                "size": t.size,
                "feature": t.feature or "0",
                "normal": t.normal,
                "rotate": t.rotate,
            })
        response = await self._send_command(CommandType.TEXT_ADD, payload)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)

    async def get_selected(self, page_size: int = 100, page_num: int = 1) -> SelectedResult:
        """获取当前选中的模型

        Args:
            page_size: 每页数量，默认 100
            page_num: 页码，从 1 开始，默认 1

        Returns:
            当前选择对象查询结果，包含选中对象列表及分页信息

        Raises:
            Exception: 获取失败时抛出异常
        """
        res = await self._send_command(
            CommandType.CURRENT_SELECT,
            {"page_size": page_size, "page_num": page_num}
        )
        if res.status == ResponseStatus.FAIL:
            raise Exception(res.error)
        payload = res.payload or {}
        return SelectedResult(
            ids=[SelectedItem(**item) for item in payload.get("ids", [])],
            page_size=payload.get("page_size", page_size),
            page_num=payload.get("page_num", page_num),
            total_count=payload.get("total_count", 0)
        )

add_text(texts) async

添加文字

Parameters:

Name	Type	Description	Default
texts	List[TextInfo]	文字信息列表	required

Raises:

Type	Description
Exception	添加失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def add_text(self, texts: List[TextInfo]) -> None:
    """添加文字

    Args:
        texts: 文字信息列表

    Raises:
        Exception: 添加失败时抛出异常
    """
    payload = []
    for t in texts:
        payload.append({
            "file": t.file,
            "layer": t.layer,
            "text": t.text,
            "position": t.position,
            "color": t.color,
            "size": t.size,
            "feature": t.feature or "0",
            "normal": t.normal,
            "rotate": t.rotate,
        })
    response = await self._send_command(CommandType.TEXT_ADD, payload)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)

close_file(file_id) async

关闭文件

Source code in dimine_python_sdk\conn\conn_client.py

async def close_file(self, file_id: str):
    """关闭文件"""
    res = await self._send_command(CommandType.FILE_CLOSE, {"file": file_id})
    if res.status == ResponseStatus.FAIL:
        raise Exception(res.error)

create_feature(file_id, feature_name, feature_define) async

添加要素

Parameters:

Name	Type	Description	Default
file_id	str	文件ID、文件名称	required
feature_name	str	要素名称	required
feature_define	List[Dict[str, str]]	属性定义列表, 每个元素包含: - name: 属性名称 - type: 属性类型 ("byte", "short", "int", "long", "float", "double", "char", "string", "color", "date", "binary")	required

Raises:

Type	Description
Exception	创建失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def create_feature(self, file_id: str, feature_name: str,
                        feature_define: List[Dict[str, str]]) -> None:
    """添加要素

    Args:
        file_id: 文件ID、文件名称
        feature_name: 要素名称
        feature_define: 属性定义列表, 每个元素包含:
            - name: 属性名称
            - type: 属性类型 ("byte", "short", "int", "long", "float", "double", "char", "string", "color", "date", "binary")

    Raises:
        Exception: 创建失败时抛出异常
    """
    payload = {
        "file": file_id,
        "feature_name": feature_name,
        "feature_define": feature_define
    }
    response = await self._send_command(CommandType.CREATE_FEATURE, payload)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)

create_file(file_path) async

在 Dimine 中新建文件

Parameters:

Name	Type	Description	Default
file_path	str	新建文件的完整路径	required

Raises:

Type	Description
Exception	创建失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def create_file(self, file_path: str) -> None:
    """在 Dimine 中新建文件

    Args:
        file_path: 新建文件的完整路径

    Raises:
        Exception: 创建失败时抛出异常
    """
    res = await self._send_command(CommandType.FILE_CREATE, {"file_path": file_path})
    if res.status == ResponseStatus.FAIL:
        raise Exception(res.error)

create_geometry(geometries, *, file_transfer='auto', file_transfer_threshold=10000) async

创建模型

Parameters:

Name	Type	Description	Default
geometries	List[GEOMETRY]	模型列表，每个模型包含以下字段: - file: 文件名称 - layer: 图层名称 - feature: 要素列名称 - entity_type: "point | line | polyline | shell" - geometry: 几何信息 (coordinates) - properties: 属性信息列表 - color: 颜色 [R, G, B]	required
file_transfer	bool | Literal['auto']	是否通过本地临时 DMF 文件传输大模型。 "auto"（默认）时根据 file_transfer_threshold 自动判断； True 时全部走文件路径；False 时全部走原 WebSocket 坐标路径。	'auto'
file_transfer_threshold	int	顶点/面片数阈值，默认 10000。	10000

Raises:

Type	Description
Exception	创建失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def create_geometry(
    self,
    geometries: List[GEOMETRY],
    *,
    file_transfer: bool | Literal["auto"] = "auto",
    file_transfer_threshold: int = 10000,
) -> None:
    """创建模型

    Args:
        geometries: 模型列表，每个模型包含以下字段:
            - file: 文件名称
            - layer: 图层名称
            - feature: 要素列名称
            - entity_type: "point | line | polyline | shell"
            - geometry: 几何信息 (coordinates)
            - properties: 属性信息列表
            - color: 颜色 [R, G, B]
        file_transfer: 是否通过本地临时 DMF 文件传输大模型。
            "auto"（默认）时根据 ``file_transfer_threshold`` 自动判断；
            ``True`` 时全部走文件路径；``False`` 时全部走原 WebSocket 坐标路径。
        file_transfer_threshold: 顶点/面片数阈值，默认 10000。

    Raises:
        Exception: 创建失败时抛出异常
    """
    if not geometries:
        return

    small_geometries = []
    large_geometries = []
    for geom in geometries:
        if self._should_use_file_transfer(geom, file_transfer, file_transfer_threshold):
            large_geometries.append(geom)
        else:
            small_geometries.append(geom)

    # 小模型保持原有批量 WebSocket 发送逻辑
    if small_geometries:
        payload = [self._format_geometry_payload(g) for g in small_geometries]
        response = await self._send_command(CommandType.CREATE_GEOMETRY, payload)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)
        for i, item in enumerate(response.payload or []):
            small_geometries[i].id = item.get("id")
            small_geometries[i].related = True

    # 大模型逐个生成临时 DMF 文件并通过 file_path 发送
    if large_geometries:
        tmp_dir = Path(tempfile.mkdtemp(prefix="dimine_create_geometry_"))
        try:
            await self._create_geometry_via_file(large_geometries, tmp_dir)
        finally:
            shutil.rmtree(tmp_dir, ignore_errors=True)

create_layer(file_id, name) async

创建图层

Parameters:

Name	Type	Description	Default
file_id	str	文件ID或文件名称	required
name	str	图层名称	required

Returns:

Type	Description
LayerInfo	创建的图层信息

Raises:

Type	Description
Exception	创建失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def create_layer(self, file_id: str, name: str) -> LayerInfo:
    """创建图层

    Args:
        file_id: 文件ID或文件名称
        name: 图层名称

    Returns:
        创建的图层信息

    Raises:
        Exception: 创建失败时抛出异常
    """
    payload = {
        "file": file_id,
        "name": name
    }
    response = await self._send_command(CommandType.CREATE_LAYER, payload)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)
    return LayerInfo(id=name, name=name, related=True)

delete_feature(file, feature_name) async

删除要素

Parameters:

Name	Type	Description	Default
file	str	文件名称	required
feature_name	str	要素名称	required

Raises:

Type	Description
Exception	删除失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def delete_feature(self, file: str, feature_name: str) -> None:
    """删除要素

    Args:
        file: 文件名称
        feature_name: 要素名称

    Raises:
        Exception: 删除失败时抛出异常
    """
    payload = {
        "file": file,
        "feature_name": feature_name
    }
    response = await self._send_command(CommandType.DELETE_FEATURE, payload)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)

delete_geometry(delete_items) async

删除模型

Parameters:

Name	Type	Description	Default
delete_items	List[Dict[str, str]]	删除项列表，每个项包含: - id: 模型ID (句柄ID) - file: 文件ID (从哪个文件删除模型)	required

Raises:

Type	Description
Exception	删除失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def delete_geometry(self, delete_items: List[Dict[str, str]]) -> None:
    """删除模型

    Args:
        delete_items: 删除项列表，每个项包含:
            - id: 模型ID (句柄ID)
            - file: 文件ID (从哪个文件删除模型)

    Raises:
        Exception: 删除失败时抛出异常
    """
    response = await self._send_command(CommandType.DELETE_GEOMETRY, delete_items)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)

delete_layer(file_id, layer_name) async

删除图层

Parameters:

Name	Type	Description	Default
file_id	str	文件ID或文件名称	required
layer_name	str	图层名称	required

Raises:

Type	Description
Exception	删除失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def delete_layer(self, file_id: str, layer_name: str) -> None:
    """删除图层

    Args:
        file_id: 文件ID或文件名称
        layer_name: 图层名称

    Raises:
        Exception: 删除失败时抛出异常
    """
    payload = {
        "file": file_id,
        "name": layer_name
    }
    response = await self._send_command(CommandType.DELETE_LAYER, payload)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)

export_file(file_id, file_path, mkdir=True) async

导出文件

Parameters:

Name	Type	Description	Default
file_id	str	文件ID或文件名称	required
file_path	str | Path	导出目标路径	required
mkdir	bool	是否自动创建目录	True

Raises: Exception: 导出失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def export_file(self, file_id: str, file_path: str| Path, mkdir: bool = True):
    """导出文件

    Args:
        file_id: 文件ID或文件名称
        file_path: 导出目标路径
        mkdir: 是否自动创建目录
    Raises:
        Exception: 导出失败时抛出异常
    """
    if isinstance(file_path, str):  # 如果是 str 对象，则转换为 Path
        file_path = Path(file_path)
        if not file_path.parent.exists():  # 如果目录不存在，报错
            if mkdir:
                file_path.parent.mkdir(parents=True, exist_ok=True)
            else:
                raise Exception(f"path {file_path.parent} not exists")
    res = await self._send_command(CommandType.FILE_EXPORT, {"file": file_id, "file_path": str(file_path)})
    if res.status == ResponseStatus.FAIL:
        raise Exception(res.error)

get_features(file_id) async

获取要素

Parameters:

Name	Type	Description	Default
file_id	str	文件ID	required

Returns:

Type	Description
List[FeatureInfo]	要素信息列表

Raises:

Type	Description
Exception	获取失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def get_features(self, file_id: str) -> List[FeatureInfo]:
    """获取要素

    Args:
        file_id: 文件ID

    Returns:
        要素信息列表

    Raises:
        Exception: 获取失败时抛出异常
    """
    response = await self._send_command(CommandType.GET_FEATURES, {"file": file_id})
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)
    return [FeatureInfo(**item, related=True) for item in response.payload]

get_files() async

获取文件对象树

Source code in dimine_python_sdk\conn\conn_client.py

async def get_files(self) -> List[FileInfo]:
    """获取文件对象树"""
    res = await self._send_command(CommandType.GET_FILE_TREE)
    if res.status == ResponseStatus.FAIL:
        raise Exception(res.error)
    return [FileInfo(**item, related=True) for item in res.payload]

get_geometry(file_id, layer_id, entity_types=[], features=None, with_geometry=False) async

获取模型

Parameters:

Name	Type	Description	Default
file_id	str	文件ID	required
layer_id	str	图层名称	required
entity_types	List[str]	获取哪些类型的模型	[]
features	Optional[List[str]]	获取哪些要素相关的模型	None
with_geometry	bool	是否返回模型坐标信息，默认 False	False

Source code in dimine_python_sdk\conn\conn_client.py

async def get_geometry(self, file_id: str, layer_id: str, entity_types: List[str]=[], features: Optional[List[str]] = None, with_geometry: bool = False) -> List[GEOMETRY]:
    """获取模型

    Args:
        file_id: 文件ID
        layer_id: 图层名称
        entity_types: 获取哪些类型的模型
        features: 获取哪些要素相关的模型
        with_geometry: 是否返回模型坐标信息，默认 False
    """
    # 先获取要素信息，
    all_features = await self.get_features(file_id)

    def _get_property_type(feature_name:str, name: str)-> str:
        """获取属性类型"""
        for feature in all_features:
            if feature.name == feature_name:
                for property in feature.properties:
                    if property.name == name:
                        return property.type

    payload = {
        "file": file_id,
        "layer": layer_id,
        "entity_type": entity_types,
        "with_geometry": with_geometry,
    }
    if features:
        payload["feature"] = features

    response = await self._send_command(CommandType.GET_GEOMETRY, payload)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)

    def _format_value(type:str, value: Any):
        if value == "" and type not in ("string", "char"):
            return None
        elif type == "double":
            return float(value)
        elif type in ("string", "char"):
            return str(value)
        elif type in ["int", "long", "short", "byte"]:
            return int(value)
        elif type == "float":
            return float(value)
        elif type == "color":
            return int(value)
        elif type == "date":
            return str(value)
        elif type == "binary":
            return value
        else:
            raise Exception(f"unknown property type: {type}")

    def _handle_property(feature_name:str, properties: List[Dict[str, Any]]):
        for property in properties:
            property["type"] = _get_property_type(feature_name, property.get("name"))
            property["value"] = _format_value(property.get("type"), property.get("value"))

    geometries = []
    for item in response.payload:
        _handle_property(item.get("feature"), item.get("properties", []))
        if isinstance(item.get("geometry", []), list):
            tdata = np.asarray(item.get("geometry", [])).astype(np.float64)
            if len(tdata.shape) == 1:
                item["geometry"] = tdata
                geometries.append(Point(**item, related=True))
            elif len(tdata.shape) == 2:
                geometries.append(Line(**item, related=True))
        elif isinstance(item.get("geometry", []), dict):
            item["geometry"]["faces"] = item["geometry"].pop("indexs") if "indexs" in item["geometry"] else []
            geometries.append(Shell(**item, related=True))
    for geom in geometries:
        geom.file = file_id
        geom.layer = layer_id
    return geometries

get_layers(file_id) async

获取图层

Source code in dimine_python_sdk\conn\conn_client.py

async def get_layers(self, file_id: str) -> List[LayerInfo]:
    """获取图层"""
    response = await self._send_command(CommandType.GET_LAYERS, {"file": file_id})
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)
    return [LayerInfo(**item, related=True) for item in response.payload]

get_selected(page_size=100, page_num=1) async

获取当前选中的模型

Parameters:

Name	Type	Description	Default
page_size	int	每页数量，默认 100	100
page_num	int	页码，从 1 开始，默认 1	1

Returns:

Type	Description
SelectedResult	当前选择对象查询结果，包含选中对象列表及分页信息

Raises:

Type	Description
Exception	获取失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def get_selected(self, page_size: int = 100, page_num: int = 1) -> SelectedResult:
    """获取当前选中的模型

    Args:
        page_size: 每页数量，默认 100
        page_num: 页码，从 1 开始，默认 1

    Returns:
        当前选择对象查询结果，包含选中对象列表及分页信息

    Raises:
        Exception: 获取失败时抛出异常
    """
    res = await self._send_command(
        CommandType.CURRENT_SELECT,
        {"page_size": page_size, "page_num": page_num}
    )
    if res.status == ResponseStatus.FAIL:
        raise Exception(res.error)
    payload = res.payload or {}
    return SelectedResult(
        ids=[SelectedItem(**item) for item in payload.get("ids", [])],
        page_size=payload.get("page_size", page_size),
        page_num=payload.get("page_num", page_num),
        total_count=payload.get("total_count", 0)
    )

open_file(file_path) async

打开文件

Source code in dimine_python_sdk\conn\conn_client.py

async def open_file(self, file_path: str| Path) -> FileInfo:
    """打开文件"""
    if isinstance(file_path, str):  # 如果是 str 对象，则转换为 Path
        file_path = Path(file_path)
        if not file_path.exists():  # 如果文件不存在，报错
            raise Exception(f"file {file_path} not exists")
    res = await self._send_command(CommandType.FILE_OPEN, {"file_path": str(file_path)})
    if res.status == ResponseStatus.FAIL:
        raise Exception(res.error)
    if res.payload:
        return FileInfo(id=res.payload[0].get("file_id"), name=res.payload[0].get("file_id"), related=True)
    else:
        raise Exception("file already opened!")

select_geometry(geometrys, mode=SelectMode.REPLACE) async

选中/取消选中模型

Parameters:

Name	Type	Description	Default
geometrys	List[Dict[str, str]]	要操作的模型列表，每个元素包含: - id: 模型ID (句柄ID) - file: 所属文件ID	required
mode	SelectMode | str	选择模式，可选 replace(替换当前选择) | add(追加) | remove(移除)	REPLACE

Returns:

Type	Description
List[Dict[str, str]]	执行后当前选中的模型列表

Raises:

Type	Description
Exception	操作失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def select_geometry(self, geometrys: List[Dict[str, str]],
                           mode: SelectMode | str = SelectMode.REPLACE) -> List[Dict[str, str]]:
    """选中/取消选中模型

    Args:
        geometrys: 要操作的模型列表，每个元素包含:
            - id: 模型ID (句柄ID)
            - file: 所属文件ID
        mode: 选择模式，可选 replace(替换当前选择) | add(追加) | remove(移除)

    Returns:
        执行后当前选中的模型列表

    Raises:
        Exception: 操作失败时抛出异常
    """
    payload = {
        "geometrys": geometrys,
        "mode": str(mode.value if isinstance(mode, SelectMode) else mode)
    }
    response = await self._send_command(CommandType.GEOMETRY_SELECT, payload)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)
    payload = response.payload or {}
    return payload.get("selected_geometrys", [])

set_camera(center_point, normal_vector) async

设置相机视角

Parameters:

Name	Type	Description	Default
center_point	List[float] | ndarray	相机中心点坐标 [x, y, z]	required
normal_vector	List[float] | ndarray	相机法向量 [x, y, z]	required

Raises:

Type	Description
Exception	操作失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def set_camera(self, center_point: List[float] | np.ndarray,
                      normal_vector: List[float] | np.ndarray) -> None:
    """设置相机视角

    Args:
        center_point: 相机中心点坐标 [x, y, z]
        normal_vector: 相机法向量 [x, y, z]

    Raises:
        Exception: 操作失败时抛出异常
    """
    payload = {
        "center_point": list(center_point),
        "normal_vector": list(normal_vector)
    }
    response = await self._send_command(CommandType.CAMERA_SET, payload)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)

set_file_visibility(file_id, action=VisibilityAction.SHOW) async

设置文件显示/隐藏状态

Parameters:

Name	Type	Description	Default
file_id	str	文件ID或文件名称	required
action	VisibilityAction | str	操作类型，可选 show(显示) | hide(隐藏) | exclusive_show(排他性显示)	SHOW

Raises:

Type	Description
Exception	操作失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def set_file_visibility(self, file_id: str,
                               action: VisibilityAction | str = VisibilityAction.SHOW) -> None:
    """设置文件显示/隐藏状态

    Args:
        file_id: 文件ID或文件名称
        action: 操作类型，可选 show(显示) | hide(隐藏) | exclusive_show(排他性显示)

    Raises:
        Exception: 操作失败时抛出异常
    """
    payload = {
        "file": file_id,
        "action": str(action)
    }
    response = await self._send_command(CommandType.FILE_VISIBILITY, payload)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)

set_geometry_visibility(geometrys, action=VisibilityAction.SHOW) async

设置模型显示/隐藏状态

Parameters:

Name	Type	Description	Default
geometrys	List[Dict[str, str]]	要操作的模型列表，每个元素包含: - id: 模型ID (句柄ID) - file: 所属文件ID	required
action	VisibilityAction | str	操作类型，可选 show(显示) | hide(隐藏) | exclusive_show(排他性显示)	SHOW

Raises:

Type	Description
Exception	操作失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def set_geometry_visibility(self, geometrys: List[Dict[str, str]],
                                   action: VisibilityAction | str = VisibilityAction.SHOW) -> None:
    """设置模型显示/隐藏状态

    Args:
        geometrys: 要操作的模型列表，每个元素包含:
            - id: 模型ID (句柄ID)
            - file: 所属文件ID
        action: 操作类型，可选 show(显示) | hide(隐藏) | exclusive_show(排他性显示)

    Raises:
        Exception: 操作失败时抛出异常
    """
    payload = {
        "geometrys": geometrys,
        "action": str(action)
    }
    response = await self._send_command(CommandType.GEOMETRY_VISIBILITY, payload)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)

set_layer_visibility(file_id, layer_name, action=VisibilityAction.SHOW) async

设置图层显示/隐藏状态

Parameters:

Name	Type	Description	Default
file_id	str	文件ID或文件名称	required
layer_name	str	图层名称	required
action	VisibilityAction | str	操作类型，可选 show(显示) | hide(隐藏) | exclusive_show(排他性显示)	SHOW

Raises:

Type	Description
Exception	操作失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def set_layer_visibility(self, file_id: str, layer_name: str,
                                action: VisibilityAction | str = VisibilityAction.SHOW) -> None:
    """设置图层显示/隐藏状态

    Args:
        file_id: 文件ID或文件名称
        layer_name: 图层名称
        action: 操作类型，可选 show(显示) | hide(隐藏) | exclusive_show(排他性显示)

    Raises:
        Exception: 操作失败时抛出异常
    """
    payload = {
        "file": file_id,
        "layer": layer_name,
        "action": str(action)
    }
    response = await self._send_command(CommandType.LAYER_VISIBILITY, payload)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)

take_screenshot() async

截图

Returns:

Type	Description
str	生成的图片完整路径

Raises:

Type	Description
Exception	操作失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def take_screenshot(self) -> str:
    """截图

    Returns:
        生成的图片完整路径

    Raises:
        Exception: 操作失败时抛出异常
    """
    import os
    from pathlib import Path

    response = await self._send_command(CommandType.CAMERA_SCREENSHOT, {})
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)
    image = response.payload.get("image", "")
    if not image:
        raise Exception("未生成图片")
    if not os.path.isabs(image):
        dimine_home = os.environ.get("DIMINE_HOME")
        if dimine_home:
            image = str(Path(dimine_home) / image)
    return image

update_feature(file_id, feature_name, feature_define) async

更新要素

Parameters:

Name	Type	Description	Default
file_id	str	文件名称或文件ID	required
feature_name	str	要素名称	required
feature_define	List[Dict[str, str]]	属性定义列表, 每个元素包含: - name: 属性名称 - type: 属性类型 ("byte", "short", "int", "long", "float", "double", "char", "string", "color", "date", "binary")	required

Raises:

Type	Description
Exception	更新失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def update_feature(self, file_id: str, feature_name: str,
                        feature_define: List[Dict[str, str]]) -> None:
    """更新要素

    Args:
        file_id: 文件名称或文件ID
        feature_name: 要素名称
        feature_define: 属性定义列表, 每个元素包含:
            - name: 属性名称
            - type: 属性类型 ("byte", "short", "int", "long", "float", "double", "char", "string", "color", "date", "binary")

    Raises:
        Exception: 更新失败时抛出异常
    """
    payload = {
        "file": file_id,
        "feature_name": feature_name,
        "feature_define": feature_define
    }
    response = await self._send_command(CommandType.UPDATE_FEATURE, payload)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)

update_geometry(geometries) async

更新模型

Parameters:

Name	Type	Description	Default
geometries	List[GEOMETRY]	模型列表，每个模型包含以下字段: - id: 模型ID (句柄ID) - layer: 图层名称 - entity_type: "point | line | polyline | shell" - geometry: 模型几何信息 (可选) - properties: 模型属性信息 (可选) - color: 颜色 [R, G, B] (可选) - feature: 模型关联要素列表 (可选)	required

Raises:

Type	Description
Exception	更新失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def update_geometry(self, geometries: List[GEOMETRY]) -> None:
    """更新模型

    Args:
        geometries: 模型列表，每个模型包含以下字段:
            - id: 模型ID (句柄ID)
            - layer: 图层名称
            - entity_type: "point | line | polyline | shell"
            - geometry: 模型几何信息 (可选)
            - properties: 模型属性信息 (可选)
            - color: 颜色 [R, G, B] (可选)
            - feature: 模型关联要素列表 (可选)

    Raises:
        Exception: 更新失败时抛出异常
    """
    # 格式化模型数据
    formatted_geometries = []
    for geom in geometries:
        formatted_geom = {
            "id": geom.id,
            "file": geom.file,
            "layer": geom.layer,
            "feature": geom.feature,
            "entity_type": geom.type,
            "geometry": geom.geometry.tolist() if geom.type!="shell" else {"indexs": geom.geometry.faces.tolist(), "points": geom.geometry.points.tolist()},
            "color": geom.color,
        }

        # 处理属性信息
        formatted_geom["properties"] = [
                {"name": prop.name, "value": prop.value}
                for prop in geom.properties
            ]
        formatted_geometries.append(formatted_geom)

    response = await self._send_command(CommandType.UPDATE_GEOMETRY, formatted_geometries)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)

update_or_create_geometry(geometries, *, file_transfer='auto', file_transfer_threshold=10000) async

创建或更新多个模型, 如果指定了file_id和layer_id, 则将模型添加到指定的文件及图层中

Source code in dimine_python_sdk\conn\conn_client.py

async def update_or_create_geometry(
    self,
    geometries: List[GEOMETRY] | GEOMETRY,
    *,
    file_transfer: bool | Literal["auto"] = "auto",
    file_transfer_threshold: int = 10000,
) -> None:
    """创建或更新多个模型, 如果指定了file_id和layer_id, 则将模型添加到指定的文件及图层中"""
    if isinstance(geometries, (Point, Line, Shell)):
        geometries = [geometries]
    update_geometries = [g for g in geometries if g.related]
    create_geometries = [g for g in geometries if not g.related]
    if update_geometries:
        await self.update_geometry(update_geometries)
    if create_geometries:
        await self.create_geometry(
            create_geometries,
            file_transfer=file_transfer,
            file_transfer_threshold=file_transfer_threshold,
        )