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_request(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_request(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_request(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_request(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_request(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]) -> None:
        """创建模型

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

        Raises:
            Exception: 创建失败时抛出异常
        """
        # 格式化模型数据
        formatted_geometries = []
        for geom in geometries:
            formatted_geom = {
                "file": geom.file,
                "layer": geom.layer,
                "feature": geom.feature,
                "entity_type": geom.type,
                "geometry": geom.geometry.astype("str").tolist() if geom.type!="shell" else {"indexs": geom.geometry.faces.tolist(), "points": geom.geometry.points.astype("str").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_request(CommandType.CREATE_GEOMETRY, formatted_geometries)
        if response.status == ResponseStatus.FAIL:
            raise Exception(response.error)
        # 更新id等信息
        for i, item in enumerate(response.payload):
            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) -> List[GEOMETRY]:
        """获取模型"""
        # 先获取要素信息，
        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
        }
        if features:
            payload["feature"] = features

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

        def _format_value(type:str, value: Any):
            if value == "" and type != "string":
                return None
            elif type == "double":
                return float(value)
            elif type == "string":
                return str(value)
            elif type in ["int","long","short"]:
                return int(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.astype("str").tolist() if geom.type!="shell" else {"indexs": geom.geometry.faces.tolist(), "points": geom.geometry.points.astype("str").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_request(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_request(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) -> None:
        """创建或更新多个模型, 如果指定了file_id和layer_id, 则将模型添加到指定的文件及图层中"""
        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)


    # 图层相关方法
    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_request(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_request(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_request(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: 属性类型 ("string", "float", "int", "long", "short")

        Raises:
            Exception: 创建失败时抛出异常
        """
        payload = {
            "file": file_id,
            "feature_name": feature_name,
            "feature_define": feature_define
        }
        response = await self.send_request(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_request(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: 属性类型 ("string", "float", "int", "long", "short")

        Raises:
            Exception: 更新失败时抛出异常
        """
        payload = {
            "file": file_id,
            "feature_name": feature_name,
            "feature_define": feature_define
        }
        response = await self.send_request(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_request(CommandType.DELETE_FEATURE, 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_request(
            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)
        )

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_request(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: 属性类型 ("string", "float", "int", "long", "short")	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: 属性类型 ("string", "float", "int", "long", "short")

    Raises:
        Exception: 创建失败时抛出异常
    """
    payload = {
        "file": file_id,
        "feature_name": feature_name,
        "feature_define": feature_define
    }
    response = await self.send_request(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_request(CommandType.FILE_CREATE, {"file_path": file_path})
    if res.status == ResponseStatus.FAIL:
        raise Exception(res.error)

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

Raises:

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

Source code in dimine_python_sdk\conn\conn_client.py

async def create_geometry(self, geometries: List[GEOMETRY]) -> None:
    """创建模型

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

    Raises:
        Exception: 创建失败时抛出异常
    """
    # 格式化模型数据
    formatted_geometries = []
    for geom in geometries:
        formatted_geom = {
            "file": geom.file,
            "layer": geom.layer,
            "feature": geom.feature,
            "entity_type": geom.type,
            "geometry": geom.geometry.astype("str").tolist() if geom.type!="shell" else {"indexs": geom.geometry.faces.tolist(), "points": geom.geometry.points.astype("str").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_request(CommandType.CREATE_GEOMETRY, formatted_geometries)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)
    # 更新id等信息
    for i, item in enumerate(response.payload):
        geometries[i].id=item.get("id")
        geometries[i].related=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_request(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_request(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_request(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_request(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_request(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_request(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_request(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) async

获取模型

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) -> List[GEOMETRY]:
    """获取模型"""
    # 先获取要素信息，
    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
    }
    if features:
        payload["feature"] = features

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

    def _format_value(type:str, value: Any):
        if value == "" and type != "string":
            return None
        elif type == "double":
            return float(value)
        elif type == "string":
            return str(value)
        elif type in ["int","long","short"]:
            return int(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_request(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_request(
        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_request(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!")

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: 属性类型 ("string", "float", "int", "long", "short")	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: 属性类型 ("string", "float", "int", "long", "short")

    Raises:
        Exception: 更新失败时抛出异常
    """
    payload = {
        "file": file_id,
        "feature_name": feature_name,
        "feature_define": feature_define
    }
    response = await self.send_request(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.astype("str").tolist() if geom.type!="shell" else {"indexs": geom.geometry.faces.tolist(), "points": geom.geometry.points.astype("str").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_request(CommandType.UPDATE_GEOMETRY, formatted_geometries)
    if response.status == ResponseStatus.FAIL:
        raise Exception(response.error)

update_or_create_geometry(geometries) 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) -> None:
    """创建或更新多个模型, 如果指定了file_id和layer_id, 则将模型添加到指定的文件及图层中"""
    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)