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:
        """判断单个几何体是否应走本地文件传输路径。"""
        # 几何类型与 numpy 均为惰性导入：远程命令路径（file-tree/layers 等）不碰它们，
        # 从而避免在导入期拉起 numpy。
        import numpy as np

        from dimine_python_sdk.models.geometry import Line, Point, Shell

        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 _resolve_entity_type(self, geom: GEOMETRY) -> str:
        """按几何数据判定写入服务端的 ``entity_type``。

        Dimine 侧对 ``line`` / ``polyline`` 有顶点数校验：``line`` 恰好 2 个顶点，
        多段折线必须是 ``polyline``，否则创建/更新都会返回
        ``The value of the key geometry has error!``。因此只要携带了几何数据，
        就必须按顶点数重新判定，不能直接用 ``geom.type``。

        例外：未携带几何数据（如只更新颜色/属性）时无从推断，此时沿用模型声明的
        ``geom.type``——服务端要求无几何的更新与库中已有实体类型完全一致，所以
        ``Line.type`` 必须能表达 ``"polyline"``（见 ``models.geometry.Line``）。
        """
        import numpy as np

        from dimine_python_sdk.models.geometry import Line, Point, Shell

        if isinstance(geom, Point):
            return "point"
        if isinstance(geom, Line):
            count = np.asarray(geom.geometry).shape[0] if geom.geometry is not None else 0
            if count == 0:
                # 未携带几何数据（如只更新颜色/属性）。此时无从按顶点数推断，
                # 沿用模型声明的类型，不能当成退化几何报错。
                return geom.type or "line"
            if count == 2:
                return "line"
            if count > 2:
                return "polyline"
            raise ValueError(
                f"线至少需要 2 个顶点，实际 {count} 个；无法创建退化的线几何体"
            )
        if isinstance(geom, Shell):
            return "shell"
        return geom.type or "line"

    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": self._resolve_entity_type(geom),
            "color": geom.color if geom.color is not None else [0, 0, 0],
            "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。

        与 WebSocket 路径共用 :meth:`_resolve_entity_type`，保证两条分支
        对同一几何体发出相同的 ``entity_type``。
        """
        return self._resolve_entity_type(geom)

    def _insert_geometry_into_dmf(self, dmf, geom: GEOMETRY) -> None:
        """将几何体写入 DMF 内存模型（图层名为 ``"0"``）。"""
        import numpy as np

        from dimine_python_sdk.models.geometry import Line, Point, Shell

        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()

        # 第一步：生成所有 DMF 文件并构建 payload 数组
        payloads = []
        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)
            payload["geometry"] = []
            payloads.append(payload)

        # 第二步：批量发送
        response = await self._send_command(CommandType.CREATE_GEOMETRY, payloads)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)

        # 第三步：回填 id 和 related
        for i, item in enumerate(response.payload or []):
            if isinstance(item, dict):
                geometries[i].id = item.get("id")
                geometries[i].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。

        坐标**不再**走 websocket：发出去的 ``with_geometry`` 恒为 ``False``，坐标数据
        一律由服务端落盘成一个临时 dmf 文件（见 :meth:`get_geometry_file`），需要坐标
        时按实体 ``id`` 从该文件里取。

        因此本方法只返回**基础信息**（id / feature / type / color / properties），
        ``geom.geometry`` 是空的。要坐标请用 :meth:`get_geometry_file`。

        Raises:
            ValueError: 显式传 ``with_geometry=True`` 时。这样调用比返回一堆空坐标更
                容易被发现——旧签名还在，静默返回空几何会让调用方以为"这个图层没坐标"。
        """
        if with_geometry:
            raise ValueError(
                "get_geometry 不再内联返回坐标：服务端把 with_geometry 当"
                "\"是否内联坐标\"用，传 True 会让坐标走 websocket（正是新协议要避免的）。"
                "请改用 get_geometry_file() 拿落盘 dmf 路径，按实体 id 读取坐标。"
            )
        result = await self._fetch_geometry(file_id, layer_id, entity_types, features)
        return result.entities

    async def get_geometry_file(self, file_id: str, layer_id: str, entity_types: List[str]=[], features: Optional[List[str]] = None) -> GeometryGetResult:
        """获取模型元信息与坐标数据所在的 dmf 文件，**不读取**该文件。

        这是获取几何数据的统一入口：坐标不走 websocket，一律由服务端落盘，本方法把
        ``geometry_file`` 路径交给调用方，由调用方自己按实体 ``id`` 解析该文件。

        返回的 ``entities`` 只带基础信息（id / feature / type / color / properties），
        ``geometry`` 为空；坐标在 ``geometry_file`` 指向的文件里，**按 id 匹配**取用
        （文件里是整个图层，可能比本次命中的实体更多，不能按顺序对齐）。

        Returns:
            ``GeometryGetResult``：``geometry_file`` 为 dmf 路径（服务端未产出文件时为
            空字符串），``entities`` 为模型元信息列表。

        Note:
            该文件由 Dimine 端管理，可能被后续请求覆盖或清理，需要时应在本次会话内
            读取完毕，不要长期持有该路径。
        """
        return await self._fetch_geometry(file_id, layer_id, entity_types, features)

    async def _fetch_geometry(
        self,
        file_id: str,
        layer_id: str,
        entity_types: List[str],
        features: Optional[List[str]],
    ) -> GeometryGetResult:
        """发送 ``geometry.get`` 并把响应解析成 :class:`GeometryGetResult`（不读 dmf）。"""
        # 先获取要素信息，
        all_features = await self.get_features(file_id)

        payload = {
            "file": file_id,
            "layer": layer_id,
            "entity_type": entity_types,
            # 固定 False：服务端把该键当"是否把坐标内联进响应"用，而不是文档描述的
            # "是否返回坐标"。实测（conn_msgs.json 49 组配对）with_geometry=True 的
            # 20 次响应坐标**全部内联**（最大一次 756 点 / 50.9KB 照发不误），
            # with_geometry=False 的 29 次**全部**带非空 geometry_file。
            # 坐标要永远不走 websocket（这正是新协议的初衷），就只能发 False，
            # 坐标一律从落盘文件读（见 get_geometry / get_geometry_file）。
            **_GET_GEOMETRY_WIRE_FLAGS,
        }
        if features:
            payload["feature"] = features

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

        result = self._parse_geometry_payload(response.payload, all_features)
        for geom in result.entities:
            geom.file = file_id
            geom.layer = layer_id
        return result

    def _parse_geometry_payload(self, payload: Any, all_features: List[FeatureInfo]) -> GeometryGetResult:
        """把 ``geometry.get`` 的响应 payload 解析成 :class:`GeometryGetResult`。

        要兼容两种 payload：

        - **新格式**（对象）：``{"geometry_file": "...", "entities": [...]}``
        - **旧格式**（裸数组）：只有实体列表，没有 ``geometry_file``

        实体自身也有两种形态：带坐标（内联）与不带坐标（坐标已落盘，``geometry``
        为空）。因此类型**不能靠坐标形状推断**——落盘时所有实体的坐标都是空的，靠形状
        只会把多段线也识别成点。这里优先按协议给出的 ``entity_type`` 分派。
        """
        import numpy as np

        from dimine_python_sdk.models.geometry import GeometryGetResult, Line, Point, Shell

        geometry_file = ""
        if isinstance(payload, dict):
            geometry_file = payload.get("geometry_file") or ""
            raw_entities = payload.get("entities") or []
        else:
            # 旧格式：payload 本身就是实体数组
            raw_entities = payload or []

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

        def _format_value(type:str, value: Any):
            # 空值一律保持为空，别让数值类型把 None 拿去 int()/float() 崩掉。
            # 也让本函数可重复调用：它会把 "" 改写为 None 并就地写回 payload，
            # 若同一个 payload 被解析第二次，就会读到自己上次写下的 None。
            if value is None:
                return None
            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: List[GEOMETRY] = []
        for item in raw_entities:
            _handle_property(item.get("feature"), item.get("properties", []))
            # 类型按协议给出的 entity_type 分派，而不是按坐标形状——落盘返回时所有
            # 实体的坐标都是空的，靠形状会把多段线也识别成点。
            entity_type = item.get("entity_type")
            geometry = item.get("geometry", [])
            if entity_type == "point":
                # 点保持 float64（历史行为：点的坐标由这里显式转换，线/壳则交给各
                # 自 validator 转 float32）。别顺手统一成 float32，输出的小数位会变。
                if isinstance(geometry, list):
                    item["geometry"] = np.asarray(geometry).astype(np.float64)
                geometries.append(Point(**item, related=True))
            elif entity_type in ("line", "polyline"):
                # 坐标原样交给 Line 的 validator；落盘时是空 list，落回默认空数组。
                # entity_type="polyline" 由 Line.type 的 AliasChoices 收下并保留，
                # 不会退化回 "line"。
                geometries.append(Line(**item, related=True))
            elif entity_type == "shell" or isinstance(geometry, dict):
                if isinstance(geometry, dict):
                    item["geometry"] = {
                        "points": geometry.get("points", []),
                        "faces": geometry.get("faces") or geometry.get("indexs") or [],
                    }
                else:
                    # 没有坐标（落盘返回）也不是 dict 时不能留个 list 给 TINGeometry
                    item.pop("geometry", None)
                geometries.append(Shell(**item, related=True))
        return GeometryGetResult(geometry_file=geometry_file, entities=geometries)

    def _load_geometry_from_dmf(self, file_path: str) -> Dict[str, Any]:
        """读取落盘返回的 dmf，返回 ``{实体id: 实体对象}`` 索引。

        落盘文件里放的是**整个图层**的实体，可能比本次响应命中的实体更多（例如按
        ``entity_type=["polyline"]`` 过滤时，文件里仍会有 shell）。所以回填**必须按
        id 匹配**，不能按顺序对齐——按顺序会把别的实体的坐标塞过来。

        实体 id 落在 dmf 实体的 ``id`` 上（与服务端响应的 ``entities[i].id`` 同值）。
        """
        DmfFile = self._get_dmf_file_class()
        try:
            dmf = DmfFile(file_path)
        except Exception as exc:
            raise RuntimeError(
                f"无法读取几何坐标落盘文件 {file_path}：{exc}"
            ) from exc

        index: Dict[str, Any] = {}
        for layer in dmf.layers:
            for entity in layer.entities:
                entity_id = getattr(entity, "id", None)
                if entity_id:
                    index[str(entity_id)] = entity
        return index

    def _fill_geometry_from_dmf(self, result: GeometryGetResult) -> None:
        """把落盘 dmf 里的坐标按 id 回填到 ``result.entities`` 上。

        已经带坐标的实体（内联返回）原样保留，因此内联与落盘混在同一个响应里也不会
        出错；dmf 里查不到对应 id 的实体保持无坐标，不抛异常（该实体可能已被清理）。

        没有可回填的实体时**不打开文件**：图层为空时服务端给的不是临时文件而是原文件
        名甚至裸文件名（如 ``新文件_0.dmf``），这时开文件只会白报错，而本来也无坐标可填。
        """
        if not result.geometry_file:
            return
        if not any(g.id is None or not _has_coordinates(g) for g in result.entities):
            return  # 全部实体都已带坐标（或不需要回填），没必要读文件

        index = self._load_geometry_from_dmf(result.geometry_file)
        for geom in result.entities:
            if geom.id is None or _has_coordinates(geom):
                continue
            source = index.get(str(geom.id))
            if source is not None:
                geom.geometry = source.geometry

    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: Dict[str, Any] = {"id": geom.id}
            # 仅发送显式设置的非 None/空字段
            if geom.file is not None:
                formatted_geom["file"] = geom.file
            if geom.layer is not None:
                formatted_geom["layer"] = geom.layer
            if geom.feature is not None:
                formatted_geom["feature"] = geom.feature
            # entity_type 必须按几何数据判定：服务端会校验实体类型与几何是否匹配
            # （对多段折线发 "line" 会报 entity type is not match）。
            # 未携带几何的单纯属性更新则沿用模型声明的 type（Line 可表达 polyline）。
            if geom.type is not None:
                formatted_geom["entity_type"] = self._resolve_entity_type(geom)
            if geom.color is not None:
                formatted_geom["color"] = geom.color
            # geometry：仅当非空时发送
            if geom.type == "shell" and geom.geometry is not None:
                geo = geom.geometry
                if geo.points is not None and len(geo.points) > 0:
                    formatted_geom.setdefault("geometry", {})["points"] = geo.points.tolist()
                if geo.faces is not None and len(geo.faces) > 0:
                    formatted_geom.setdefault("geometry", {})["indexs"] = geo.faces.tolist()
            # polyline 也是线：get_geometry 会把服务端下发的 "polyline" 原样带进
            # Line.type，只判 "line" 会让这类实体的坐标在更新时被静默丢掉。
            elif geom.type in ("point", "line", "polyline") and geom.geometry is not None and len(geom.geometry) > 0:
                formatted_geom["geometry"] = geom.geometry.tolist()

            # 处理属性信息
            if geom.properties:
                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, 则将模型添加到指定的文件及图层中"""
        from dimine_python_sdk.models.geometry import Line, Point, Shell

        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_layers_visibility(self, layers: List[Dict[str, str]],
                                     action: VisibilityAction | str = VisibilityAction.SHOW) -> None:
        """批量设置图层显示/隐藏状态

        Args:
            layers: 要操作的图层列表，每个元素包含:
                - file: 文件ID或文件名称
                - layer: 图层名称
            action: 操作类型，可选 show(显示) | hide(隐藏) | exclusive_show(排他性显示)

        Raises:
            Exception: 操作失败时抛出异常
        """
        if not layers:
            return
        payload = {
            "layers": layers,
            "action": str(action)
        }
        response = await self._send_command(CommandType.LAYER_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: 操作失败时抛出异常
        """
        await self.set_layers_visibility(
            [{"file": file_id, "layer": layer_name}], action=action
        )

    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)
        )

    # 场景相关方法
    async def search_scene(
        self,
        keyword: str,
        *,
        scope: Optional[Sequence[SearchScope | str]] = None,
        files: Optional[Sequence[str]] = None,
        case_sensitive: bool = False,
    ) -> SceneSearchResult:
        """场景全文检索

        按正则表达式检索当前场景中已打开的文件、图层、要素、实体（含文字标注）。
        结果按 文件 → 图层 → 要素 → 实体 的顺序返回。

        Args:
            keyword: 检索正则表达式，对名称、要素名、实体属性字段名/值、文字内容等进行匹配，
                如 ``^(BH|ZK)-\\d+$`` 匹配孔号、``.*cu.*`` 匹配含 cu 的内容
            scope: 限定检索范围，可选 file | layer | feature | entity，
                默认检索全部类型。文字标注属于实体的一部分，检索 entity 时一并命中
            files: 限定检索的文件（文件id或文件名），默认检索当前场景所有已打开的文件
            case_sensitive: 是否区分大小写，默认 False（不区分）

        Returns:
            检索结果，命中项按 文件 → 图层 → 要素 → 实体 顺序排列

        Raises:
            Exception: 检索失败时抛出异常
        """
        payload: Dict[str, Any] = {
            "keyword": keyword,
            "case_sensitive": case_sensitive,
        }
        if scope:
            payload["scope"] = [
                str(s.value if isinstance(s, SearchScope) else s) for s in scope
            ]
        if files:
            payload["files"] = list(files)

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

    @staticmethod
    def _parse_scene_search_result(payload: Any) -> SceneSearchResult:
        """解析 ``scene.search`` 响应"""
        payload = payload or {}
        if not isinstance(payload, dict):
            raise Exception(f"scene.search 响应格式错误: {payload!r}")

        # 数采端失败时信封层仍返回 status=success，错误放在载荷里
        # （如 {"error": "Unknown exception!", "success": false}），需显式识别，
        # 否则会被当成"无分组的空结果"而静默吞掉。
        if payload.get("success") is False or (
            payload.get("error") and not any(
                payload.get(key) for _, keys in _SCENE_SEARCH_GROUPS for key in keys
            )
        ):
            raise Exception(payload.get("error") or "scene.search 执行失败")

        # 数采端文档给出两种结果结构：文档正文的按类型分组形式
        # （files/layers/features/entities/texts）与说明中的扁平 results 形式，两者都兼容。
        if "results" in payload:
            raw_results = payload.get("results") or []
            hits = [SceneSearchHit(**item) for item in raw_results]
        else:
            hits = _flatten_scene_search_groups(payload)
        return SceneSearchResult(
            results=hits,
            total_count=payload.get("total_count") or len(hits),
        )

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

