AutoSyncHost/ashost.py

#!/usr/bin/env python3
import os
### Copyright 2024 Thomas Herrmann
### Released under GPL v3

import time
import mido

def check_devices() -> tuple:
    # Retrieve the list of available input devices
    devices_in = mido.get_input_names()
    devices_out = mido.get_output_names()

    devices_in = remove_through_ports(devices_in)
    devices_out = remove_through_ports(devices_out)
    print("Available MIDI input devices:", devices_in)

    return devices_in, devices_out



def match_device_name(input_name, output_list):
    # Function to match device names flexibly
    for device in output_list:
        if input_name == device:
            return device

        #If name differs
        elif any(substring in device for substring in input_name.split()):
            return device
    return None

def remove_through_ports(ports):
    # Filter out ports containing "Through" in their names
    filtered_ports = [port for port in ports if 'through' not in port.lower()]

    return filtered_ports

def determine_host(devices_in: list) -> str:
    master = None
    """
    Determine the host by listening for 'start' messages on input ports.

    Args:
    - devices_in (list of str): The names of the MIDI input ports to listen to.

    Returns:
    - str: The name of the input port that received the 'start' message.
    """
    # Open all the input ports
    inports = [mido.open_input(device) for device in devices_in]

    print("Listening for messages...")

    try:
        while not master:
            for port in inports:
                for message in port.iter_pending():
                    #print(f"Received message: ({port.name}, {message})")
                    if 'start' in str(message) or 'continue' in str(message):
                        master = port.name
                        break

    except KeyboardInterrupt:
        print("Interrupted by user.")

    finally:
        # Ensure all ports are closed, important to not get a stuck script without errors
        for port in inports:
            print("Inport closing", port)
            port.close()
        print("Closed all input ports.")
        if master:
            return master


def relay_midi_host_to_out(host_name, devices_out):
    """
    Relay MIDI messages from a specific input port to all specified output ports.

    Args:
    - host_name (str): The name of the MIDI input port to listen to.
    - devices_out (list of str): The names of the MIDI output ports to send messages to.
    """
    #host_name = f'{host_name}'
    reset = False
    # Open the input port
    print("Opening Input", host_name)

    input_port = mido.open_input(host_name)
    output_ports = [mido.open_output(device) for device in devices_out]

    print(f"Listening for messages on {host_name} and relaying to {devices_out}")
    recent_messages = []

    os.system(f'speaker-test -t sine -f 261 -l 1')

    try:
        last_stop_time = 0
        debounce_delay = 0.05 # 50 milliseconds debounce time

        while not reset:
            # Process all pending messages
            for message in input_port.iter_pending():
                read_message = str(message)
                #current_time = time.time()

                if 'clock' in read_message:
                    for dev in output_ports:
                        dev.send(message)

                if 'start' in read_message or 'continue' in read_message:
                    for dev in output_ports:
                        dev.send(message)
                        recent_messages.append(read_message)

                elif 'stop' in read_message:
                    #if current_time - last_stop_time > debounce_delay:
                    for dev in output_ports:
                        dev.send(message)
                    recent_messages.append(read_message)
                        #last_stop_time = current_time
                #print(recent_messages)

                # Keep only the last 3 messages
                if len(recent_messages) > 3:
                    recent_messages.pop(0)

                # Check if the last 3 messages are all 'stop'
                if len(recent_messages) >= 3 and all(msg.startswith('stop') for msg in recent_messages[-3:]):
                    print("Exiting", recent_messages)
                    recent_messages.clear()
                    reset = True
                    break

    except KeyboardInterrupt:
        # Handle user interrupt (e.g., Ctrl+C)
        print("MIDI relay interrupted by user.")

    finally:
        # Close all ports
        input_port.close()
        print(input_port)

        for port in output_ports:
            port.close()
            print(port)
        print("Closed all MIDI ports.")

        #Repeat host detection a nd main loop
        time.sleep(1)
        main_run()

def main_run():

    devices_in, devices_out = check_devices()

    # Check if at least one and one output device is connected
    while len(devices_in) < 1 and len(devices_out) < 1:
        time.sleep(3)
        main_run()

    base_frequency = 130  # Starting frequency in Hz
    increment = 50  # Frequency increment for each device

    for i in range(len(devices_out)):
        # Calculate the frequency for the current device
        frequency = base_frequency + (i * increment)

        # Play the beep sound with the calculated frequency
        os.system(f'speaker-test -t sine -f {frequency} -l 1')

    # Listen for devices that send start and return the host name
    midi_host_in = determine_host(devices_in)
    print(midi_host_in)

    # Find and remove the matching output device to avoid feedback
    matching_output_device = match_device_name(midi_host_in, devices_out)
    print("Filtering", matching_output_device)

    if matching_output_device:
        devices_out.remove(matching_output_device)

    if midi_host_in:
        relay_midi_host_to_out(midi_host_in, devices_out)

main_run()