---

constants/architecture_constants.py
    CONFIG_PATH: str = f"/data/{STAND_ENV_NAMING}/configs"
    SIGNAL_UNIT_CONVERSION_RULES_FILE_NAME: str = "signal_unit_conversion_rules.json"
    SIGNAL_UNIT_CONVERSION_RULES_BACKUP_DIR: str = "original_conversion_rules"


constants/enums.py

внизу 

class MeasureConversionRule(Enum):
    MPA_MEASURE = "MPA_MEASURE"
    KG_CM_MEASURE = "KG_CM_MEASURE"
    




constants/test_constants.py

внизу
class MeasureUnitConstants:
    MPA_MEASURE: str = "MPa"
    KG_CM_MEASURE: str = "kgf/cm^2"





infra/cmd_generator.py
class SignalUnitConversionCmdGenerator(BaseCmdGenerator):
    def generate_scp_signal_rules_from_stand_cmd(self) -> str:
        """
        Генерирует команду копирования signal_unit_conversion_rules.json со стенда на runner
        """
        remote_path = self._generate_path_to_remote_signal_rules()
        return f"{self._scp_cmd} {self._username}@{self._host}:{remote_path} ."

    def generate_scp_signal_rules_to_stand_cmd(self, local_file_path: str) -> str:
        """
        Генерирует команду копирования локального файла на стенд 
        """
        remote_path = self._generate_path_to_remote_signal_rules()
        return f"{self._scp_cmd} {local_file_path} {self._username}@{self._host}:{remote_path}"

    @staticmethod
    def _generate_path_to_remote_signal_rules() -> PurePosixPath:
        return PurePosixPath(Im_const.CONFIG_PATH) / Im_const.SIGNAL_UNIT_CONVERSION_RULES_FILE_NAME





infra/signal_unit_conversion_manager.py
import json
import logging
import shutil
from pathlib import Path
from typing import Any

from clients.subprocess_client import SubprocessClient
from constants.architecture_constants import ClickhouseConstants as CH_const
from constants.architecture_constants import ImitatorConstants as Im_const
from constants.enums import MeasureConversionRule
from infra.cmd_generator import SignalUnitConversionCmdGenerator
from utils.helpers.signal_unit_conversion_utils import (
    apply_measure_conversion_rule,
    conversion_rules_need_update,
)

logger = logging.getLogger(__name__)


