Function verify_segment

apps/proxy/hls_proxy/server.py:622–676

Verify MPEG-TS segment integrity and structure. Args: data: Raw segment data bytes Returns: dict containing: valid (bool): True if segment passes all checks packets (int): Number of valid packets found size (int): Total s

(data: bytes)

# Update verify_segment with more thorough checks
def verify_segment(data: bytes) -> dict:
    """
    Verify MPEG-TS segment integrity and structure.
    
    Args:
        data: Raw segment data bytes
        
    Returns:
        dict containing:
            valid (bool): True if segment passes all checks
            packets (int): Number of valid packets found
            size (int): Total segment size in bytes
            error (str): Description if validation fails
            
    Checks:
    - Minimum size requirements
    - Packet size alignment
    - Sync byte presence
    - Transport error indicators
    """

    # Check minimum size
    if len(data) < 188:
        return {'valid': False, 'error': 'Segment too short'}
        
    # Verify segment size is multiple of packet size
    if len(data) % 188 != 0:
        return {'valid': False, 'error': 'Invalid segment size'}
    
    valid_packets = 0
    total_packets = len(data) // 188
    
    # Scan all packets in segment
    for i in range(0, len(data), 188):
        packet = data[i:i+188]
        
        # Check packet completeness
        if len(packet) != 188:
            return {'valid': False, 'error': 'Incomplete packet'}
            
        # Verify sync byte
        if packet[0] != 0x47:
            return {'valid': False, 'error': f'Invalid sync byte at offset {i}'}
            
        # Check transport error indicator
        if packet[1] & 0x80:
            return {'valid': False, 'error': 'Transport error indicator set'}
            
        valid_packets += 1
    
    return {
        'valid': True,
        'packets': valid_packets,
        'size': len(data)
    }

def fetch_stream(fetcher: StreamFetcher, stop_event: threading.Event, start_sequence: int = 0):
    """