Changed the distribution of functions by classes

card0_manager.py

@@ -0,0 +1,109 @@
+import json
+import os
+from pathlib import Path
+
+from logging_settings import set_logger
+
+
+class CARD0:
+
+    @staticmethod
+    def check_cable_status(fp):
+        if Path(fp).is_file():
+            with open(fp, 'r') as file:
+                status = file.read().replace('\n', '')
+                return status
+
+        else:
+            return 'disconnected'
+
+    @staticmethod
+    def get_cables_path_condition_from_card0():
+        dirs = os.listdir(os.environ['CABLES_DIR'])
+        cables_dirs = list(filter(lambda path: 'HDMI' in path or 'DP' in path, dirs))
+        cables_path = [f'{os.environ["CABLES_DIR"]}{dir}' for dir in cables_dirs]
+        cables_path_condition = {}
+
+        for cable_path in cables_path:
+            with open(f'{cable_path}/status', 'r', encoding='utf-8') as file:
+                cables_path_condition[cable_path] = file.read().replace('\n', '')
+
+        return cables_path_condition
+
+    @staticmethod
+    def check_if_cable_was_disconnected(past_cables_conditions, cables_conditions_file_path):
+        removed_cables = []
+
+        # проходимся по списку из старых кабелей, отдельно получаем текущую информацию для кабеля, сравниваем
+        for port_path, port_past_condition in past_cables_conditions.copy().items():
+            current_cable_status = CARD0.check_cable_status(f'{port_path}/status').replace('\n', '')
+
+            if current_cable_status != port_past_condition:
+                removed_cables.append(port_path.split("/")[-1])
+                past_cables_conditions[port_path] = current_cable_status
+
+                with open(cables_conditions_file_path, 'w') as file:
+                    json.dump(past_cables_conditions, file, indent=4)
+
+            # TODO: does this continue still needed?
+            continue
+
+        return removed_cables
+
+    @staticmethod
+    def check_if_cable_was_connected(past_cables_conditions, current_cables_conditions_from_card0):
+        new_cables = []
+
+        # проходимся по каждой кабелю из текущего состояния, смотрим, есть ли он в предыдущем состоянии
+        for port_path, current_condition in current_cables_conditions_from_card0.items():
+
+            # Проверяем, не появился ли новый кабель в card0
+            if not past_cables_conditions.get(port_path):
+                new_cables.append(port_path.split("/")[-1])
+
+            continue
+
+        return new_cables
+
+    @staticmethod
+    def check_cable_conditions_until_it_change(cabels_conditions_file_path):
+        p = Path('.')
+        PATH_TO_CURRENT_DIRECTORY = p.absolute()
+
+        while True:
+            path_for_check = Path(f'{PATH_TO_CURRENT_DIRECTORY}/{cabels_conditions_file_path}')
+            if not path_for_check.is_file() or path_for_check.stat().st_size == 0:
+                logger.debug(f'Файл конфигов был пуст или не существовал')
+
+                with open(cabels_conditions_file_path, 'w') as file:
+                    json.dump(CARD0.get_cables_path_condition_from_card0(), file, indent=4)
+                return True
+
+            else:
+                with open(cabels_conditions_file_path, 'r') as file:
+                    past_cables_conditions = json.load(file)
+
+                current_cables_conditions_from_card0 = CARD0.get_cables_path_condition_from_card0()
+
+                cables_changes = {
+                    'new_cables': CARD0.check_if_cable_was_connected(past_cables_conditions,
+                                                                     current_cables_conditions_from_card0),
+                    'removed_cables': CARD0.check_if_cable_was_disconnected(past_cables_conditions,
+                                                                            cabels_conditions_file_path)
+                }
+
+                with open(cabels_conditions_file_path, 'w') as file:
+                    json.dump(current_cables_conditions_from_card0, file, indent=4)
+
+                if cables_changes['new_cables']:
+                    logger.debug(f'Обнаружены новыe подключения: {cables_changes["new_cables"]}')
+                    return True
+
+                if cables_changes['removed_cables']:
+                    logger.debug(f'Следующие подключения были удалены: {cables_changes["removed_cables"]}')
+                    return True
+
+                logger.debug('Подключения портов не изменились')
+
+
+logger = set_logger()

monitors_manager.py

