pretty good
This commit is contained in:
186
analyzer/protocols/decoders/image.py
Normal file
186
analyzer/protocols/decoders/image.py
Normal file
@@ -0,0 +1,186 @@
|
||||
"""
|
||||
Image Data decoders for Chapter 10 data types
|
||||
Supports Image Data Formats 2-7 (0x4A-0x4F)
|
||||
"""
|
||||
|
||||
import struct
|
||||
from typing import Dict, Any, Optional, Tuple
|
||||
from .base import DataTypeDecoder, DecodedPayload
|
||||
|
||||
|
||||
class ImageDecoder(DataTypeDecoder):
|
||||
"""Decoder for Image Data types (0x4A-0x4F)"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.data_type_base = 0x4A
|
||||
self.data_type_name = "Image Data"
|
||||
self.supported_formats = list(range(0x4A, 0x50))
|
||||
|
||||
def can_decode(self, data_type: int) -> bool:
|
||||
return 0x4A <= data_type <= 0x4F
|
||||
|
||||
def get_data_type_name(self, data_type: int) -> str:
|
||||
format_names = {
|
||||
0x4A: "Image Data Format 2 (Dynamic Imagery)",
|
||||
0x4B: "Image Data Format 3",
|
||||
0x4C: "Image Data Format 4",
|
||||
0x4D: "Image Data Format 5",
|
||||
0x4E: "Image Data Format 6",
|
||||
0x4F: "Image Data Format 7"
|
||||
}
|
||||
return format_names.get(data_type, f"Image Data Format {data_type & 0x0F}")
|
||||
|
||||
def decode(self, payload: bytes, ch10_header: Dict[str, Any]) -> Optional[DecodedPayload]:
|
||||
"""Decode Image Data payload"""
|
||||
data_type = ch10_header.get('data_type', 0)
|
||||
|
||||
if not self.can_decode(data_type):
|
||||
return None
|
||||
|
||||
if data_type == 0x4A:
|
||||
return self._decode_dynamic_imagery(payload, ch10_header)
|
||||
else:
|
||||
return self._decode_generic_image(payload, ch10_header)
|
||||
|
||||
def _decode_dynamic_imagery(self, payload: bytes, ch10_header: Dict[str, Any]) -> DecodedPayload:
|
||||
"""Decode Dynamic Imagery (Format 2)"""
|
||||
decoded_data = {}
|
||||
errors = []
|
||||
|
||||
# Parse IPH
|
||||
iph = self._parse_intra_packet_header(payload)
|
||||
if iph:
|
||||
decoded_data.update(iph)
|
||||
data_start = iph['data_start']
|
||||
else:
|
||||
data_start = 0
|
||||
errors.append("Failed to parse intra-packet header")
|
||||
|
||||
# Parse image header
|
||||
if data_start + 32 <= len(payload):
|
||||
img_header = self._safe_unpack('<IIHHHHHHHHHH', payload, data_start)
|
||||
if img_header:
|
||||
decoded_data.update({
|
||||
'image_timestamp': img_header[0],
|
||||
'image_id': img_header[1],
|
||||
'image_format': img_header[2],
|
||||
'image_width': img_header[3],
|
||||
'image_height': img_header[4],
|
||||
'bits_per_pixel': img_header[5],
|
||||
'compression_type': img_header[6],
|
||||
'image_size': img_header[7],
|
||||
'x_offset': img_header[8],
|
||||
'y_offset': img_header[9],
|
||||
'frame_number': img_header[10],
|
||||
'reserved': img_header[11]
|
||||
})
|
||||
|
||||
# Decode format and compression
|
||||
decoded_data['format_description'] = self._decode_image_format(img_header[2])
|
||||
decoded_data['compression_description'] = self._decode_compression(img_header[6])
|
||||
|
||||
# Extract image data
|
||||
image_data_start = data_start + 32
|
||||
image_size = img_header[7]
|
||||
if image_data_start + image_size <= len(payload):
|
||||
image_data = payload[image_data_start:image_data_start + image_size]
|
||||
decoded_data['image_data_length'] = len(image_data)
|
||||
decoded_data['image_data_hash'] = hash(image_data) & 0xFFFFFFFF
|
||||
|
||||
# Don't include raw image data in output for performance
|
||||
# Store first few bytes for analysis
|
||||
decoded_data['image_header_bytes'] = image_data[:16].hex()
|
||||
else:
|
||||
errors.append("Image data extends beyond payload")
|
||||
else:
|
||||
errors.append("Failed to parse image header")
|
||||
else:
|
||||
errors.append("Insufficient data for image header")
|
||||
|
||||
return DecodedPayload(
|
||||
data_type=0x4A,
|
||||
data_type_name="Dynamic Imagery",
|
||||
format_version=2,
|
||||
decoded_data=decoded_data,
|
||||
raw_payload=payload,
|
||||
errors=errors,
|
||||
metadata={'decoder': 'ImageDecoder'}
|
||||
)
|
||||
|
||||
def _decode_generic_image(self, payload: bytes, ch10_header: Dict[str, Any]) -> DecodedPayload:
|
||||
"""Decode generic image data (Formats 3-7)"""
|
||||
data_type = ch10_header.get('data_type', 0)
|
||||
decoded_data = {}
|
||||
errors = []
|
||||
|
||||
# Parse IPH
|
||||
iph = self._parse_intra_packet_header(payload)
|
||||
if iph:
|
||||
decoded_data.update(iph)
|
||||
data_start = iph['data_start']
|
||||
else:
|
||||
data_start = 0
|
||||
errors.append("Failed to parse intra-packet header")
|
||||
|
||||
# Generic image data parsing
|
||||
if data_start < len(payload):
|
||||
image_data = payload[data_start:]
|
||||
decoded_data['image_data_length'] = len(image_data)
|
||||
decoded_data['image_data_hash'] = hash(image_data) & 0xFFFFFFFF
|
||||
decoded_data['header_bytes'] = image_data[:32].hex() if len(image_data) >= 32 else image_data.hex()
|
||||
|
||||
# Try to identify image format from magic bytes
|
||||
format_info = self._identify_image_format(image_data)
|
||||
decoded_data.update(format_info)
|
||||
|
||||
return DecodedPayload(
|
||||
data_type=data_type,
|
||||
data_type_name=self.get_data_type_name(data_type),
|
||||
format_version=data_type & 0x0F,
|
||||
decoded_data=decoded_data,
|
||||
raw_payload=payload,
|
||||
errors=errors,
|
||||
metadata={'decoder': 'ImageDecoder'}
|
||||
)
|
||||
|
||||
def _decode_image_format(self, format_code: int) -> str:
|
||||
"""Decode image format code"""
|
||||
formats = {
|
||||
0: "Monochrome",
|
||||
1: "RGB",
|
||||
2: "YUV 4:2:2",
|
||||
3: "YUV 4:2:0",
|
||||
4: "RGBA",
|
||||
5: "Bayer Pattern"
|
||||
}
|
||||
return formats.get(format_code, f"Unknown ({format_code})")
|
||||
|
||||
def _decode_compression(self, compression_code: int) -> str:
|
||||
"""Decode compression type"""
|
||||
compressions = {
|
||||
0: "Uncompressed",
|
||||
1: "JPEG",
|
||||
2: "H.264",
|
||||
3: "MPEG-2",
|
||||
4: "PNG",
|
||||
5: "Lossless"
|
||||
}
|
||||
return compressions.get(compression_code, f"Unknown ({compression_code})")
|
||||
|
||||
def _identify_image_format(self, data: bytes) -> Dict[str, Any]:
|
||||
"""Identify image format from magic bytes"""
|
||||
if len(data) < 8:
|
||||
return {'detected_format': 'Unknown (insufficient data)'}
|
||||
|
||||
# Check common image formats
|
||||
if data[:2] == b'\xFF\xD8':
|
||||
return {'detected_format': 'JPEG', 'magic_bytes': data[:4].hex()}
|
||||
elif data[:8] == b'\x89PNG\r\n\x1a\n':
|
||||
return {'detected_format': 'PNG', 'magic_bytes': data[:8].hex()}
|
||||
elif data[:4] in [b'RIFF', b'AVI ']:
|
||||
return {'detected_format': 'AVI/RIFF', 'magic_bytes': data[:4].hex()}
|
||||
elif data[:4] == b'\x00\x00\x00\x20' or data[:4] == b'\x00\x00\x00\x18':
|
||||
return {'detected_format': 'AVIF/HEIF', 'magic_bytes': data[:4].hex()}
|
||||
else:
|
||||
return {'detected_format': 'Unknown/Raw', 'magic_bytes': data[:8].hex()}
|
||||
Reference in New Issue
Block a user