104 lines
4.1 KiB
Python
104 lines
4.1 KiB
Python
"""Helpers for saving raw payload bins and rotating saved files for VideoReceiverSFP.
|
|
"""
|
|
import os
|
|
import logging
|
|
from collections import deque
|
|
|
|
logger = logging.getLogger("VideoReceiverSFP.payload_saver")
|
|
|
|
|
|
def save_bin_payload(module, payload: bytes, ity: int, dbgdir: str, ts: str, force_save: bool = False) -> str:
|
|
"""Save raw payload to disk, using module._dump_manager when available.
|
|
Returns path or None.
|
|
"""
|
|
try:
|
|
binpath = None
|
|
if (ity == 1 and getattr(module, '_save_bin', False) is False and not force_save) and (ity != 1 and ity != 2):
|
|
# If not forced and not configured, skip
|
|
pass
|
|
# Use DumpManager if available
|
|
if getattr(module, '_dump_manager', None) is not None:
|
|
try:
|
|
category = 'mfd' if ity == 1 else ('sar' if ity == 2 else 'unknown')
|
|
binpath = module._dump_manager.save_bin(payload, category=category)
|
|
return binpath
|
|
except Exception:
|
|
logger.exception('save_bin_payload: DumpManager.save_bin failed')
|
|
# Fallback write
|
|
try:
|
|
os.makedirs(dbgdir, exist_ok=True)
|
|
fname = f'VideoReceiverSFP_payload_{ts}.bin'
|
|
binpath = os.path.join(dbgdir, fname)
|
|
with open(binpath, 'wb') as fh:
|
|
fh.write(payload)
|
|
# maintain module._saved_bins if present
|
|
try:
|
|
sb = getattr(module, '_saved_bins', None)
|
|
if sb is None:
|
|
module._saved_bins = deque()
|
|
sb = module._saved_bins
|
|
sb.append(binpath)
|
|
keep = getattr(module, '_dump_keep', 100)
|
|
while len(sb) > keep:
|
|
old = sb.popleft()
|
|
try:
|
|
os.unlink(old)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
logger.exception('save_bin_payload: failed to rotate saved bins')
|
|
return binpath
|
|
except Exception:
|
|
logger.exception('save_bin_payload: fallback write failed')
|
|
return None
|
|
except Exception:
|
|
logger.exception('save_bin_payload: unexpected error')
|
|
return None
|
|
|
|
|
|
def save_unknown_sample(module, payload: bytes, dbgdir: str, ts: str) -> str:
|
|
"""Save an unknown-type payload sample, using DumpManager if available.
|
|
Mirrors the diagnostic save in original code.
|
|
"""
|
|
try:
|
|
if getattr(module, '_dump_manager', None) is not None:
|
|
try:
|
|
path = module._dump_manager.save_bin(payload, category='unknown')
|
|
if path:
|
|
logging.getLogger().info('VideoReceiverSFP: saved unknown-type payload to %s', path)
|
|
return path
|
|
except Exception:
|
|
logger.exception('save_unknown_sample: DumpManager.save_bin failed')
|
|
# Fallback
|
|
try:
|
|
os.makedirs(dbgdir, exist_ok=True)
|
|
fname = f'VideoReceiverSFP_unknown_TYPE_{ts}.bin'
|
|
path = os.path.join(dbgdir, fname)
|
|
with open(path, 'wb') as fh:
|
|
fh.write(payload)
|
|
# rotate
|
|
try:
|
|
sb = getattr(module, '_saved_bins', None)
|
|
if sb is None:
|
|
from collections import deque
|
|
module._saved_bins = deque()
|
|
sb = module._saved_bins
|
|
sb.append(path)
|
|
keep = getattr(module, '_dump_keep', 100)
|
|
while len(sb) > keep:
|
|
old = sb.popleft()
|
|
try:
|
|
os.unlink(old)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
logger.exception('save_unknown_sample: failed to rotate saved bins')
|
|
logging.getLogger().info('VideoReceiverSFP: saved unknown-type payload to %s', path)
|
|
return path
|
|
except Exception:
|
|
logger.exception('save_unknown_sample: fallback write failed')
|
|
return None
|
|
except Exception:
|
|
logger.exception('save_unknown_sample: unexpected error')
|
|
return None
|