<commit_before>import os
import re
import tempfile
import shutil
import json
from tqdm import tqdm

from ._streams import AudioStream, VideoStream, SubtitleStream
from ._errors import FFmpegNormalizeError
from ._cmd_utils import NUL, CommandRunner, DUR_REGEX, to_ms
from ._logger import setup_custom_logger
logger = setup_custom_logger('ffmpeg_normalize')


class MediaFile():
    """
    Class that holds a file, its streams and adjustments
    """

    def __init__(self, ffmpeg_normalize, input_file, output_file=None):
        """
        Initialize a media file for later normalization.

        Arguments:
            ffmpeg_normalize {FFmpegNormalize} -- reference to overall settings
            input_file {str} -- Path to input file

        Keyword Arguments:
            output_file {str} -- Path to output file (default: {None})
        """
        self.ffmpeg_normalize = ffmpeg_normalize
        self.skip = False
        self.input_file = input_file
        self.output_file = output_file
        self.streams = {
            'audio': {},
            'video': {},
            'subtitle': {}
        }

        self.parse_streams()

    def _stream_ids(self):
        return list(self.streams['audio'].keys()) + \
            list(self.streams['video'].keys()) + \
            list(self.streams['subtitle'].keys())

    def __repr__(self):
        return os.path.basename(self.input_file)

    def parse_streams(self):
        """
        Try to parse all input streams from file
        """
        logger.debug("Parsing streams of {}".format(self.input_file))

        cmd = [
            self.ffmpeg_normalize.ffmpeg_exe, '-i', self.input_file,
            '-c', 'copy', '-t', '0', '-map', '0',
            '-f', 'null', NUL
        ]

        cmd_runner = CommandRunner(cmd)
        cmd_runner.run_command()
        output = cmd_runner.get_output()

        logger.debug("Stream parsing command output:")
        logger.debug(output)

        output_lines = [line.strip() for line in output.split('\n')]

        duration = None
        for line in output_lines:

            if 'Duration' in line:
                duration_search = DUR_REGEX.search(line)
                if not duration_search:
                    logger.warning("Could not extract duration from input file!")
                else:
                    duration = duration_search.groupdict()
                    duration = to_ms(**duration) / 1000
                    logger.debug("Found duration: " + str(duration) + " s")

            if not line.startswith('Stream'):
                continue

            stream_id_match = re.search(r'#0:([\d]+)', line)
            if stream_id_match:
                stream_id = int(stream_id_match.group(1))
                if stream_id in self._stream_ids():
                    continue
            else:
                continue


            if 'Audio' in line:
                logger.debug("Found audio stream at index {}".format(stream_id))
                sample_rate_match = re.search(r'(\d+) Hz', line)
                sample_rate = int(sample_rate_match.group(1)) if sample_rate_match else None
                bit_depth_match = re.search(r's(\d+)p?,', line)
                bit_depth = int(bit_depth_match.group(1)) if bit_depth_match else None
                self.streams['audio'][stream_id] = AudioStream(self, stream_id, sample_rate, bit_depth, duration)

            elif 'Video' in line:
                logger.debug("Found video stream at index {}".format(stream_id))
                self.streams['video'][stream_id] = VideoStream(self, stream_id)

            elif 'Subtitle' in line:
                logger.debug("Found subtitle stream at index {}".format(stream_id))
                self.streams['subtitle'][stream_id] = SubtitleStream(self, stream_id)

        if not self.streams['audio']:
            raise FFmpegNormalizeError(
                "Input file {} does not contain any audio streams"
                .format(self.input_file))

        if os.path.splitext(self.output_file)[1].lower() in ['.wav', '.mp3', '.aac']:
            logger.warning(
                "Output file only supports one stream. "
                "Keeping only first audio stream."
            )
            first_stream = list(self.streams['audio'].values())[0]
            self.streams['audio'] = {first_stream.stream_id: first_stream}
            self.streams['video'] = {}
            self.streams['subtitle'] = {}

    def run_normalization(self):
        logger.debug("Running normalization for {}".format(self.input_file))

        # run the first pass to get loudness stats
        self._first_pass()

        # run the second pass as a whole
        if self.ffmpeg_normalize.progress:
            with tqdm(
                total=100,
                position=1,
                desc="Second Pass"
            ) as pbar:
                for progress in self._second_pass():
                    pbar.update(progress - pbar.n)
        else:
            for _ in self._second_pass():
                pass

    def _first_pass(self):
        logger.debug("Parsing normalization info for {}".format(self.input_file))<commit_msg>do not print stream warning when there is only one stream<commit_after>import os
import re
import tempfile
import shutil
import json
from tqdm import tqdm

