Skip to content

Refactor/component based architecture

Valentina Kustikova requested to merge github/fork/Serebro1/refactor/component-based-architecture into main

Created by: Serebro1

В данном pull request я попытался разделить обязанности визуализатора отображать данные в определённом виде и всей работы нашего приложения. В результате я выделил 2 отдельных класса: Pipeline Components и Detection Pipeline

Основные изменения:

Pipeline Components

Данный класс является некоторым хранилищем всех используемых компонентов, за исключением AccuracyCalculator, так как данный модуль используется отдельно для вычисления метрик с использованием готового файла с детекциями и готовой разметкой. С помощью этого класса можно сформировать разные конфигурации для запуска детектора, например, с использованием UI библиотеки OPENCV или консольного вывода. Код:

@dataclass
class PipelineComponents:
    """Container for all pipeline components"""
    reader: FrameDataReader
    detector: Detector
    visualizer: BaseVisualizer
    writer: Writer = None
    gt_reader: dr.DataReader = None

Detection Pipeline

Данный класс является ядром программы. В него передаётся объект класса PipelineComponents, заранее сформированная конфигурация работы нашего детектора. Код:

class DetectionPipeline:
    def __init__(self, components: PipelineComponents):
        """Если в компонентах не передана визуальная часть,
           детектор или датаридер видео/изображений, то выкинется исключение"""
        self.components = components
        self.gtboxes = None
    def run(self):
        try:
            with self.components.reader as reader:
                self.components.visualizer.initialize(reader.get_total_images())
                if self.components.gt_reader:
                    self.gtboxes = self.components.gt_reader.read()

                for frame_idx, frame in enumerate(reader):
                    self._process_frame(frame_idx, frame)
                    self.components.visualizer.update_progress()
                    if self._should_exit():
                        break

        except Exception as e:
            self._handle_error(e)
        finally:
            self._finalize()

Главный метод данного класса run() выполняет следующее:

  1. Использование датаридера видео и изображения реализуется контекстным менеджером, который в конструкции with проинициализирует ресурсы и в конце освободит их (важно для датаридера видео).
  2. Инициализация визуальной части, передаётся число кадров, для отслеживания процесса работы.
  3. Если есть разметка, то считываем её и сохраняем в классе для дальнейшего использования
  4. Запускается цикл прохода по кадрам. 4.1. Обработка изображения. Детектор выдаёт набор срабатываний, который при наличии компонента write запишет результат в файл и происходит визуализация изображения, в зависимости от того, какой визуализатор используем. 4.2. В визуализаторе обновляются данные о прогрессе и выводит статус выполнения (прогресс бар или обычный вывод в консоль) 4.3. Если была нажата клавиша прерывания, то цикл прервётся.
  5. Если произошло исключение, происходит очищение файла с набором записанных в процессе детекций и исключение перекидывается дальше.
  6. В конце работы цикла или при исключении происходит освобождение всех ресурсов и подтверждение об окончании цикла.

Main()

Таким образом примерно следующий main, с использованием новых классов, можно просто импортировать различные конфигурации и использовать в одном main. Полный код представлен в файле new_main.py

def cli_argument_parser():
    parser = argparse.ArgumentParser()
    """аргументы, ожидаемые в консоли"""
    args = parser.parse_args()
    return args

def config_visual_main(args: argparse.Namespace):
    return PipelineComponents(
            reader = FrameDataReader.create(args.mode, (args.video_path or args.images_path)),
            detector = Detector.create( "fake" ),
            visualizer = vis.GUIVisualizer(),
            writer = Writer.create(args.write_path) if args.write_path else None,
            gt_reader = dr.CsvGTReader(args.groundtruth_path) if args.groundtruth_path else None)

def config_cli_main(args: argparse.Namespace):
    return PipelineComponents(
            reader = FrameDataReader.create(args.mode, (args.video_path or args.images_path)),
            detector = Detector.create( "fake" ),
            visualizer = vis.CLIVisualizer(),
            writer = Writer.create(args.write_path) if args.write_path else None,
            gt_reader = dr.CsvGTReader(args.groundtruth_path) if args.groundtruth_path else None)

def main():
    try:
        args = cli_argument_parser()

        components = config_cli_main(args)

        pipeline = DetectionPipeline(components)
        pipeline.run()

        if args.groundtruth_path and args.write_path is not None:
            accur_calc = AccuracyCalculator()
            accur_calc.load_detections(args.write_path)
            accur_calc.load_groundtruths(args.groundtruth_path)
            print (f"TPR: {accur_calc.calc_tpr()}\n"
                f"FDR: {accur_calc.calc_fdr()}\n"
                f"MAP: {accur_calc.calc_map()}")

    except Exception as e:
        print(e)


if __name__ == '__main__':
    main()

В данном main мы вводим аргументы, передаём конфигуратору, который создаст компоненты и вернёт их. Далее передаём их конвейеру и запускаем. Если мы передали в аргументы путь к разметке и путь к файлу с детекциями, то считаем и выводим метрики.

Остальные изменения:

Класс FrameDataReader Добавлены функции enter и exit для использования контекстного менеджера и get_total_frames для получения числа кадров видео и изображений в папке.

Модуль Visualizer Создан абстрактный класс BaseVisualizer и наследники GUIVisualizer и CLIVisualizer

class BaseVisualizer(ABC):
   @abstractmethod
   def initialize(self, total_frames: int):
       """Инициализация и вывод статуса начала работы """

   @abstractmethod
   def update_progress(self):
       """Обновление данных и вывод статуса о текущем процессе выполнения """

   @abstractmethod
   def visualize_frame(self, frame: numpy.ndarray,
                    detections: list, ground_truth: list = None):
       """Визуализация данных"""

   @abstractmethod
   def check_exit(self):
       """Проверка была ли нажата кнопка для прерывания основного цикла"""

   @abstractmethod
   def finalize(self):
       """Очищение ресурсов и статус окончания работы"""

Merge request reports