code refactor

custom_components/easy_computer_manager/__init__.py

@@ -1,9 +1,8 @@
 """The Easy Dualboot Computer Manager integration."""
 
-# Some code is from the official wake_on_lan integration
+# Some snippets of code are from the official wake_on_lan integration (inspiration for this custom component)
 
 from __future__ import annotations
-
 import logging
 from functools import partial
 
@@ -28,10 +27,10 @@ WAKE_ON_LAN_SEND_MAGIC_PACKET_SCHEMA = vol.Schema(
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
-    """Set up the wake on LAN component."""
+    """Set up the Easy Dualboot Computer Manager integration."""
 
     async def send_magic_packet(call: ServiceCall) -> None:
-        """Send magic packet to wake up a device."""
+        """Send a magic packet to wake up a device."""
         mac_address = call.data.get(CONF_MAC)
         broadcast_address = call.data.get(CONF_BROADCAST_ADDRESS)
         broadcast_port = call.data.get(CONF_BROADCAST_PORT)
@@ -43,7 +42,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
             service_kwargs["port"] = broadcast_port
 
         _LOGGER.info(
-            "Send magic packet to mac %s (broadcast: %s, port: %s)",
+            "Sending magic packet to MAC %s (broadcast: %s, port: %s)",
             mac_address,
             broadcast_address,
             broadcast_port,
@@ -70,7 +69,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
 
 async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
-    """Unload a config entry."""
+    """Unload the Easy Dualboot Computer Manager integration."""
     return await hass.config_entries.async_forward_entry_unload(
         entry, "switch"
     )

custom_components/easy_computer_manager/config_flow.py

@@ -7,6 +7,7 @@ from typing import Any
 import voluptuous as vol
 from homeassistant import config_entries, exceptions
 from homeassistant.core import HomeAssistant
+
 from paramiko.ssh_exception import AuthenticationException
 
 from . import utils
@@ -48,8 +49,7 @@ class Hub:
     async def test_connection(self) -> bool:
         """Test connectivity to the computer is OK."""
         try:
-            return utils.test_connection(
-                utils.create_ssh_connection(self._host, self._username, self._password, self._port))
+            return utils.test_connection(utils.create_ssh_connection(self._host, self._username, self._password, self._port))
         except AuthenticationException:
             return False
 
@@ -61,10 +61,9 @@ async def validate_input(hass: HomeAssistant, data: dict) -> dict[str, Any]:
     if len(data["host"]) < 3:
         raise InvalidHost
 
-    hub = Hub(hass, data["host"], data["username"], data["password"], data["port"])
+    hub = Hub(hass, **data)
 
-    result = await hub.test_connection()
-    if not result:
+    if not await hub.test_connection():
         raise CannotConnect
 
     return {"title": data["host"]}
@@ -81,23 +80,14 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
         if user_input is not None:
             try:
                 info = await validate_input(self.hass, user_input)
-
                 return self.async_create_entry(title=info["title"], data=user_input)
-            except AuthenticationException:
-                errors["host"] = "cannot_connect"
-            except CannotConnect:
-                errors["base"] = "cannot_connect"
-            except InvalidHost:
-                errors["host"] = "cannot_connect"
-            except Exception:  # pylint: disable=broad-except
-                _LOGGER.exception("Unexpected exception")
+            except (AuthenticationException, CannotConnect, InvalidHost) as ex:
+                errors["base"] = str(ex)
+            except Exception as ex:  # pylint: disable=broad-except
+                _LOGGER.exception("Unexpected exception: %s", ex)
                 errors["base"] = "unknown"
 
-        # If there is no user input or there were errors, show the form again, including any errors that were found
-        # with the input.
-        return self.async_show_form(
-            step_id="user", data_schema=DATA_SCHEMA, errors=errors
-        )
+        return self.async_show_form(step_id="user", data_schema=DATA_SCHEMA, errors=errors)
 
 
 class CannotConnect(exceptions.HomeAssistantError):

custom_components/easy_computer_manager/const.py

@@ -10,3 +10,4 @@ SERVICE_RESTART_COMPUTER = "restart_computer"
 SERVICE_CHANGE_MONITORS_CONFIG = "change_monitors_config"
 SERVICE_STEAM_BIG_PICTURE = "steam_big_picture"
 SERVICE_CHANGE_AUDIO_CONFIG = "change_audio_config"
+SERVICE_DEBUG_INFO = "debug_info"

custom_components/easy_computer_manager/services.yaml

@@ -1,6 +1,6 @@
 send_magic_packet:
   name: Send Magic Packet
-  description: Send a 'magic packet' to awaken a device with 'Wake-On-LAN' capabilities.
+  description: Send a 'magic packet' to wake up a device with 'Wake-On-LAN' capabilities.
   fields:
     mac:
       name: MAC Address
@@ -139,3 +139,11 @@ change_audio_config:
       example: "Starship/Matisse HD Audio Controller Stereo Analog"
       selector:
         text:
+
+debug_info:
+  name: Debug Information
+  description: Display debug information to help with setup and troubleshooting. You can use this data (such as monitor resolutions, audio device names/IDs, etc.) with others services such as change_audio_config or change_monitors_config
+  target:
+    entity:
+      integration: easy_computer_manager
+      domain: switch

custom_components/easy_computer_manager/switch.py

@@ -23,7 +23,7 @@ from homeassistant.const import (
     CONF_PASSWORD,
     CONF_PORT,
     CONF_USERNAME, )
-from homeassistant.core import HomeAssistant
+from homeassistant.core import HomeAssistant, ServiceResponse, SupportsResponse
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import (
     config_validation as cv,
@@ -37,7 +37,7 @@ from paramiko.ssh_exception import AuthenticationException
 from . import utils
 from .const import SERVICE_RESTART_TO_WINDOWS_FROM_LINUX, SERVICE_PUT_COMPUTER_TO_SLEEP, \
     SERVICE_START_COMPUTER_TO_WINDOWS, SERVICE_RESTART_COMPUTER, SERVICE_RESTART_TO_LINUX_FROM_WINDOWS, \
-    SERVICE_CHANGE_MONITORS_CONFIG, SERVICE_STEAM_BIG_PICTURE, SERVICE_CHANGE_AUDIO_CONFIG
+    SERVICE_CHANGE_MONITORS_CONFIG, SERVICE_STEAM_BIG_PICTURE, SERVICE_CHANGE_AUDIO_CONFIG, SERVICE_DEBUG_INFO
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -139,6 +139,14 @@ async def async_setup_entry(
         SERVICE_CHANGE_AUDIO_CONFIG,
     )
 
+    # Register the service to print debug info
+    platform.async_register_entity_service(
+        SERVICE_DEBUG_INFO,
+        {},
+        SERVICE_DEBUG_INFO,
+        supports_response=SupportsResponse.ONLY
+    )
+
 
 class ComputerSwitch(SwitchEntity):
     """Representation of a computer switch."""
@@ -335,3 +343,7 @@ class ComputerSwitch(SwitchEntity):
                 )
 
             self._state = False
+
+    def debug_info(self) -> ServiceResponse:
+        """Prints debug info."""
+        return utils.get_debug_info(self._connection)

custom_components/easy_computer_manager/utils.py

@@ -263,32 +263,61 @@ def silent_install_nircmd(connection: Connection):
                 return
 
 
-def parse_gnome_monitor_config(output):
-    # SHOULD NOT BE USED YET, STILL IN DEVELOPMENT
+def get_monitors_config(connection: Connection) -> dict:
     """Parse the output of the gnome-monitor-config command to get the current monitor configuration."""
 
-    monitors = []
-    current_monitor = None
+    if is_unix_system(connection):
+        result = connection.run("gnome-monitor-config list")
+        if result.return_code != 0:
+            raise HomeAssistantError(f"Could not get monitors config on system running at {connection.host}.")
 
-    for line in output.split('\n'):
-        monitor_match = re.match(r'^Monitor \[ (.+?) \] (ON|OFF)$', line)
-        if monitor_match:
-            if current_monitor:
-                monitors.append(current_monitor)
-            source, status = monitor_match.groups()
-            current_monitor = {'source': source, 'status': status, 'names': [], 'resolutions': []}
-        elif current_monitor:
-            display_name_match = re.match(r'^\s+display-name: (.+)$', line)
-            resolution_match = re.match(r'^\s+(\d+x\d+@\d+(?:\.\d+)?).*$', line)
-            if display_name_match:
-                current_monitor['names'].append(display_name_match.group(1).replace('"', ''))
-            elif resolution_match:
-                current_monitor['resolutions'].append(resolution_match.group(1))
+        monitors = []
+        current_monitor = None
 
-    if current_monitor:
-        monitors.append(current_monitor)
+        for line in result.stdout.split('\n'):
+            monitor_match = re.match(r'^Monitor \[ (.+?) \] (ON|OFF)$', line)
+            if monitor_match:
+                if current_monitor:
+                    monitors.append(current_monitor)
+                source, status = monitor_match.groups()
+                current_monitor = {'source': source, 'status': status, 'names': [], 'resolutions': []}
+            elif current_monitor:
+                display_name_match = re.match(r'^\s+display-name: (.+)$', line)
+                resolution_match = re.match(r'^\s+(\d+x\d+@\d+(?:\.\d+)?).*$', line)
+                if display_name_match:
+                    current_monitor['names'].append(display_name_match.group(1).replace('"', ''))
+                elif resolution_match:
+                    # Don't include resolutions under 1280x720
+                    if int(resolution_match.group(1).split('@')[0].split('x')[0]) >= 1280:
 
-    return monitors
+                        # If there are already resolutions in the list, check if the framerate between the last is >1
+                        if len(current_monitor['resolutions']) > 0:
+                            last_resolution = current_monitor['resolutions'][-1]
+                            last_resolution_size = last_resolution.split('@')[0]
+                            this_resolution_size = resolution_match.group(1).split('@')[0]
+
+                            # Only truncate some framerates if the resolution are the same
+                            if last_resolution_size == this_resolution_size:
+                                last_resolution_framerate = float(last_resolution.split('@')[1])
+                                this_resolution_framerate = float(resolution_match.group(1).split('@')[1])
+
+                                # If the difference between the last resolution framerate and this one is >1, ignore it
+                                if last_resolution_framerate - 1 > this_resolution_framerate:
+                                    current_monitor['resolutions'].append(resolution_match.group(1))
+                            else:
+                                # If the resolution is different, this adds the new resolution
+                                # to the list without truncating
+                                current_monitor['resolutions'].append(resolution_match.group(1))
+                        else:
+                            # This is the first resolution, add it to the list
+                            current_monitor['resolutions'].append(resolution_match.group(1))
+
+        if current_monitor:
+            monitors.append(current_monitor)
+
+        return monitors
+    else:
+        raise HomeAssistantError("Not implemented yet for Windows OS.")
 
 
 def steam_big_picture(connection: Connection, action: str):
@@ -420,3 +449,39 @@ def change_audio_config(connection: Connection, volume: int, mute: bool, input_d
                     f"Could not change audio config on system running on {connection.host}, check logs with debug")
     else:
         raise HomeAssistantError("Not implemented yet for Windows OS.")
+
+
+def get_debug_info(connection: Connection):
+    """Return debug information about the host system."""
+
+    data = {}
+
+    data_os = {
+        'name': get_operating_system(connection),
+        'version': get_operating_system_version(connection),
+        'is_unix': is_unix_system(connection)
+    }
+
+    data_ssh = {
+        'is_connected': connection.is_connected,
+        'username': connection.user,
+        'host': connection.host,
+        'port': connection.port
+    }
+
+    data_grub = {
+        'windows_entry': get_windows_entry_in_grub(connection)
+    }
+
+    data_audio = {
+        'speakers': get_audio_config(connection).get('sinks'),
+        'microphones': get_audio_config(connection).get('sources')
+    }
+
+    data['os'] = data_os
+    data['ssh'] = data_ssh
+    data['grub'] = data_grub
+    data['audio'] = data_audio
+    data['monitors'] = get_monitors_config(connection)
+
+    return data