https://pastein.ru/t/tDW
Загрузка данных
from enum import Enum, IntEnum, IntFlag
from typing import Mapping
class BaseStrEnum(Enum):
def __str__(self) -> str:
return f"{self.name} ({self.value})"
class BaseStrIntFlag(IntFlag):
def __str__(self) -> str:
raw_value = int(self)
if raw_value == 0:
return "0"
active_flags = [f"{flag.name} ({flag.value})" for flag in type(self) if flag.value and flag & self == flag]
if active_flags:
return f"{', '.join(active_flags)}"
return str(raw_value)
class BaseReasonEnum(IntFlag):
"""
.report_text - название на русском языке.
Вывод в формате report_text(value)
"""
def __new__(cls, value: int, report_text: str):
member = int.__new__(cls, value)
member._value_ = value
member.report_text = report_text
return member
def __str__(self) -> str:
raw_value = int(self)
if raw_value == 0:
return "0"
active_flags = [
f"{flag.report_text} ({flag.value})" for flag in type(self) if flag.value and flag & self == flag
]
if active_flags:
return f"{', '.join(active_flags)}"
return str(raw_value)
@classmethod
def report_text_by_value(cls, status_value: int) -> str | None:
"""Текст по числовому значению статуса"""
try:
return cls(status_value).report_text
except ValueError:
return None
class TU(Enum):
"""Legacy-идентификация ТУ: имитатор (tn{id}_tags.txt), tags и конфигурация стенда."""
YAROSLAVL_MOSCOW = (1, "Ярославль - Москва", "volga.json")
TIKHORETSK_NOVOROSSIYSK_2 = (2, "Тихорецк-Новороссийск-2", "tn2.json")
TIKHORETSK_NOVOROSSIYSK_3 = (3, "Тихорецк-Новороссийск-3", "tn3.json")
RODIONOVSKAYA_TIKHORETSKAYA = (4, "Родионовская–Тихорецкая", "lt3_rt.json")
TIKHORETSKAYA_GRUSHEVAYA = (5, "Тихорецкая-6-Грушовая", "tn4_t6g.json")
def __init__(self, tu_id: int, description: str, file_name: str) -> None:
self.id = tu_id
self.description = description
self.file_name = file_name
def __str__(self):
return f"{self.id} - {self.description}"
@classmethod
def get_file_name_by_id(cls, target_id: int) -> str:
for item in cls:
if item.id == target_id:
return item.file_name
raise ValueError(f"ТУ с id = {target_id} не найден")
class AdminTU(Enum):
"""ТУ в Администрировании и WS-подписках"""
TIKHORETSK_NOVOROSSIYSK_3_AUTOTEST = (
"ТН-3-Автотест",
TU.TIKHORETSK_NOVOROSSIYSK_3,
)
TIKHORETSK_NOVOROSSIYSK_3_AUTOTEST_REJECT = (
"ТН-3-Автотест-Отбраковки",
TU.TIKHORETSK_NOVOROSSIYSK_3,
)
TIKHORETSK_NOVOROSSIYSK_3_AUTOTEST_DATA_ABSENCE_FALSE = (
"ТН-3-Автотест-DATA-ABSENCE-FALSE",
TU.TIKHORETSK_NOVOROSSIYSK_3,
)
TIKHORETSK_NOVOROSSIYSK_3_AUTOTEST_DATA_ABSENCE_FALSE_BIK = (
"ТН-3-Автотест-DATA-ABSENCE-BIK",
TU.TIKHORETSK_NOVOROSSIYSK_3,
)
TIKHORETSK_NOVOROSSIYSK_3_AUTOTEST_DATA_ABSENCE_FALSE_SELECT = (
"ТН-3-Автотест-DATA-ABSENCE-SELECT",
TU.TIKHORETSK_NOVOROSSIYSK_3,
)
def __init__(self, admin_name: str, legacy_tu: TU) -> None:
self.admin_name = admin_name
self.legacy_tu = legacy_tu
class ReplyStatus(Enum):
OK = 200
BAD_REQUEST = 400
UNAUTHORIZED = 401
FORBIDDEN = 403
NOT_FOUND = 404
REQUEST_TIMEOUT = 408
CONFLICT = 409
PRECONDITION_FAILED = 412
RANGE_NOT_SATISFIABLE = 416
TOO_MANY_REQUESTS = 429
INTERNAL_SERVER_ERROR = 500
NOT_IMPLEMENTED = 501
SERVICE_UNAVAILABLE = 503
GATEWAY_TIMEOUT = 504
UNKNOWN_ERROR = 520
class ExportedDataType(IntEnum):
"""
Тип экспортируемых данных
"""
STATIONARY_STATUS_REPORT = 6
LDS_STATUS_REPORT = 5
LEAKS_REPORT = 4
REJECTED_REPORT = 7
JOURNAL_REPORT = 1
def to_download_name(self) -> str:
"""Строковый тип для DownloadExportedDataRequest.exportedDataType"""
return _EXPORTED_DATA_TYPE_DOWNLOAD_NAMES[self]
_EXPORTED_DATA_TYPE_DOWNLOAD_NAMES = {
ExportedDataType.STATIONARY_STATUS_REPORT: "MnStateReport",
ExportedDataType.LDS_STATUS_REPORT: "LdsStateReport",
ExportedDataType.LEAKS_REPORT: "LeaksReport",
ExportedDataType.REJECTED_REPORT: "RejectedSignalsReport",
}
class ExportStatus(IntEnum):
"""Статус формирования отчёта в ReportDataExportedNotification.replyContent.exportStatus."""
NOT_READY = 0
DONE = 1
class SouAdminStatus(BaseStrEnum):
"""Статус СОУ в разделе Администрирование (GetBasicInfoAdminResponse)."""
STOPPED = (1, 'СОУ выключена')
RUNNING = (2, 'СОУ включена')
def __new__(cls, value: int, report_text: str) -> "SouAdminStatus":
member = object.__new__(cls)
member._value_ = value
member.report_text = report_text
return member
@classmethod
def report_text_by_value(cls, status_value: int) -> str | None:
"""Текст статуса СОУ в Администрировании по числовому значению."""
try:
return cls(status_value).report_text
except ValueError:
return None
class StationaryStatus(BaseStrEnum):
UNSTATIONARY = (1, 'Нестационарный режим работы МТ')
STATIONARY = (2, 'Стационарный режим работы МТ')
STOPPED = (3, 'МТ в режиме остановленной перекачки')
def __new__(cls, value: int, report_text: str) -> "StationaryStatus":
member = object.__new__(cls)
member._value_ = value
member.report_text = report_text
return member
@classmethod
def report_text_by_value(cls, status_value: int) -> str | None:
"""Текст режима СОУ для отчёта по числовому значению статуса"""
try:
return cls(status_value).report_text
except ValueError:
return None
class LeakStatus(BaseStrEnum):
CONFIRMED = 2
WAITING = 1
POSSIBLE = 3
class LeakLocationStatus(BaseStrEnum):
NODATA = 1 # нет данных
LEFT_FROM_PUMP_STATION = 2 # Слева от МНС
RIGHT_FROM_PUMP_STATION = 3 # Справа от МНС
INSIDE_PUMP_STATION = 4 # Внутри МНС
INSIDE_NPS = 5 # Внутри НПС при неработающей/отсутствующей МНС
class FieldName(Enum):
SECTION_TYPE = "sectionType"
SIGNAL_TYPE = "signalType"
class FilterCriteriaType(Enum):
ONE_OF = "oneOf"
ALL_OF = "allOf"
class FilterCriteriaValue(Enum):
MASK = "mask"
MASK_REASON = "maskReason"
LEAK = "leak"
LEAK_COORDINATE = "leakCoordinate"
PUMPING_STATUS = "pumpingStatus"
FREE_FLOW = "freeFlow"
ACKNOWLEDGE = "acknowledge"
LEAK_TIME = "leakTime"
LEAK_VOLUME = "leakVolume"
LDS_STATUS = "ldsStatus"
CONTROLLED_SITES = "controlledSites"
LINEAR_PARTS = "linearParts"
SERVER_DOWN = "serverDown"
TIME_SYNCHRONIZATION_DISABLE = "timeSynchronizationDisable"
FREE_FLOW_START_COORDINATE = "freeFlowStartCoordinate"
class SortingParam(Enum):
OBJECT_NAME = "objectName"
ADDRESS = "address"
class SortingType(Enum):
ASCENDING = "ascending"
DESCENDING = "descending"
class Direction(Enum):
"""Направление прокрутки"""
PREV = 1
NEXT = 2
FIRST = 3
LAST = 4
class LdsStatus(BaseStrEnum):
"""
Режим работы СОУ.
report_text - значение колонки 'Режим работы СОУ' в xlsx-отчёте об утечках.
"""
FAULTY = (1, "СОУ неисправна")
INITIALIZATION = (2, "СОУ в инициализации")
DEGRADATION = (3, "СОУ в ухудшенных характеристиках")
SERVICEABLE = (4, "СОУ исправна")
def __new__(cls, value: int, report_text: str) -> "LdsStatus":
member = object.__new__(cls)
member._value_ = value
member.report_text = report_text
return member
@classmethod
def report_text_by_value(cls, status_value: int) -> str | None:
"""Текст режима СОУ для отчёта по числовому значению статуса"""
try:
return cls(status_value).report_text
except ValueError:
return None
class ConfirmationStatus(BaseStrEnum):
FAULTY = 0 # Неисправность
AWAITING = 1 # Предварительная
NOT_CONFIRMED = 2 # Не подтверждена
CONFIRMED = 3 # Подтверждена
CONFIRMED_AND_LEAK_CLOSED = 4 # Завершена
class ReservedType(BaseStrEnum):
"""Алгоритмы СОУ"""
FAULTY = 0 # Неисправность
STOP = 1 # Дифференциальный
STATIONARY_FLOW = 2 # Стационарный
UNSTATIONARY_FLOW = 3 # Модельный
BALANCE_IN_NPS = 4 # Баланс внутри НПС
CHANGED_IN_DECISION_MAKING = 5 # Стационарный + Изменено в АПР
CREATED_IN_DECISION_MAKING = 6 # Создано в АПР
class MessageType(BaseStrIntFlag):
USER_ACTION = 0
AUTHENTICATION = 1 # Вход в систему
REJECTION = 1 << 2 # Отбраковка сигналов
LDS_STATUS = 1 << 3 # Режим работы СОУ
INPUT_SIGNALS = 1 << 6 # Входные сигналы
PUMPING_STATUS = 1 << 7 # Режим работы МТ
MASKING_LDS = 1 << 8 # Маскирование СОУ
FREE_FLOWS = 1 << 9 # Самотечное течение
LEAKS = 1 << 10 # Утечка
class MessagePriority(BaseStrIntFlag):
LOW = 1 # Прочее
COMMON = 1 << 1 # Информационное
MEDIUM = 1 << 2 # Значительное
HIGH = 1 << 3 # Важное
VERY_HIGH = 1 << 4 # Особой важности
class LdsStatusDegradation(BaseReasonEnum):
"""
Причины режима работы СОУ: Ухудшение характеристик
"""
LEAK_ON_ADJACENT_DIAGNOSTIC_AREAS = (1 << 0, 'Утечка на соседнем диагностическом участке')
ADDITIVE_INJECTORS_OPERATION = (1 << 1, 'Наличие ПТП')
PIG_SENSOR_PASSAGE = (1 << 2, 'Наличие СОД')
TRIGGERING_EMERGENCY_RESET = (1 << 3, 'Срабатывание аварийного сброса или предохранительных клапанов')
STARTING_PUMPING_OUT_PUMPS = (1 << 4, 'Работа насосов откачки')
EXCEEDING_DISTANCE_BETWEEN_SERVICEABLE_PRESSURE_SENSORS = (
1 << 5,
'Расстояние между ближайшими исправными СИ давления на пути перекачки более 50 км',
)
FAULTY_PRESSURE_SENSORS_AT_PUMP_STATION_NODES = (1 << 6, 'Отказ СИ давления на входе/выходе НПС')
REJECTION_TEMPERATURE_SENSOR = (1 << 7, 'Отказ СИ температуры')
REJECTION_VISCOSITY_SENSOR = (1 << 8, 'Отказ СИ вязкости')
REJECTION_DENSITY_SENSOR = (1 << 9, 'Отказ СИ плотности')
GRAVITY_SECTION_IN_PUMPING_MODE = (1 << 10, 'Наличие самотечного участка/участка с неполным сечением')
ABSENCE_MIN_PRESSURE_SENSORS_REQUIRED_NUMBER = (1 << 11, 'Менее 4 исправных СИ давления на разных КП ЛЧ и НПС')
EXCEEDING_DISTANCE_BETWEEN_FLOW_METERS = (
1 << 12,
'Расстояние между ближайшими исправными СИ расхода на пути перекачки более 200 км',
)
GRAVITY_SECTION_IN_STOPPED_PUMPING_MODE = (
1 << 13,
'Наличие самотечного участка/участка с неполным сечением в режиме остановленной перекачки',
)
class LdsStatusFaulty(BaseReasonEnum):
"""
Причины режима работы СОУ: Неисправность
"""
NO_DATA_SOURCE_CONNECTION = (1 << 0, 'Потеря связи СОУ с СДКУ')
ABSENCE_MIN_PRESSURE_SENSORS_REQUIRED_NUMBER = (1 << 1, 'Менее 4 КП с достоверными СИ давления')
ABSENCE_MIN_FLOW_METERS_REQUIRED_NUMBER = (1 << 2, 'Недостоверность граничного СИ расхода')
class LdsStatusInitialization(BaseReasonEnum):
"""
Причины режима работы СОУ: Инициализация
"""
ACCUMULATION_DATA = (1 << 0, 'Накопление данных')
EXITING_FAULTY_MODE = (1 << 1, 'Выход СОУ из режима «Неисправна»')
COLD_START_OF_SERVERS = (1 << 2, 'Одновременный «холодный» запуск нескольких серверов СОУ')
SWITCHING_SHUT_OFF_IN_STOPPED_PUMPING_MODE = (
1 << 3,
'Переключение запорной арматуры в режиме остановленной перекачки',
)
USER_ACTION = (1 << 4, 'По команде пользователя')
class StationaryReason(BaseReasonEnum):
"""
Причины режима работы МТ: Стационар для ЭФ Журнал
"""
PRESSURE_AND_FLOW_MOVING_AVERAGES_MEET_CRITERIA = (
1 << 0,
'Отклонения давления и расхода не превышают допустимых отклонений',
)
ABSENCE_GRAVITY_SECTION_AND_TECHNOLOGICAL_SWITCHING = (
1 << 1,
'Окончание периода времени после технологических переключений и отсутствия самотечного участка',
)
class UnStationaryReason(BaseReasonEnum):
"""
Причины режима работы МТ: Нестационар
"""
CHANGING_EQUIPMENT_STATUS = (
1 << 0,
'Пуск/остановка трубопровода; включение/отключение магистрального насоса; включение/отключение НПС',
)
CHANGING_WORKING_OF_PUMPING_OUT_PUMPS = (
1 << 1,
'Начало/окончание работы насосов откачки емкостей на НПС и ЛЧ технологического участка',
)
CHANGING_MAIN_PUMPS_ROTATION_SPEED = (
1 << 2,
'Изменение частоты вращения в ручном режиме и/или изменение уставки регулирования '
'в автоматическом режиме работы МНА с ЧРП',
)
CHANGING_BLOCK_VALVES_STATUS = (1 << 3, 'Полное или частичное открытие/закрытие задвижки')
SWITCHING_TANKS = (1 << 4, 'Переключение резервуаров')
CHANGING_ACCEPTANCE_OR_DELIVERY_STATE = (1 << 5, 'Начало или прекращение приема/сдачи нефти/нефтепродуктов')
TRIGGERING_EMERGENCY_RESET_OR_PWSS_OPERATION = (1 << 6, 'Задействование аварийного сброса')
SAFETY_VALVES_ACTUATION = (1 << 7, 'Срабатывание предохранительных клапанов')
CHANGING_PRESSURE_SETTING = (
1 << 8,
'Изменение уставки регулирования по давлению узлов регулирования давления, '
'работающих в автоматическом режиме управления',
)
CHANGING_OPENING_PERCENTAGE_VALVE = (
1 << 9,
'Изменение процента открытия/закрытия заслонки узлов регулирования давления, '
'работающих в ручном режиме управления',
)
CHANGING_ADDITIVE_INJECTOR_STATUS_OR_FLOW = (
1 << 10,
'Начало/окончание ввода ПТП или изменение расхода вводимой ПТП',
)
LEAK_END = (1 << 11, 'Окончание утечки')
TO_OPEN_OR_TO_CLOSE_STATUS = (
1 << 12,
'Наличие сигнала статуса «Открывается»/»Закрывается» запорной арматуры (не в режиме имитации), '
'расположенной в точке, гидравлически связанной с рассматриваемым ДУ',
)
ADJACENT_TU = (
1 << 13,
'Нестационарный режим работы/отсутствие сигнала о режиме работы смежного ТУ, '
'работающего в единой гидравлической системе с защищаемым ТУ',
)
COLD_START = (1 << 14, 'Одновременный «холодный» запуск нескольких серверов СОУ')
class StoppedPumpingReason(BaseReasonEnum):
"""
Причины режима работы МТ: Остановленный
"""
STOPPING_PUMPS = (
1 << 0,
'На ДУ отсутствуют работающие НА, при этом показания СИ расхода не превышают 1 % от максимального значения '
'диапазона измерений всех СИ расхода на технологическом участке',
)
CUTOFF_AREA = (1 << 1, 'Участок отсечен запорной арматурой от подкачек/откачек')
class RejectionCriteria(IntFlag):
"""Критерии отбраковки сигналов criteriaNames"""
QUALITY = 1 << 0 # qualityRejection
RANGE = 1 << 1 # rangeRejection
EMPTY = 1 << 2 # emptyRejection
TIME = 1 << 3 # timeRejection
CONSTANT_SIGNAL = 1 << 4 # constantSignalRejection
DISCHARGE = 1 << 5 # dischargeRejection
VTOR = 1 << 6 # VTORRejection
NEARBY = 1 << 7 # nearbyRejection
DIAGNOSTIC_INFO = 1 << 8 # diagnInfoRejection
@property
def backend_name(self) -> str:
names = {
"QUALITY": "qualityRejection",
"RANGE": "rangeRejection",
"EMPTY": "emptyRejection",
"TIME": "timeRejection",
"CONSTANT_SIGNAL": "constantSignalRejection",
"DISCHARGE": "dischargeRejection",
"SIGMA3": "sigma3Rejection",
"VTOR": "VTORRejection",
"NEARBY": "nearbyRejection",
"DIAGNOSTIC_INFO": "diagnInfoRejection",
}
return names.get(self.name or "", str(int(self)))
def __str__(self) -> str:
raw_value = int(self)
if raw_value == 0:
return "0"
active_flags = [flag.backend_name for flag in type(self) if flag.value and flag & self == flag]
if active_flags:
return f"{'|'.join(active_flags)} ({raw_value})"
return str(raw_value)
class RejectionSensorTag(Enum):
"""
Теги датчиков для тестов отбраковки (id, description=tag)
Айди тегов подставляется на ходу из текущей версии конфы с сервера
"""
KP_8_Pin = (0, "AK.CHTN.LU_TIHVEL.KP_8.SW_8-3.Pin") # nearby_pressure_pin range_upper_pressure range_lower_pres
NPS_TIH_5_Vmom = (0, "AK.CHTN.NPS_TIH_5.UZR_1.Vmom") # diagnostic_info_flowrange_upper_flow range_lower_flow
KP_8_Pout = (0, "AK.CHTN.LU_TIHVEL.KP_8.SW_8-3.Pout") # nearby_pressure_pout
KP_209_1_Pin = (0, "AK.CHTN.LU_VELKRIM.KP_209-1.SW_215-3-1.Pin") # empty_pressure
KP_7_Pin = (0, "AK.CHTN.LU_TIHVEL.KP_7.SW_6-3.Pin") # vtor_pressure
NPS_KRIM_P_Vmom = (0, "AK.CHTN.NPS_KRIM_P.UZR_1.Vmom") # empty_flow quality_flow
def __init__(self, sensor_id: int, description: str) -> None:
self.id = sensor_id
self.description = description
@classmethod
def update_ids_from_config(cls, sensor_ids_by_address: Mapping[str, int]) -> None:
"""
Обновляет sensor_id по tag из конфигурации стенда.
"""
missing_tags = []
for sensor in cls:
sensor_id = sensor_ids_by_address.get(sensor.description)
if sensor_id is None:
missing_tags.append(sensor.description)
continue
sensor.id = sensor_id
if missing_tags:
raise ValueError(f"Не найдены sensor_id для tags: {', '.join(missing_tags)}")
def __str__(self):
return f"{self.id} - {self.description}"
class UserActions(IntFlag):
USER_LOGIN = 1 # Вход пользователя
USER_EXIT = 1 << 1 # Выход пользователя
FAILED_USER_LOGIN = 1 << 2 # Неуспешная попытка входа пользователя
ALGORITHMS_REINITIALIZATION = 1 << 3 # Переинициализация алгоритмов
SIGNAL_MASK_SIM = 1 << 4 # Маскирование и имитация входных сигналов
LDS_MASKING = 1 << 5 # Маскирование СОУ
EXPORT = 1 << 6 # Экспорт и выгрузки
SETTINGS_CHANGE = 1 << 7 # Изменение настроек
LEAK_ACK = 1 << 8 # Квитирование сообщения об утечке
LEAK_REMOVE = 1 << 9 # Исключение неактивных утечек
LDS_ADMIN = 1 << 10 # Администрирование СОУ
PIG_CONTROL = 1 << 11 # Управление СОД
class SiteKpKp(Enum):
"""controlledSiteId, segmentId"""
TIXORECZKAYA_NOVOVELICHKOVSKAYA = (6012, 6013)
NOVOVELICHKOVSKAYA_KRYMSKAYA = (6074, 6075)
KRYMSKAYA_GRUSHOVAYA = (6220, 6221)
BACKUP_ROUTE_BEJSUG = (6242, 6243)
BACKUP_ROUTE_PONURA = (6076, 6077)
BACKUP_ROUTE_KUBAN = (6088, 6089)
NPZ_AFIPSKIJ = (6244, 6245)
NPZ_ILINSKIJ = (6120, 6121)
@property
def controlledSiteId(self) -> int:
return self.value[0]
@property
def segmentId(self) -> int:
return self.value[1]
@property
def controlledSiteSegmentDict(self) -> dict:
return {'controlledSiteId': self.controlledSiteId, 'segmentId': self.segmentId}
class SignalType(IntFlag):
"""Типы сигналов: режим МТ - 8, режим СОУ - 256, самотеки -16"""
REGLU = 1 << 3
REGSOU = 1 << 8
GRAVITYPIPE = 1 << 4
@property
def backend_name(self) -> str:
names = {
"REGLU": "PumpingStatus",
"REGSOU": "LdsStatus",
"GRAVITYPIPE": "FreeFlow",
}
return names.get(self.name or "", str(int(self)))
def __str__(self) -> str:
raw_value = int(self)
if raw_value == 0:
return "0"
active_flags = [flag.backend_name for flag in type(self) if flag.value and flag & self == flag]
if active_flags:
return f"{'|'.join(active_flags)} ({raw_value})"
return str(raw_value)
class GravityPipe(Enum):
present_gravity = (1, "Наличие самотека")
absent_gravity = (0, "Отсутствие самотека")
def __init__(self, status_id: int, description: str) -> None:
self.id = status_id
self.description = description
def __str__(self):
return f"{self.id} - {self.description}"
class MeasureConversionRule(Enum):
MPA_MEASURE = "MPA_MEASURE"
KG_CM_MEASURE = "KG_CM_MEASURE"
