src/models.py - master - serohvostov/radio_proxy

src/models.py
327 строк14 KB
serohvostov
wip
22 фев 2026, 13:59
22 фев 2026, 13:597490624
"""
Модели данных для Сервиса Прокси Распознавания Речи
"""

from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
import struct
import time


@dataclass
class PacketHeader:
    """Структура заголовка пакета"""
    magic_flag: int = 0x47454223
    packet_type: int = 0
    data_length: int = 0
    
    # Максимальный разумный размер пакета (50 МБ)
    MAX_PACKET_SIZE: int = 50 * 1024 * 1024
    
    def to_bytes(self) -> bytes:
        """Преобразовать заголовок в байты используя little-endian формат"""
        return struct.pack('<III', self.magic_flag, self.packet_type, self.data_length)
    
    @classmethod
    def from_bytes(cls, data: bytes) -> 'PacketHeader':
        """Парсить заголовок из байтов"""
        if len(data) < 12:
            raise ValueError("Недостаточно данных для заголовка пакета")
        
        magic_flag, packet_type, data_length = struct.unpack('<III', data[:12])
        
        if magic_flag != 0x47454223:
            raise ValueError(f"Неверный магический флаг: {magic_flag:08x}")
        
        if data_length > cls.MAX_PACKET_SIZE:
            raise ValueError(f"Размер пакета слишком большой: {data_length} байт (максимум: {cls.MAX_PACKET_SIZE})")
        
        if data_length < 0:
            raise ValueError(f"Отрицательный размер пакета: {data_length}")
        
        return cls(magic_flag=magic_flag, packet_type=packet_type, data_length=data_length)


@dataclass
class RegistrationRequest:
    """Данные пакета запроса регистрации"""
    client_number: int
    protocol_version: int
    
    def __post_init__(self):
        """Validate fields after initialization"""
        if not isinstance(self.client_number, int) or self.client_number < 0:
            raise ValueError(f"Неверный client_number: {self.client_number}")
        if not isinstance(self.protocol_version, int) or self.protocol_version < 1:
            raise ValueError(f"Неверная protocol_version: {self.protocol_version}")


@dataclass
class RegistrationResponse:
    """Registration response packet data"""
    registration_result: int  # 0 = success, 1 = failure
    protocol_version: int
    
    def __post_init__(self):
        """Validate fields after initialization"""
        if self.registration_result not in (0, 1):
            raise ValueError(f"Invalid registration_result: {self.registration_result}")
        if not isinstance(self.protocol_version, int) or self.protocol_version < 1:
            raise ValueError(f"Invalid protocol_version: {self.protocol_version}")


@dataclass
class BitstreamPacket:
    """Bitstream packet data"""
    timestamp: int  # Unix timestamp in milliseconds
    unique_id: int
    content_type: int  # 1=WAV, 2=AMBE, 3=JPEG
    binary_data: bytes
    
    def __post_init__(self):
        """Validate fields after initialization"""
        if not isinstance(self.timestamp, int) or self.timestamp < 0:
            raise ValueError(f"Invalid timestamp: {self.timestamp}")
        if not isinstance(self.unique_id, int) or self.unique_id < 0:
            raise ValueError(f"Invalid unique_id: {self.unique_id}")
        if self.content_type not in (ContentTypes.UNKNOWN, ContentTypes.WAV, ContentTypes.AMBE, ContentTypes.JPEG):
            raise ValueError(f"Invalid content_type: {self.content_type}")
        if not isinstance(self.binary_data, bytes):
            raise ValueError(f"Invalid binary_data type: {type(self.binary_data)}")
        if len(self.binary_data) == 0:
            raise ValueError("binary_data cannot be empty")