class SignalUnitConversionManager:
    """
    Управляет signal_unit_conversion_rules.json на стенде:
    - скачивает оригинал в original_conversion_rules/ перед прогоном набора
    - подкладывает изменённую версию в CONFIG_PATH (имя на сервере не меняется)
    - восстанавливает оригинал в teardown
    """

    def __init__(
        self,
        stand_client: SubprocessClient,
        measure_conversion_rule: MeasureConversionRule | None,
    ) -> None:
        self._stand_client = stand_client
        self._measure_conversion_rule = measure_conversion_rule
        self._cmd_generator = SignalUnitConversionCmdGenerator(stand_client.username, stand_client.host)
        self._local_file = Path(Im_const.SIGNAL_UNIT_CONVERSION_RULES_FILE_NAME)
        self._backup_file = (
            Path(Im_const.SIGNAL_UNIT_CONVERSION_RULES_BACKUP_DIR) / Im_const.SIGNAL_UNIT_CONVERSION_RULES_FILE_NAME
        )
        self._modified = False

    def setup_signal_unit_conversion_rules(self) -> None:
        """
        Скачивает файл со стенда, при необходимости меняет единицы и загружает обратно.
        """
        if self._measure_conversion_rule is None:
            logger.info("[SETUP] [SKIP] measure_conversion_rules не задан для набора данных")
            return

        try:
            self._download_from_stand()
            rules_json = self._read_local_file()

            if not conversion_rules_need_update(rules_json, self._measure_conversion_rule):
                logger.info(
                    "[SETUP] [OK] signal_unit_conversion_rules.json уже настроен корректно для набора данных "
                    f"(правило {self._measure_conversion_rule.name})"
                )
                return

            self._save_backup()
            modified_rules = apply_measure_conversion_rule(rules_json, self._measure_conversion_rule)
            self._write_local_file(modified_rules)
            self._upload_to_stand(self._local_file)
            self._modified = True
            logger.info(
                "[SETUP] [OK] signal_unit_conversion_rules.json обновлён по правилу "
                f"{self._measure_conversion_rule.name}"
            )
        except Exception as error:
            error_msg = "[SETUP] [ERROR] Ошибка при подготовке signal_unit_conversion_rules.json"
            logger.exception(error_msg)
            raise RuntimeError(error_msg) from error

    def restore_signal_unit_conversion_rules(self) -> None:
        """
        Возвращает оригинальный signal_unit_conversion_rules.json на стенд
        """
        if not self._modified:
            logger.info("[TEARDOWN] [SKIP] signal_unit_conversion_rules.json не изменялся")
            return

        if not self._backup_file.exists():
            error_msg = (
                "[TEARDOWN] [ERROR] Оригинал signal_unit_conversion_rules.json не найден: "
                f"{self._backup_file}"
            )
            logger.error(error_msg)
            raise RuntimeError(error_msg)

        try:
            self._upload_to_stand(self._backup_file)
            self._modified = False
            logger.info(
                "[TEARDOWN] [OK] signal_unit_conversion_rules.json восстановлен на стенде "
                f"из {self._backup_file}"
            )
        except Exception as error:
            error_msg = "[TEARDOWN] [ERROR] Ошибка при восстановлении signal_unit_conversion_rules.json"
            logger.exception(error_msg)
            raise RuntimeError(error_msg) from error

    def _download_from_stand(self) -> None:
        copy_cmd = self._cmd_generator.generate_scp_signal_rules_from_stand_cmd()
        self._stand_client.run_cmd(copy_cmd, timeout=CH_const.LONG_PROCESS_TIMEOUT_S, use_ssh=False)
        if not self._local_file.exists():
            raise FileNotFoundError(
                f"Не удалось скачать {Im_const.SIGNAL_UNIT_CONVERSION_RULES_FILE_NAME} со стенда"
            )

    def _save_backup(self) -> None:
        self._backup_file.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(self._local_file, self._backup_file)

    def _upload_to_stand(self, local_file: Path) -> None:
        upload_cmd = self._cmd_generator.generate_scp_signal_rules_to_stand_cmd(local_file.as_posix())
        self._stand_client.run_cmd(upload_cmd, timeout=CH_const.LONG_PROCESS_TIMEOUT_S, use_ssh=False)

    def _read_local_file(self) -> dict[str, Any]:
        error_msg = (
            f"[SETUP] [ERROR] Не удалось декодировать файл {self._local_file} "
            f"в кодировках {Im_const.DEFAULT_ENCODINGS}"
        )
        for encoding in Im_const.DEFAULT_ENCODINGS:
            try:
                with open(self._local_file, "r", encoding=encoding, errors="strict") as rules_file:
                    data = json.load(rules_file)
                    if not data:
                        raise ValueError(f"Пустой json (кодировка:{encoding})")
                    return data
            except UnicodeDecodeError:
                continue
            except json.JSONDecodeError as error:
                logger.exception(error_msg)
                raise OSError(error_msg) from error
        logger.exception(error_msg)
        raise OSError(error_msg)

    @staticmethod
    def _write_local_file(rules_json: dict[str, Any]) -> None:
        with open(
            Im_const.SIGNAL_UNIT_CONVERSION_RULES_FILE_NAME,
            "w",
            encoding=Im_const.ENCODING_UTF_8,
        ) as rules_file:
            json.dump(rules_json, rules_file, ensure_ascii=False, indent=2)



















infra/stand_setup_manager.py


from constants.enums import MeasureConversionRule, TU




from infra.signal_unit_conversion_manager import SignalUnitConversionManager



        test_data_name: str,  # Название архива данных имитатора для загрузки из TestOps
        tu_id: int,
        measure_conversion_rules: MeasureConversionRule | None = None,




        self._test_data_name = test_data_name
        self._tu_id = tu_id
        self._measure_conversion_rules = measure_conversion_rules




                self._uploader.upload_with_confirm()

            self.stop_all_containers()
            self._signal_unit_conversion_manager.setup_signal_unit_conversion_rules()







            logger.exception(error_msg)
            raise RuntimeError(error_msg) from error

    def restore_signal_unit_conversion_rules(self) -> None:
        """
        Возвращает оригинальный signal_unit_conversion_rules.json на стенд.
        """
        self._signal_unit_conversion_manager.restore_signal_unit_conversion_rules()





            self._configuration_manager = ConfigurationManager(self._configuration_file_name)
            self._signal_unit_conversion_manager = SignalUnitConversionManager(
                self._stand_client, self._measure_conversion_rules
            )






