@@ -0,0 +1,126 @@
+import os
+import time
+from collections import OrderedDict
+
+from logging_settings import set_logger
+from xrandr_manager import XRANDR
+
+
+class MONITORS:
+
+    @staticmethod
+    def set_monitors_position_manually():
+        monitors_data_from_xrandr = XRANDR.get_monitors_data_from_xrandr()
+
+        cables_ids_names = XRANDR.create_cable_id_cable_name_dict([*monitors_data_from_xrandr])
+
+        logger.info(
+            f"\n\nВот список подключенных мониторов в формате 'monitor_id --- monitor_name':\n"
+            f"{[f'{cable_id} --- {cable_name}' for cable_id, cable_name in cables_ids_names.items()]}\n"
+            f'\n'
+            f'Ниже через запятую введите список id мониторов в последовательности, которую вы хотели вы наблюдать:\n')
+        # TODO: проверить инпут на верность (мб там лишние значения есть например)
+        order_of_monitor_ids_from_input = [int(count) for count in input().strip().split(',')]
+
+        ordered_monitors = []
+
+        for monitor_id in order_of_monitor_ids_from_input:
+            cable_name = cables_ids_names[monitor_id]
+            ordered_monitors.append(cable_name)
+
+        return ordered_monitors
+
+    @staticmethod
+    def match_monitor_with_cable():
+        monitors_data_from_xrandr = XRANDR.get_monitors_data_from_xrandr()
+
+        # TODO: вот это у меня во многих местах встречается, нужно в отдельную функцию вынести
+
+        cables_ids_names = XRANDR.create_cable_id_cable_name_dict(monitors_data_from_xrandr.keys())
+
+        # TODO: вот это в отдельную функцию вынести, ибо используется в разных местах
+        logger.info(
+            f"\n\nВот список портов, к которым подключены мониторы в формате 'port_id --- port_name':\n"
+            f"{[f'{cable_id} --- {cable_name}' for cable_id, cable_name in cables_ids_names.items()]}\n"
+            f'\n'
+            f'Ниже введите id порта, чтобы узнать, какой монитор к нему подключен\n')
+        # TODO: проверить инпут на верность (мб там лишние значения есть например)
+        cable_id_from_input = int(input().strip())
+
+        logger.info(
+            'Сейчас на 3 секунды монитор, подключенный к выбранному вами кабелю поменяет яркость. Не прервайте работу скрипта!')
+
+        string_for_execute = f'xrandr --output {cables_ids_names[cable_id_from_input]} --brightness 0.5'
+        os.system(string_for_execute)
+
+        time.sleep(3)
+
+        string_for_execute = f'xrandr --output {cables_ids_names[cable_id_from_input]} --brightness 1'
+        os.system(string_for_execute)
+
+        logger.debug('Работа скрипта закончена')
+
+    @staticmethod
+    def delete_saved_config(saved_cables_positions):
+
+        saved_cable_position_with_id = XRANDR.create_layout_id_layout_name_dict(saved_cables_positions)
+
+        logger.info('Next strings will show monitors_positions in format: {config: [cable_1, cable_2, etc]}\n')
+        for id, monitors_names in saved_cable_position_with_id.items():
+            print(f'\n{id}: {monitors_names}')
+
+        logger.info('Write separate by commas ids of configs and script will delete them from saved configs')
+
+        # filter is needed in case of string: "1, 2, "
+        input_ids = [int(id) for id in list(filter(None, input().strip().split(',')))]
+        configs_for_delete = [saved_cable_position_with_id[id] for id in input_ids]
+
+        new_configs = []
+
+        for saved_config in saved_cables_positions:
+            if saved_config not in configs_for_delete:
+                new_configs.append(saved_config)
+
+        return new_configs
+
+    @staticmethod
+    def connect_monitors_automatically(data_from_xrandr):
+
+        if len(data_from_xrandr) > 1:
+            monitors_sorted_by_size = list(OrderedDict(sorted(list(data_from_xrandr.items()),
+                                                              key=lambda value: XRANDR.get_monitor_dimensions(
+                                                                  value[1]['monitor_size']))).keys())
+            # TODO: нужно описать кейсы, когда в этой функции вообще есть смысл
+            # например, если скрипт начал работу на заднем плане. Чтоб хоть как-то моники подрубились... Удобнее работать?
+
+            return monitors_sorted_by_size
+
+        else:
+            os.system('xrandr --auto')
+            return data_from_xrandr
+
+    @staticmethod
+    def print_connected_cables(cables_conditions):
+        connected_cables = []
+
+        for cable_path, condition in cables_conditions.items():
+            if condition == 'connected':
+                connected_cables.append(cable_path.split('/')[-1])
+
+        logger.debug(f'Подключены следующие кабели: {connected_cables}')
+
+    @staticmethod
+    def create_new_collections_of_mon_positions(new_mon_pos, previous_monitor_positions):
+        for id, monitors_names_from_settings in enumerate(previous_monitor_positions):
+
+            if monitors_names_from_settings == new_mon_pos:
+                previous_monitor_positions.insert(0, previous_monitor_positions.pop(id))
+                new_monitors_positions = previous_monitor_positions
+                return new_monitors_positions
+
+        previous_monitor_positions.insert(0, new_mon_pos)
+        new_monitors_positions = previous_monitor_positions
+        return new_monitors_positions
+
+
+logger = set_logger()

