mt5-manager/mt5_control.py

'''
File: mt5_control.py
Description: Логика управления запуском и остановкой терминалов MetaTrader 5
'''

__version__ = '0.1.0'

import subprocess
import psutil


# Путь к папке с экземплярами терминалов
MT5_FOLDER = 'C:/MT5'

# Имя запускаемого файла терминала
MT5_EXE = 'terminal64.exe'


def instance_path(name: str) -> str:
    '''
    Формирует полный путь по имени экземпляра терминала

    Args:
        name (str): Имя экземпляра терминала.

    Returns:
        path (str): Полый путь к запускаемому файлу экземпляра терминала.
    '''
    return f'{MT5_FOLDER}/{name}/{MT5_EXE}'


# Список имён экземпляров (пока фиксированный)
instances_names = ['MetaTrader5.1', 'MetaTrader5.2', 'MetaTrader5.3']

# Словарь соответствия имя -> путь
instanсes_paths = {name: instance_path(name) for name in instances_names}

# Словарь соответствия путь -> имя
paths_instanсes = {instance_path(name): name for name in instances_names}


def load_instances() -> dict:
    '''Получение информации о статусе экземпляров 
    терминала MetaTrader 5 из заданных папок.

    Returns:
        instances (dict): Информация об экземплярах терминала.
    '''

    # Формируем словарь для всех экземпляров по списку имён
    instances = {name: {'pid': 0, 'status': 'stopped'}
                 for name in instances_names}

    # Перебираем все запущенные процессы
    for proc in psutil.process_iter(['pid', 'name', 'exe']):
        try:
            # Получаем путь к запускаемому файлу процесса
            exe_path = str(proc.info['exe']).replace('\\', '/')

            # Если он есть среди путей экземпляров терминала
            if exe_path and exe_path in paths_instanсes:
                # Определяем имя экземпляра для данного пути
                name = paths_instanсes[exe_path]

                # Берём нужный экземпляр
                instance = instances[name]

                # Запоминаем идентификатор и статус процесса
                instance['pid'] = proc.info['pid']
                instance['status'] = 'running'
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue

    return instances


def start_mt5(name: str) -> dict:
    '''
    Запуск экземпляра терминала

    Args:
        name (str): Имя экземпляра терминала.

    Returns:
        instance (dict): Информация о запущенном терминале.
    '''

    # Получаем информацию об экземплярах терминала
    instances = load_instances()

    # Если имя экземпляра есть среди имеющихся
    if name in instances:
        # Берём нужный экземпляр
        instance = instances[name]

        # Если для него нет запущенного процесса, то
        if not instance['pid']:
            # Запускаем новый процесс
            process = subprocess.Popen([instanсes_paths[name], '/portable'])

            # Идентификатор запущенного процесса
            pid = process.pid

            # Запоминаем идентификатор и статус процесса
            instance['pid'] = pid
            instance['status'] = 'running'

        # Возвращаем результат - успешный запуск
        return {name: instance}

    # Возвращаем результат - имя не найдено
    return {name: {'status': 'not found'}}


def stop_mt5(name: str) -> dict:
    '''
    Остановка терминала

    Args:
        name (str): Имя экземпляра терминала.

    Returns:
        instance (dict): Информация о запущенном терминале.
    '''

    # Получаем информацию об экземплярах терминала
    instances = load_instances()

    # Если имя экземпляра есть среди имеющихся
    if name in instances:
        # Берём нужный экземпляр
        instance = instances[name]

        # Берём идентификатор его процесса
        pid = instance['pid']

        # Если ранее терминал был запущен
        if pid:
            try:
                # Получаем объект процесса терминала
                p = psutil.Process(pid)

                # Останавливаем процесс
                p.terminate()
                p.wait(10)
            except psutil.NoSuchProcess:
                pass

        # Удаляем информацию о запущенном ранее процессе
        instance['pid'] = 0
        instance['status'] = 'stopped'

        # Возвращаем результат - успешная остановка
        return {name: instance}

    # Возвращаем результат - процесс не найден
    return {name: {'status': 'not found'}}