@dataclass
class MetadataPacket:
    """Metadata packet data"""
    session_id: str  # UUID format (e.g., "c5a10612-96e9-4d7c-b4c5-c44f8ea98b7c")
    time_start: str  # ISO 8601 format
    time_end: str    # ISO 8601 format
    frequency: int
    signal_type: str
    radionet_name: str
    radiodata: List[Dict[str, Any]] = field(default_factory=list)
    files: List[int] = field(default_factory=list)  # Array of packet 1210 IDs
    places: List[Dict[str, Any]] = field(default_factory=list)
    Frequency_average: Optional[str] = None
    colour_code_s: Optional[int] = None
    is_bs: Optional[str] = None
    seance_type: Optional[str] = None
    tmo_dmo: Optional[str] = None
    from_phone: Optional[str] = None  # Поле "from" из пакета 1211
    to_phone: Optional[str] = None    # Поле "to" из пакета 1211
    rpp_name: Optional[str] = None    # Поле "rpp_name" из пакета 1211
    
    def __post_init__(self):
        """Validate fields after initialization"""
        if not isinstance(self.session_id, str) or not self.session_id.strip():
            raise ValueError(f"Invalid session_id: {self.session_id}")
        if not isinstance(self.time_start, str) or not self.time_start.strip():
            raise ValueError(f"Invalid time_start: {self.time_start}")
        # time_end может быть пустым
        if not isinstance(self.time_end, str):
            raise ValueError(f"Invalid time_end type: {type(self.time_end)}")
        # frequency может быть 0 (допустимое значение)
        if not isinstance(self.frequency, int) or self.frequency < 0:
            raise ValueError(f"Invalid frequency: {self.frequency}")
        if not isinstance(self.signal_type, str) or not self.signal_type.strip():
            raise ValueError(f"Invalid signal_type: {self.signal_type}")
        if not isinstance(self.radionet_name, str) or not self.radionet_name.strip():
            raise ValueError(f"Invalid radionet_name: {self.radionet_name}")
        if not isinstance(self.radiodata, list):
            raise ValueError(f"Invalid radiodata type: {type(self.radiodata)}")
        if not isinstance(self.files, list):
            raise ValueError(f"Invalid files type: {type(self.files)}")
        # Validate that all elements in files are integers
        for file_id in self.files:
            if not isinstance(file_id, int) or file_id < 0:
                raise ValueError(f"Invalid file_id in files array: {file_id}")
        if not isinstance(self.places, list):
            raise ValueError(f"Invalid places type: {type(self.places)}")


@dataclass
class ResultsPacket:
    """Results packet data"""
    request_id: int
    text: Optional[str] = None  # Present only if recognition successful
    
    def __post_init__(self):
        """Validate fields after initialization"""
        if not isinstance(self.request_id, int) or self.request_id < 0:
            raise ValueError(f"Invalid request_id: {self.request_id}")
        if self.text is not None and not isinstance(self.text, str):
            raise ValueError(f"Invalid text type: {type(self.text)}")


@dataclass
class Connection:
    """Connection state information"""
    connection_id: str
    client_number: Optional[int] = None
    protocol_version: Optional[int] = None
    is_registered: bool = False
    pending_requests: Dict[int, Any] = field(default_factory=dict)


@dataclass
class JobInfo:
    """Job tracking information"""
    job_id: str
    request_id: int
    connection_id: str
    audio_filepath: str
    metadata: Dict[str, Any]
    created_at: float = field(default_factory=time.time)  # Unix timestamp создания задания


# Packet type constants
class PacketTypes:
    REGISTRATION_REQUEST = 1
    REGISTRATION_RESPONSE = 2
    BITSTREAM = 1210
    METADATA = 1211
    RESULTS = 1212


# Content type constants
class ContentTypes:
    UNKNOWN = 0  # Неизвестный тип контента
    WAV = 1
    AMBE = 2
    JPEG = 3


# Grid mode models
@dataclass
class GridConfig:
    """Конфигурация режима grid"""
    mode: str = "ts"  # "ts" или "grid"
    files_directory: str = "./grid_files"
    results_directory: str = "./grid_results"
    filename_template_config: str = "./config/filename_template.json"
    jinja_template_path: str = "./config/templates/grid_format.j2"  # путь к Jinja2 шаблону
    monitoring_interval: float = 1.0  # секунды
    cleanup_processed_files: bool = True
    
    def __post_init__(self):
        """Валидация полей после инициализации"""
        if self.mode not in ("ts", "grid"):
            raise ValueError(f"Неверный режим: {self.mode}. Должен быть 'ts' или 'grid'")
        if not isinstance(self.files_directory, str) or not self.files_directory.strip():
            raise ValueError(f"Неверный files_directory: {self.files_directory}")
        if not isinstance(self.results_directory, str) or not self.results_directory.strip():
            raise ValueError(f"Неверный results_directory: {self.results_directory}")
        if not isinstance(self.filename_template_config, str) or not self.filename_template_config.strip():
            raise ValueError(f"Неверный filename_template_config: {self.filename_template_config}")
        if not isinstance(self.jinja_template_path, str) or not self.jinja_template_path.strip():
            raise ValueError(f"Неверный jinja_template_path: {self.jinja_template_path}")
        if not isinstance(self.monitoring_interval, (int, float)) or self.monitoring_interval <= 0:
            raise ValueError(f"Неверный monitoring_interval: {self.monitoring_interval}")


