157 lines
5.6 KiB
Python
157 lines
5.6 KiB
Python
'''
|
|
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'}}
|