坐标不再走 websocket：发出去的 with_geometry 恒为 False，坐标数据 一律由服务端落盘成一个临时 dmf 文件（见 :meth:get_geometry_file），需要坐标 时按实体 id 从该文件里取。

因此本方法只返回基础信息（id / feature / type / color / properties）， geom.geometry 是空的。要坐标请用 :meth:get_geometry_file。

Raises:

Type	Description
ValueError	显式传 with_geometry=True 时。这样调用比返回一堆空坐标更 容易被发现——旧签名还在，静默返回空几何会让调用方以为"这个图层没坐标"。

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。

    坐标**不再**走 websocket：发出去的 ``with_geometry`` 恒为 ``False``，坐标数据
    一律由服务端落盘成一个临时 dmf 文件（见 :meth:`get_geometry_file`），需要坐标
    时按实体 ``id`` 从该文件里取。

    因此本方法只返回**基础信息**（id / feature / type / color / properties），
    ``geom.geometry`` 是空的。要坐标请用 :meth:`get_geometry_file`。

    Raises:
        ValueError: 显式传 ``with_geometry=True`` 时。这样调用比返回一堆空坐标更
            容易被发现——旧签名还在，静默返回空几何会让调用方以为"这个图层没坐标"。
    """
    if with_geometry:
        raise ValueError(
            "get_geometry 不再内联返回坐标：服务端把 with_geometry 当"
            "\"是否内联坐标\"用，传 True 会让坐标走 websocket（正是新协议要避免的）。"
            "请改用 get_geometry_file() 拿落盘 dmf 路径，按实体 id 读取坐标。"
        )
    result = await self._fetch_geometry(file_id, layer_id, entity_types, features)
    return result.entities

get_geometry_file(file_id, layer_id, entity_types=[], features=None) async

获取模型元信息与坐标数据所在的 dmf 文件，不读取该文件。

这是获取几何数据的统一入口：坐标不走 websocket，一律由服务端落盘，本方法把 geometry_file 路径交给调用方，由调用方自己按实体 id 解析该文件。

返回的 entities 只带基础信息（id / feature / type / color / properties）， geometry 为空；坐标在 geometry_file 指向的文件里，按 id 匹配取用 （文件里是整个图层，可能比本次命中的实体更多，不能按顺序对齐）。

Returns:

Type	Description
GeometryGetResult	GeometryGetResult：geometry_file 为 dmf 路径（服务端未产出文件时为
GeometryGetResult	空字符串），entities 为模型元信息列表。

Note

该文件由 Dimine 端管理，可能被后续请求覆盖或清理，需要时应在本次会话内 读取完毕，不要长期持有该路径。

Source code in dimine_python_sdk\conn\conn_client.py

async def get_geometry_file(self, file_id: str, layer_id: str, entity_types: List[str]=[], features: Optional[List[str]] = None) -> GeometryGetResult:
    """获取模型元信息与坐标数据所在的 dmf 文件，**不读取**该文件。

    这是获取几何数据的统一入口：坐标不走 websocket，一律由服务端落盘，本方法把
    ``geometry_file`` 路径交给调用方，由调用方自己按实体 ``id`` 解析该文件。

    返回的 ``entities`` 只带基础信息（id / feature / type / color / properties），
    ``geometry`` 为空；坐标在 ``geometry_file`` 指向的文件里，**按 id 匹配**取用
    （文件里是整个图层，可能比本次命中的实体更多，不能按顺序对齐）。

    Returns:
        ``GeometryGetResult``：``geometry_file`` 为 dmf 路径（服务端未产出文件时为
        空字符串），``entities`` 为模型元信息列表。

    Note:
        该文件由 Dimine 端管理，可能被后续请求覆盖或清理，需要时应在本次会话内
        读取完毕，不要长期持有该路径。
    """
    return await self._fetch_geometry(file_id, layer_id, entity_types, features)

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!")

search_scene(keyword, *, scope=None, files=None, case_sensitive=False) async

场景全文检索

按正则表达式检索当前场景中已打开的文件、图层、要素、实体（含文字标注）。 结果按 文件 → 图层 → 要素 → 实体 的顺序返回。

Parameters:

Name	Type	Description	Default
keyword	str	检索正则表达式，对名称、要素名、实体属性字段名/值、文字内容等进行匹配， 如 ^(BH|ZK)-\d+$ 匹配孔号、.*cu.* 匹配含 cu 的内容	required
scope	Optional[Sequence[SearchScope | str]]	限定检索范围，可选 file | layer | feature | entity， 默认检索全部类型。文字标注属于实体的一部分，检索 entity 时一并命中	None
files	Optional[Sequence[str]]	限定检索的文件（文件id或文件名），默认检索当前场景所有已打开的文件	None
case_sensitive	bool	是否区分大小写，默认 False（不区分）	False

Returns:

Type	Description
SceneSearchResult	检索结果，命中项按 文件 → 图层 → 要素 → 实体 顺序排列

Raises:

Type	Description
Exception	检索失败时抛出异常

Source code in dimine_python_sdk\conn\conn_client.py

async def search_scene(
    self,
    keyword: str,
    *,
    scope: Optional[Sequence[SearchScope | str]] = None,
    files: Optional[Sequence[str]] = None,
    case_sensitive: bool = False,
) -> SceneSearchResult:
    """场景全文检索

    按正则表达式检索当前场景中已打开的文件、图层、要素、实体（含文字标注）。
    结果按 文件 → 图层 → 要素 → 实体 的顺序返回。

    Args:
        keyword: 检索正则表达式，对名称、要素名、实体属性字段名/值、文字内容等进行匹配，
            如 ``^(BH|ZK)-\\d+$`` 匹配孔号、``.*cu.*`` 匹配含 cu 的内容
        scope: 限定检索范围，可选 file | layer | feature | entity，
            默认检索全部类型。文字标注属于实体的一部分，检索 entity 时一并命中
        files: 限定检索的文件（文件id或文件名），默认检索当前场景所有已打开的文件
        case_sensitive: 是否区分大小写，默认 False（不区分）

    Returns:
        检索结果，命中项按 文件 → 图层 → 要素 → 实体 顺序排列

    Raises:
        Exception: 检索失败时抛出异常
    """
    payload: Dict[str, Any] = {
        "keyword": keyword,
        "case_sensitive": case_sensitive,
    }
    if scope:
        payload["scope"] = [
            str(s.value if isinstance(s, SearchScope) else s) for s in scope
        ]
    if files:
        payload["files"] = list(files)

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

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: 操作失败时抛出异常
    """
    await self.set_layers_visibility(
        [{"file": file_id, "layer": layer_name}], action=action
    )