xrandr_manager.py

@@ -0,0 +1,87 @@
+import re
+import subprocess
+
+
+class XRANDR:
+    @staticmethod
+    def get_monitors_data_from_xrandr():
+        monitors = subprocess.check_output('xrandr').decode()
+
+        iterator = re.finditer(r"^\S[\W\w]*?(?=^(\S|$))", monitors, re.MULTILINE)
+        monitors = [match.group() for match in iterator]
+        connected_monitors = list(filter(lambda monitor_data: ' connected' in monitor_data, monitors))
+
+        monitors_data = list(zip([monitor.split()[0] for monitor in connected_monitors], connected_monitors))
+        physically_connected_monitors_data = {elem[0]: elem[1] for elem in monitors_data}
+
+        data_from_xrandr_about_physically_connected_monitors = {}
+
+        for port, data in physically_connected_monitors_data.items():
+            resolutions = list(filter(lambda word: 'x' in word, ''.join(data.splitlines()[1:-1]).split()))
+
+            monitor_size = data.splitlines()[0].split('y axis) ')[1]
+            monitor_size = [int(option.replace('mm', '')) for option in monitor_size.split('x')]
+
+            data_from_xrandr_about_physically_connected_monitors[port] = {'resolutions': resolutions,
+                                                                          'monitor_size': monitor_size}
+
+        return data_from_xrandr_about_physically_connected_monitors
+
+    @staticmethod
+    def create_cable_id_cable_name_dict(cables_names: list[str]):
+        # TODO: XRANDR_MANAGER. Почему? Это разве не действия с мониторами?
+        cables_ids_names = {id: cable for id, cable in enumerate(list(cables_names))}
+        return cables_ids_names
+
+    @staticmethod
+    def create_layout_id_layout_name_dict(cables_layout: list[list[str]]):
+        cables_ids_names = {id: [*monitors_position] for id, monitors_position in
+                            enumerate(cables_layout)}
+
+        return cables_ids_names
+
+    @staticmethod
+    def get_monitor_dimensions(monitor_dimensions):
+        return monitor_dimensions[0] * monitor_dimensions[1]
+
+    @staticmethod
+    def create_string_for_execute(monitors_data: list[str]):
+        # TODO: XRANDR_MANAGER (приватная функция)
+        execute_command = []
+
+        # TODO: сюда еще дописать логгирование: в xrandr нет указанного вами монитора: monitor_name
+        xrandr_data = XRANDR.get_monitors_data_from_xrandr()
+
+        for id, cable_name in enumerate(monitors_data):
+            if len(monitors_data) == id + 1:
+                break
+
+            """ 
+            sometimes, on some devices, xrandr works better, if u will enter commands sequentially.
+            so, its better to write command, each parts of which divided be |
+            also, we creating command with top quality of each monitor
+            """
+
+            # TODO: первый аргумент в пользу данных из xrandr в формате словаря: можно будет спокойно получать информацию о мониторе по названию подключенного к нему кабеля. При использовании списка приходится юзать filter
+
+            execute_command.append(
+                f'xrandr --output {cable_name} --pos {xrandr_data[cable_name]["resolutions"][0]} --left-of {monitors_data[id + 1]} --pos {xrandr_data[monitors_data[id + 1]]["resolutions"][0]} | ')
+
+        execute_command = ' '.join(execute_command)
+
+        # cleaning spaces
+        execute_command = execute_command[0: -3]
+
+        return execute_command
+
+    @staticmethod
+    def get_needed_monitors_layout_config_automatically(data_from_xrandr, saved_configs):
+        # TODO: вот здесь нужно посмотреть, что происходт
+        # TODO: не очень понятно, к какому классу это стоит отнести
+        current_connection = set([elem for elem in [*data_from_xrandr]])
+
+        for saved_config in saved_configs:
+            if current_connection == set(saved_config):
+                return saved_config
+
+        return None