core/data_logger.py - master - DimerTrue/ecg_gui

core/data_logger.py
155 строк5 KB
DimerTrue
ui bug fixed
20 июл 2026, 17:04
20 июл 2026, 17:045fabad5
# data_logger.py
from core.app_logger import AppLogger

logger = AppLogger.get_logger(__name__)

import threading
import queue
from dataclasses import dataclass
from typing import Any, Dict, List
import os
from datetime import datetime
import time

MAIN_FOLDER_PATH = "results"
FORMAT = ".txt"

HEADERS = [
    "Дата",
    "Время",
    "Ток_ЭХГ",
    "Напряжение_ЭХГ",
    "Ток_АКБ",
    "Напряжение_АКБ",
    "Ток_потребителя",
    "Напряжение_потребителя",
    "Статус_чтение",
    "Статус_запись",
    "Серийный_номер_прошивки",
    "Темп_платы",
    "Темп_АКБ",
    "Темп_трубок",
    "Давление_топлива",
    "Расход_топлива",
    "Минимальное_напряжение_выхода_з",
    "Максимальное_напряжение_выхода_з",
    "Номинальный_ток_зарядки_АКБ_з",
    "Номинальный_ток_зарядки_АКБ_ч",
    "Минимальное_напряжение_выхода_ч",
    "Максимальное_напряжение_выхода_ч",
    "Ошибки",
    "Серийый_номер_генератора",
    "ID_485",
    "Скорость_обмена",
    "Темп_корп",
    "Давл_возд",
    "Влаж_возд",
]


class DataLogger(threading.Thread):
    """
    Поток для асинхронной записи данных в файлы.
    """

    def __init__(
        self,
        device_ids: List[int],
        devices_types: List[bool],
        data_queues: List[queue.Queue],
        stop_event: threading.Event,
    ):
        super().__init__(daemon=True)

        self.device_ids = device_ids
        self.devices_types = devices_types
        self.data_queues = data_queues
        self.stop_event = stop_event

        self.folder_path = MAIN_FOLDER_PATH
        self.full_dir_path: str
        self.filenames: List[str] = []
        self.file_handles: List[Any] = []

        self._init_date = datetime.now()

        self.create_folder()
        self.create_data_files()

    def create_folder(self):
        """Создаем директорию для результатов измерений"""
        # os.makedirs(self.folder_path, exist_ok=True)
        now = self._init_date

        # Создаем всю цепочку папок
        result_dir_name = f"{now.strftime('%d_%m_%Y %H_%M_%S')} result"
        self.full_dir_path = os.path.join(self.folder_path, result_dir_name)
        os.makedirs(self.full_dir_path, exist_ok=True)

        logger.info(f"Создана директория для результатов {self.full_dir_path}")

    def create_data_files(self):
        """Создает данные для сохранения данных"""
        now = self._init_date

        self.file_handles = []

        for i, id in enumerate(self.device_ids):
            if self.devices_types[i]:
                filename = (
                    f"sim - {now.strftime('%d_%m_%Y %H_%M_%S')} ID_{id} log{FORMAT}"
                )
            else:
                filename = f"{now.strftime('%d_%m_%Y %H_%M_%S')} ID_{id} log{FORMAT}"

            result_path = os.path.join(self.full_dir_path, filename)

            header = " ".join(HEADERS)

            file = open(result_path, "w", encoding="utf-8")
            file.write(header + "\n")

            self.file_handles.append(file)
            logger.info(f"Файл создан по пути {result_path}")

    def run(self):
        """Основной метод потока"""
        while not self.stop_event.is_set():
            for idx, data_queue in enumerate(self.data_queues):
                try:
                    if idx >= len(self.file_handles):
                        continue

                    # data = data_queue.get_nowait()
                    data = data_queue.get(timeout=0.1)
                    data_str = " ".join(data)

                    file = self.file_handles[idx]

                    # Проверяем, жив ли файл, перед записью
                    if file is None or file.closed:
                        continue

                    file.write(data_str + "\n")
                    file.flush()

                    # # Дополнительная проверка на случай, если файл закрыт
                    # if file and not file.closed:
                    #     file.write(str(data) + "\n")
                    #     file.flush()
                except queue.Empty:
                    continue
                except Exception as e:
                    logger.exception(f"Ошибка записи в файл {idx}")

            # time.sleep(0.01)

    def stop(self):
        """Остановка логгера и закрытие файлов"""
        if self.file_handles:
            for file in self.file_handles:
                try:
                    if file and not file.closed:
                        file.close()
                except Exception as e:
                    logger.exception(f"Ошибка закрытия файла")