Function exec_command

custom_components/ssh_command/__init__.py:31–68

(call: ServiceCall)

    default = config.get(DOMAIN) or DEFAULT_SCHEMA({})

    def exec_command(call: ServiceCall) -> dict:
        kwargs = default | call.data
        kwargs["hostname"] = kwargs.pop("host")
        kwargs["username"] = kwargs.pop("user")
        kwargs["password"] = kwargs.pop("pass", None)
        kwargs["key_filename"] = kwargs.pop("private_key", None)

        commands = kwargs.pop("command")
        if isinstance(commands, str):
            commands = [commands]

        client = SSHClient()
        client.set_missing_host_key_policy(AutoAddPolicy())

        try:
            client.connect(**kwargs)

            for command in commands:
                _, stdout, stderr = client.exec_command(
                    command, timeout=kwargs.get("timeout")
                )

            # noinspection PyUnboundLocalVariable
            return {
                "stdout": stdout.read().decode("utf-8"),
                "stderr": stderr.read().decode("utf-8"),
            }

        except TimeoutError as e:
            _LOGGER.error(f"Command execution timeout")
            return {"error": repr(e)}

        except Exception as e:
            _LOGGER.error(f"Failed to connect: {repr(e)}")
            return {"error": repr(e)}

        finally:
            client.close()

    try:
        # ServiceResponse from Hass 2023.7