midgard_progress_monitor.py

import re
import subprocess
import time

import paramiko
import os

# SSH host to connect to
ssh_host = 'thornode'

# Kubernetes command to monitor logs
kubectl_command = 'kubectl logs midgard-0 -f --tail=1'

# Regular expression pattern to extract progress
progress_pattern = r"progress=(\d+\.\d+)%"

# Progress step to voice the progress
progress_step = 0.5  # %

# Recv buffer size
buff_size = 1024


# Function to parse ~/.ssh/config file
def parse_ssh_config():
    ssh_config = paramiko.SSHConfig()

    # Path to the ~/.ssh/config file
    ssh_config_file = os.path.expanduser('~/.ssh/config')

    # Parse the SSH config file
    with open(ssh_config_file) as f:
        ssh_config.parse(f)

    # Get all the host entries from the parsed config
    return ssh_config


# Function to voice the progress using macOS `say` command
def voice_progress(text):
    # Call macOS `say` command to voice the progress
    subprocess.run(['say', text])


# Regular expression pattern to remove ANSI escape sequences and control characters
ansi_escape = re.compile(r'(?:\x1B[@-_][0-?]*[ -/]*[@-~])|[\x00-\x1f\x7f]')


def clean_log_line(line):
    """
    Remove all ANSI escape sequences and control characters from the log line.
    """
    return ansi_escape.sub('', line)


def ping_sound():
    """
    Play a system sound to indicate the process has finished.
    """
    os.system("afplay /System/Library/Sounds/Ping.aiff")


def main():
    # Establish SSH connection
    client = paramiko.SSHClient()
    # client.load_system_host_keys()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    # Get all the host entries from the parsed config
    config = parse_ssh_config()
    # Get SSH configuration for the specified host
    host_config = config.lookup(ssh_host)

    try:
        client.connect(
            hostname=host_config['hostname'],
            port=int(host_config.get('port', 22)),
            username=host_config.get('user'),
            key_filename=host_config.get('identityfile'),
            passphrase=host_config.get('identityfile'),
            timeout=30
        )

        # Open a SSH session
        ssh_session = client.get_transport().open_session()
        print("Connected to the SSH server.")

        # Execute the kubectl command
        print(f"Executing command: {kubectl_command}")
        ssh_session.exec_command(kubectl_command)

        prev_progress = 0.0

        # Read and monitor the command output
        while True:
            if ssh_session.recv_ready():
                output = ssh_session.recv(buff_size).decode('utf-8')

                print(output)  # Print the log output for debugging

                output_clean = clean_log_line(output)
                # Search for progress in the output
                match = re.search(progress_pattern, output_clean)
                if match:
                    progress = float(match.group(1))

                    print(f"Progress: {progress}%")

                    if progress >= 100.0:
                        print("Finished!")
                        ping_sound()
                        voice_progress("The process has finished!")
                        break

                    if progress - prev_progress >= progress_step:
                        prev_progress = progress
                        voice_progress(f'The progress is {progress} percent')

            time.sleep(1)  # Sleep for 1 second before checking again

    except paramiko.AuthenticationException:
        print("Authentication failed. Please check your credentials.")
    except paramiko.SSHException as ssh_ex:
        print(f"SSH error: {ssh_ex}")
    except Exception as ex:
        print(f"Error: {ex}")
    finally:
        client.close()


if __name__ == "__main__":
    main()

Don't forget to pip install paramiko