跳转至

Serial Application Service

controller.application.services.serial_application_service

串口应用服务 - Application层 只负责串口连接管理和状态协调,不处理数据收发 数据收发由各个Domain服务直接使用SerialDomainService

SerialApplicationService

SerialApplicationService(message_display: MessageDisplay, serial_domain_service: SerialDomainService)

Bases: BaseService

串口应用服务。

只负责串口连接管理和状态协调,不处理数据收发。 数据收发由各个Domain服务直接使用SerialDomainService。

属性:

名称 类型 描述
connection_status_changed pyqtSignal

串口连接状态变化信号。

port_list_updated pyqtSignal

端口列表更新信号。

初始化串口应用服务。

源代码位于: src/controller/controller/application/services/serial_application_service.py
27
28
29
30
31
32
33
34
def __init__(self, message_display: MessageDisplay, serial_domain_service: SerialDomainService):
    """初始化串口应用服务。"""
    super().__init__(message_display)

    self.serial_domain_service = serial_domain_service

    # 连接Domain层信号到Application层
    self._connect_domain_signals()

refresh_ports

refresh_ports() -> List[str]

刷新可用端口列表。

返回:

类型 描述
List[str]

List[str]: 可用端口名称列表。

源代码位于: src/controller/controller/application/services/serial_application_service.py
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def refresh_ports(self) -> List[str]:
    """刷新可用端口列表。

    Returns:
        List[str]: 可用端口名称列表。
    """
    try:
        ports = self.serial_domain_service.scan_available_ports()
        self.port_list_updated.emit(ports)
        self._display_message(f"发现 {len(ports)} 个可用端口", "系统")
        return ports
    except Exception as e:
        self._display_message(f"刷新端口失败: {str(e)}", "错误")
        return []

get_port_info

get_port_info(port_name: str) -> Optional[Dict[str, str]]

获取指定端口信息。

参数:

名称 类型 描述 默认
port_name str

端口名称。

必需

返回:

类型 描述
Optional[Dict[str, str]]

Optional[Dict[str, str]]: 端口信息字典,失败时返回None。

源代码位于: src/controller/controller/application/services/serial_application_service.py
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def get_port_info(self, port_name: str) -> Optional[Dict[str, str]]:
    """获取指定端口信息。

    Args:
        port_name (str): 端口名称。

    Returns:
        Optional[Dict[str, str]]: 端口信息字典,失败时返回None。
    """
    try:
        return self.serial_domain_service.get_port_info(port_name)
    except Exception as e:
        self._display_message(f"获取端口信息失败: {str(e)}", "错误")
        return None

get_available_ports

get_available_ports() -> List[str]

获取当前可用端口列表(不刷新)。

返回:

类型 描述
List[str]

List[str]: 可用端口名称列表。

源代码位于: src/controller/controller/application/services/serial_application_service.py
66
67
68
69
70
71
72
73
74
75
76
def get_available_ports(self) -> List[str]:
    """获取当前可用端口列表(不刷新)。

    Returns:
        List[str]: 可用端口名称列表。
    """
    try:
        return self.serial_domain_service.scan_available_ports()
    except Exception as e:
        self._display_message(f"获取端口列表失败: {str(e)}", "错误")
        return []

get_current_port

get_current_port() -> Optional[str]

获取当前连接的端口名称。

返回:

类型 描述
Optional[str]

Optional[str]: 端口名称,未连接时返回None。

源代码位于: src/controller/controller/application/services/serial_application_service.py
78
79
80
81
82
83
84
def get_current_port(self) -> Optional[str]:
    """获取当前连接的端口名称。

    Returns:
        Optional[str]: 端口名称,未连接时返回None。
    """
    return self.serial_domain_service.get_current_port()

connect_serial

connect_serial(port: str, config: Dict[str, Any]) -> bool

连接串口。

参数:

名称 类型 描述 默认
port str

串口名称。

必需
config Dict[str, Any]

串口配置参数。

必需

返回:

名称 类型 描述
bool bool

连接是否成功。

源代码位于: src/controller/controller/application/services/serial_application_service.py
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def connect_serial(self, port: str, config: Dict[str, Any]) -> bool:
    """连接串口。

    Args:
        port (str): 串口名称。
        config (Dict[str, Any]): 串口配置参数。

    Returns:
        bool: 连接是否成功。
    """
    self._display_message(f"连接串口: {port}, 配置: {config}", "系统")
    try:
        success = self.serial_domain_service.connect_port(port, config)
        if success:
            self._display_message(f"成功连接到端口 {port}", "系统")
        return success
    except Exception as e:
        self._display_message(f"连接失败: {str(e)}", "错误")
        return False

disconnect_serial

disconnect_serial() -> bool

断开串口连接。

返回:

名称 类型 描述
bool bool

断开是否成功。

源代码位于: src/controller/controller/application/services/serial_application_service.py
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def disconnect_serial(self) -> bool:
    """断开串口连接。

    Returns:
        bool: 断开是否成功。
    """
    try:
        success = self.serial_domain_service.disconnect_port()
        if success:
            self._display_message("已断开串口连接", "系统")
        return success
    except Exception as e:
        self._display_message(f"断开连接失败: {str(e)}", "错误")
        return False

is_connected

is_connected() -> bool

检查串口是否已连接。

返回:

名称 类型 描述
bool bool

是否已连接。

源代码位于: src/controller/controller/application/services/serial_application_service.py
121
122
123
124
125
126
127
def is_connected(self) -> bool:
    """检查串口是否已连接。

    Returns:
        bool: 是否已连接。
    """
    return self.serial_domain_service.is_connected()

cleanup

cleanup() -> None

清理资源。

源代码位于: src/controller/controller/application/services/serial_application_service.py
145
146
147
148
149
150
151
def cleanup(self) -> None:
    """清理资源。"""
    try:
        self._disconnect_domain_signals()
        self.serial_domain_service.cleanup()
    except Exception:
        pass  # 忽略清理时的异常