set_layers_visibility(layers, action=VisibilityAction.SHOW) async

批量设置图层显示/隐藏状态

Parameters:

Name	Type	Description	Default
layers	List[Dict[str, str]]	要操作的图层列表，每个元素包含: - file: 文件ID或文件名称 - layer: 图层名称	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_layers_visibility(self, layers: List[Dict[str, str]],
                                 action: VisibilityAction | str = VisibilityAction.SHOW) -> None:
    """批量设置图层显示/隐藏状态

    Args:
        layers: 要操作的图层列表，每个元素包含:
            - file: 文件ID或文件名称
            - layer: 图层名称
        action: 操作类型，可选 show(显示) | hide(隐藏) | exclusive_show(排他性显示)

    Raises:
        Exception: 操作失败时抛出异常
    """
    if not layers:
        return
    payload = {
        "layers": layers,
        "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: Dict[str, Any] = {"id": geom.id}
        # 仅发送显式设置的非 None/空字段
        if geom.file is not None:
            formatted_geom["file"] = geom.file
        if geom.layer is not None:
            formatted_geom["layer"] = geom.layer
        if geom.feature is not None:
            formatted_geom["feature"] = geom.feature
        # entity_type 必须按几何数据判定：服务端会校验实体类型与几何是否匹配
        # （对多段折线发 "line" 会报 entity type is not match）。
        # 未携带几何的单纯属性更新则沿用模型声明的 type（Line 可表达 polyline）。
        if geom.type is not None:
            formatted_geom["entity_type"] = self._resolve_entity_type(geom)
        if geom.color is not None:
            formatted_geom["color"] = geom.color
        # geometry：仅当非空时发送
        if geom.type == "shell" and geom.geometry is not None:
            geo = geom.geometry
            if geo.points is not None and len(geo.points) > 0:
                formatted_geom.setdefault("geometry", {})["points"] = geo.points.tolist()
            if geo.faces is not None and len(geo.faces) > 0:
                formatted_geom.setdefault("geometry", {})["indexs"] = geo.faces.tolist()
        # polyline 也是线：get_geometry 会把服务端下发的 "polyline" 原样带进
        # Line.type，只判 "line" 会让这类实体的坐标在更新时被静默丢掉。
        elif geom.type in ("point", "line", "polyline") and geom.geometry is not None and len(geom.geometry) > 0:
            formatted_geom["geometry"] = geom.geometry.tolist()

        # 处理属性信息
        if geom.properties:
            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, 则将模型添加到指定的文件及图层中"""
    from dimine_python_sdk.models.geometry import Line, Point, Shell

    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,
        )