Loki/loki/probe.py

import sys, subprocess, logging

def probe(path, ffprobe_exec='ffprobe', logger=None):
  streams = []
  
  # Logging
  if not logger:
    logger = logging.getLogger(__name__)
    logging.basicConfig(level=logging.WARN, stream=sys.stderr)

  # ffprobe
  args = [ ffprobe_exec, '-hide_banner', '-show_streams', path ]
  logger.debug(f"ffprobe command: {args}")
  probe = subprocess.run(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  logger.debug(f"ffprobe result: {probe}")
  data = probe.stdout.decode('utf-8')
  if probe.returncode != 0:
    logger.error(f"ffprobe error ({probe.returncode}): {probe.stderr.decode('utf-8')}")
    logger.debug(f"ffprobe stdout: {data}")
    sys.exit(probe.returncode)

  # Parse streams
  stream = None
  for line in data.splitlines():
    if line == "[STREAM]":
      logger.debug('Stream start')
      stream = {}
    elif line == "[/STREAM]":
      logger.debug('Stream end')
      streams.append(stream)
      steam = None
    elif stream is not None:
      logger.debug(f"\tStream data: {line}")
      try:
        (key, val) = line.split('=', 1)
      except ValueError:
        logger.error(f"Skipping unexpected stream data: {line}")
        continue
      stream[key] = val

  # Return
  return streams