test_config/models_for_tests.py

    LdsStatus,
    MeasureConversionRule,




    # ===== Технологический участок =====
    technological_unit: TU = TU.TIKHORETSK_NOVOROSSIYSK_3

    # ===== Правила конвертации единиц измерения давления на стенде =====
    measure_conversion_rules: Optional[MeasureConversionRule] = None
















utils/helpers/signal_unit_conversion_utils.py

import copy
from typing import Any

from constants.enums import MeasureConversionRule
from constants.test_constants import MeasureUnitConstants


def _resolve_unit_mapping(rule: MeasureConversionRule) -> tuple[str, str]:
    if rule == MeasureConversionRule.MPA_MEASURE:
        return MeasureUnitConstants.KG_CM_MEASURE, MeasureUnitConstants.MPA_MEASURE
    if rule == MeasureConversionRule.KG_CM_MEASURE:
        return MeasureUnitConstants.MPA_MEASURE, MeasureUnitConstants.KG_CM_MEASURE
    raise ValueError(f"Неизвестное правило конвертации единиц измерения: {rule}")


def conversion_rules_need_update(rules_json: dict[str, Any], rule: MeasureConversionRule) -> bool:
    """
    Проверяет, есть ли в файле единицы давления, которые нужно заменить по правилу набора.
    """
    source_unit, _ = _resolve_unit_mapping(rule)
    return any(signal.get("OriginUnit") == source_unit for signal in rules_json.get("Signals", []))


def apply_measure_conversion_rule(rules_json: dict[str, Any], rule: MeasureConversionRule) -> tuple[dict[str, Any], int]:
    """
    Возвращает копию rules_json с заменой единиц давления и число заменённых сигналов.

    Замена строго по полному совпадению OriginUnit с исходной единицей давления
    (kgf/cm^2 <-> MPa). Остальные единицы (cSt, m^3/h, rpm и т.д.) не затрагиваются.
    """
    source_unit, target_unit = _resolve_unit_mapping(rule)
    result = copy.deepcopy(rules_json)

    for signal in result.get("Signals", []):
        origin_unit = signal.get("OriginUnit")
        if origin_unit == source_unit:
            signal["OriginUnit"] = target_unit
            
    return result























conftest.py


from test_config.datasets import ALL_SMOKE_CONFIGS, get_config_by_name




def _find_config_by_suite_name(suite_name: str):
    """Находит конфиг по имени набора данных."""
    try:
        return get_config_by_name(suite_name)
    except ValueError:
        return None





        if stand_manager := cfg["stand_manager"]:
            stand_manager.stop_imitator_wrapper()
            try:
                stand_manager.restore_signal_unit_conversion_rules()
            except Exception:
                logger.exception(
                    "[ERROR] [SETUP] Ошибка при восстановлении signal_unit_conversion_rules.json "
                    "перед запуском нового набора"
                )







        imitator_duration = compute_imitator_duration(item, current_test_suite)

        suite_config = _find_config_by_suite_name(current_test_suite)
        measure_conversion_rules = (
            suite_config.measure_conversion_rules if suite_config is not None else None
        )







        stand_manager = StandSetupManager(
            duration_m=imitator_duration,
            test_data_id=data_id,
            test_data_name=test_data_name,
            tu_id=tu_id,
            measure_conversion_rules=measure_conversion_rules,
        )






        if stand_manager := cfg["stand_manager"]:
            stand_manager.stop_imitator_wrapper()
            try:
                stand_manager.restore_signal_unit_conversion_rules()
            except Exception:
                logger.exception(
                    "[ERROR] [TEARDOWN] Ошибка при восстановлении signal_unit_conversion_rules.json"
                )






                stand_manager.stop_imitator_wrapper()
            except Exception:
                logger.exception("[ERROR] [TEARDOWN] Ошибка при остановке имитатора")
            try:
                stand_manager.restore_signal_unit_conversion_rules()
            except Exception:
                logger.exception(
                    "[ERROR] [TEARDOWN] Ошибка при восстановлении signal_unit_conversion_rules.json"
                )