From 6f3544fa47f0714f9defa88c90e35d67f0b22992 Mon Sep 17 00:00:00 2001 From: Marc Mueller <30130371+cdce8p@users.noreply.github.com> Date: Fri, 3 Jan 2025 02:53:08 +0100 Subject: [PATCH] Add types package for pexpect (#134461) --- .../components/aruba/device_tracker.py | 6 ++-- .../components/cisco_ios/device_tracker.py | 16 ++++------- .../components/pandora/media_player.py | 28 +++++++++---------- requirements_test.txt | 1 + 4 files changed, 24 insertions(+), 27 deletions(-) diff --git a/homeassistant/components/aruba/device_tracker.py b/homeassistant/components/aruba/device_tracker.py index ef622ef9826..911fab441e5 100644 --- a/homeassistant/components/aruba/device_tracker.py +++ b/homeassistant/components/aruba/device_tracker.py @@ -90,7 +90,7 @@ class ArubaDeviceScanner(DeviceScanner): """Retrieve data from Aruba Access Point and return parsed result.""" connect = f"ssh {self.username}@{self.host} -o HostKeyAlgorithms=ssh-rsa" - ssh = pexpect.spawn(connect) + ssh: pexpect.spawn[str] = pexpect.spawn(connect, encoding="utf-8") query = ssh.expect( [ "password:", @@ -125,12 +125,12 @@ class ArubaDeviceScanner(DeviceScanner): ssh.expect("#") ssh.sendline("show clients") ssh.expect("#") - devices_result = ssh.before.split(b"\r\n") + devices_result = (ssh.before or "").splitlines() ssh.sendline("exit") devices: dict[str, dict[str, str]] = {} for device in devices_result: - if match := _DEVICES_REGEX.search(device.decode("utf-8")): + if match := _DEVICES_REGEX.search(device): devices[match.group("ip")] = { "ip": match.group("ip"), "mac": match.group("mac").upper(), diff --git a/homeassistant/components/cisco_ios/device_tracker.py b/homeassistant/components/cisco_ios/device_tracker.py index 1f78f95c259..b882f046a8e 100644 --- a/homeassistant/components/cisco_ios/device_tracker.py +++ b/homeassistant/components/cisco_ios/device_tracker.py @@ -3,7 +3,6 @@ from __future__ import annotations import logging -import re from pexpect import pxssh import voluptuous as vol @@ -101,11 +100,11 @@ class CiscoDeviceScanner(DeviceScanner): return False - def _get_arp_data(self): + def _get_arp_data(self) -> str | None: """Open connection to the router and get arp entries.""" try: - cisco_ssh = pxssh.pxssh() + cisco_ssh: pxssh.pxssh[str] = pxssh.pxssh(encoding="uft-8") cisco_ssh.login( self.host, self.username, @@ -115,12 +114,11 @@ class CiscoDeviceScanner(DeviceScanner): ) # Find the hostname - initial_line = cisco_ssh.before.decode("utf-8").splitlines() + initial_line = (cisco_ssh.before or "").splitlines() router_hostname = initial_line[len(initial_line) - 1] router_hostname += "#" # Set the discovered hostname as prompt - regex_expression = f"(?i)^{router_hostname}".encode() - cisco_ssh.PROMPT = re.compile(regex_expression, re.MULTILINE) + cisco_ssh.PROMPT = f"(?i)^{router_hostname}" # Allow full arp table to print at once cisco_ssh.sendline("terminal length 0") cisco_ssh.prompt(1) @@ -128,13 +126,11 @@ class CiscoDeviceScanner(DeviceScanner): cisco_ssh.sendline("show ip arp") cisco_ssh.prompt(1) - devices_result = cisco_ssh.before - - return devices_result.decode("utf-8") except pxssh.ExceptionPxssh as px_e: _LOGGER.error("Failed to login via pxssh: %s", px_e) + return None - return None + return cisco_ssh.before def _parse_cisco_mac_address(cisco_hardware_addr): diff --git a/homeassistant/components/pandora/media_player.py b/homeassistant/components/pandora/media_player.py index a231ba1b596..eea88da6e36 100644 --- a/homeassistant/components/pandora/media_player.py +++ b/homeassistant/components/pandora/media_player.py @@ -92,13 +92,13 @@ class PandoraMediaPlayer(MediaPlayerEntity): self._attr_source_list = [] self._time_remaining = 0 self._attr_media_duration = 0 - self._pianobar: pexpect.spawn | None = None + self._pianobar: pexpect.spawn[str] | None = None def turn_on(self) -> None: """Turn the media player on.""" if self.state != MediaPlayerState.OFF: return - self._pianobar = pexpect.spawn("pianobar") + self._pianobar = pexpect.spawn("pianobar", encoding="utf-8") _LOGGER.debug("Started pianobar subprocess") mode = self._pianobar.expect( ["Receiving new playlist", "Select station:", "Email:"] @@ -135,8 +135,9 @@ class PandoraMediaPlayer(MediaPlayerEntity): self._pianobar.terminate() except pexpect.exceptions.TIMEOUT: # kill the process group - os.killpg(os.getpgid(self._pianobar.pid), signal.SIGTERM) - _LOGGER.debug("Killed Pianobar subprocess") + if (pid := self._pianobar.pid) is not None: + os.killpg(os.getpgid(pid), signal.SIGTERM) + _LOGGER.debug("Killed Pianobar subprocess") self._pianobar = None self._attr_state = MediaPlayerState.OFF self.schedule_update_ha_state() @@ -209,7 +210,7 @@ class PandoraMediaPlayer(MediaPlayerEntity): try: match_idx = self._pianobar.expect( [ - rb"(\d\d):(\d\d)/(\d\d):(\d\d)", + r"(\d\d):(\d\d)/(\d\d):(\d\d)", "No song playing", "Select station", "Receiving new playlist", @@ -222,21 +223,20 @@ class PandoraMediaPlayer(MediaPlayerEntity): self._log_match() if match_idx == 1: # idle. - response = None - elif match_idx == 2: + return None + if match_idx == 2: # stuck on a station selection dialog. Clear it. _LOGGER.warning("On unexpected station list page") self._pianobar.sendcontrol("m") # press enter self._pianobar.sendcontrol("m") # do it again b/c an 'i' got in self.update_playing_status() - response = None - elif match_idx == 3: + return None + if match_idx == 3: _LOGGER.debug("Received new playlist list") self.update_playing_status() - response = None - else: - response = self._pianobar.before.decode("utf-8") - return response + return None + + return self._pianobar.before def _update_current_station(self, response: str) -> None: """Update current station.""" @@ -307,7 +307,7 @@ class PandoraMediaPlayer(MediaPlayerEntity): """List defined Pandora stations.""" assert self._pianobar is not None self._send_station_list_command() - station_lines = self._pianobar.before.decode("utf-8") + station_lines = self._pianobar.before or "" _LOGGER.debug("Getting stations: %s", station_lines) self._attr_source_list = [] for line in station_lines.split("\r\n"): diff --git a/requirements_test.txt b/requirements_test.txt index ce9743d758e..b3a50bd96a6 100644 --- a/requirements_test.txt +++ b/requirements_test.txt @@ -42,6 +42,7 @@ types-caldav==1.3.0.20241107 types-chardet==0.1.5 types-decorator==5.1.8.20240310 types-paho-mqtt==1.6.0.20240321 +types-pexpect==4.9.0.20241208 types-pillow==10.2.0.20240822 types-protobuf==5.29.1.20241207 types-psutil==6.1.0.20241221