from ._streams import AudioStream, VideoStream, SubtitleStream
from ._errors import FFmpegNormalizeError
from ._cmd_utils import NUL, CommandRunner, DUR_REGEX, to_ms
from ._logger import setup_custom_logger
logger = setup_custom_logger('ffmpeg_normalize')


class MediaFile():
    """
    Class that holds a file, its streams and adjustments
    """

    def __init__(self, ffmpeg_normalize, input_file, output_file=None):
        """
        Initialize a media file for later normalization.

        Arguments:
            ffmpeg_normalize {FFmpegNormalize} -- reference to overall settings
            input_file {str} -- Path to input file

        Keyword Arguments:
            output_file {str} -- Path to output file (default: {None})
        """
        self.ffmpeg_normalize = ffmpeg_normalize
        self.skip = False
        self.input_file = input_file
        self.output_file = output_file
        self.streams = {
            'audio': {},
            'video': {},
            'subtitle': {}
        }

        self.parse_streams()

    def _stream_ids(self):
        return list(self.streams['audio'].keys()) + \
            list(self.streams['video'].keys()) + \
            list(self.streams['subtitle'].keys())

    def __repr__(self):
        return os.path.basename(self.input_file)

    def parse_streams(self):
        """
        Try to parse all input streams from file
        """
        logger.debug("Parsing streams of {}".format(self.input_file))

        cmd = [
            self.ffmpeg_normalize.ffmpeg_exe, '-i', self.input_file,
            '-c', 'copy', '-t', '0', '-map', '0',
            '-f', 'null', NUL
        ]

        cmd_runner = CommandRunner(cmd)
        cmd_runner.run_command()
        output = cmd_runner.get_output()

        logger.debug("Stream parsing command output:")
        logger.debug(output)

        output_lines = [line.strip() for line in output.split('\n')]

        duration = None
        for line in output_lines:

            if 'Duration' in line:
                duration_search = DUR_REGEX.search(line)
                if not duration_search:
                    logger.warning("Could not extract duration from input file!")
                else:
                    duration = duration_search.groupdict()
                    duration = to_ms(**duration) / 1000
                    logger.debug("Found duration: " + str(duration) + " s")

            if not line.startswith('Stream'):
                continue

            stream_id_match = re.search(r'#0:([\d]+)', line)
            if stream_id_match:
                stream_id = int(stream_id_match.group(1))
                if stream_id in self._stream_ids():
                    continue
            else:
                continue


            if 'Audio' in line:
                logger.debug("Found audio stream at index {}".format(stream_id))
                sample_rate_match = re.search(r'(\d+) Hz', line)
                sample_rate = int(sample_rate_match.group(1)) if sample_rate_match else None
                bit_depth_match = re.search(r's(\d+)p?,', line)
                bit_depth = int(bit_depth_match.group(1)) if bit_depth_match else None
                self.streams['audio'][stream_id] = AudioStream(self, stream_id, sample_rate, bit_depth, duration)

            elif 'Video' in line:
                logger.debug("Found video stream at index {}".format(stream_id))
                self.streams['video'][stream_id] = VideoStream(self, stream_id)

            elif 'Subtitle' in line:
                logger.debug("Found subtitle stream at index {}".format(stream_id))
                self.streams['subtitle'][stream_id] = SubtitleStream(self, stream_id)

        if not self.streams['audio']:
            raise FFmpegNormalizeError(
                "Input file {} does not contain any audio streams"
                .format(self.input_file))

        if (
            os.path.splitext(self.output_file)[1].lower() in ['.wav', '.mp3', '.aac'] and
            len(self.streams['audio'].values()) > 1
        ):
            logger.warning(
                "Output file only supports one stream. "
                "Keeping only first audio stream."
            )
            first_stream = list(self.streams['audio'].values())[0]
            self.streams['audio'] = {first_stream.stream_id: first_stream}
            self.streams['video'] = {}
            self.streams['subtitle'] = {}

    def run_normalization(self):
        logger.debug("Running normalization for {}".format(self.input_file))

        # run the first pass to get loudness stats
        self._first_pass()

        # run the second pass as a whole
        if self.ffmpeg_normalize.progress:
            with tqdm(
                total=100,
                position=1,
                desc="Second Pass"
            ) as pbar:
                for progress in self._second_pass():
                    pbar.update(progress - pbar.n)
        else:
            for _ in self._second_pass():
                pass

    def _first_pass(self):
        logger.debug("Parsing normalization info for {}".format(self.input_file))
{'fraction_non_alphanumeric': 0.10827110855829983, 'fraction_numerical': 0.003350564809496458, 'mean_word_length': 1.9287917017101206, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4868626', 'n_tokens_mistral': 3001, 'n_tokens_neox': 2849, 'n_words': 710}