@dataclass
class FilenameTemplate:
    """Шаблон для генерации имен файлов"""
    template: str  # например, "${time}_${frequency}_test"
    fields: List[str]  # поля из пакета 1211 для использования
    constants: Dict[str, str] = field(default_factory=dict)  # константные значения
    time_format: str = "YYYY_MM_DD_HH_MM_SS"
    
    def __post_init__(self):
        """Валидация полей после инициализации"""
        if not isinstance(self.template, str) or not self.template.strip():
            raise ValueError(f"Неверный template: {self.template}")
        if not isinstance(self.fields, list):
            raise ValueError(f"Неверный тип fields: {type(self.fields)}")
        if not isinstance(self.constants, dict):
            raise ValueError(f"Неверный тип constants: {type(self.constants)}")
        if not isinstance(self.time_format, str) or not self.time_format.strip():
            raise ValueError(f"Неверный time_format: {self.time_format}")


@dataclass
class PacketPair:
    """Пара пакетов для режима grid"""
    bitstream_packet: Optional[BitstreamPacket] = None
    metadata_packet: Optional[MetadataPacket] = None
    timestamp: float = field(default_factory=time.time)
    
    def is_complete(self) -> bool:
        """Проверить, что пара пакетов полная"""
        return self.bitstream_packet is not None and self.metadata_packet is not None
    
    def __post_init__(self):
        """Валидация полей после инициализации"""
        if self.bitstream_packet is not None and not isinstance(self.bitstream_packet, BitstreamPacket):
            raise ValueError(f"Неверный тип bitstream_packet: {type(self.bitstream_packet)}")
        if self.metadata_packet is not None and not isinstance(self.metadata_packet, MetadataPacket):
            raise ValueError(f"Неверный тип metadata_packet: {type(self.metadata_packet)}")
        if not isinstance(self.timestamp, (int, float)) or self.timestamp < 0:
            raise ValueError(f"Неверный timestamp: {self.timestamp}")


@dataclass
class GridFileInfo:
    """Информация о файлах в режиме grid"""
    audio_filename: str
    json_filename: str
    request_id: int
    connection_id: str
    created_at: float = field(default_factory=time.time)
    
    def __post_init__(self):
        """Валидация полей после инициализации"""
        if not isinstance(self.audio_filename, str) or not self.audio_filename.strip():
            raise ValueError(f"Неверный audio_filename: {self.audio_filename}")
        if not isinstance(self.json_filename, str) or not self.json_filename.strip():
            raise ValueError(f"Неверный json_filename: {self.json_filename}")
        if not isinstance(self.request_id, int) or self.request_id < 0:
            raise ValueError(f"Неверный request_id: {self.request_id}")
        if not isinstance(self.connection_id, str):
            raise ValueError(f"Неверный тип connection_id: {type(self.connection_id)}")
        # Разрешаем пустые connection_id для тестовых сценариев
        if not isinstance(self.created_at, (int, float)) or self.created_at < 0:
            raise ValueError(f"Неверный created_at: {self.created_at}")


# Exception classes
class ProxyServiceError(Exception):
    """Base exception for proxy service"""
    pass


class ProtocolError(ProxyServiceError):
    """Protocol-related errors"""
    pass


class RecognitionServiceError(ProxyServiceError):
    """Recognition service errors"""
    pass


class FileProcessingError(ProxyServiceError):
    """File processing errors"""
    pass


class GridModeError(ProxyServiceError):
    """Ошибки режима grid"""
    pass


class TemplateConfigError(GridModeError):
    """Ошибки конфигурации шаблонов имен файлов"""
    pass


class JSONMappingError(GridModeError):
    """Ошибки конфигурации маппинга JSON"""
    pass


class FileMonitoringError(GridModeError):
    """Ошибки мониторинга файловой системы"""
    pass