<commit_before># This file is part of beets.
# Copyright 2015, <NAME>, <NAME>, and <NAME>.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.

from __future__ import (division, absolute_import, print_function,
                        unicode_literals)

import subprocess
import os
import collections
import itertools
import sys
import warnings
import re

from beets import logging
from beets import ui
from beets.plugins import BeetsPlugin
from beets.util import syspath, command_output, displayable_path
from beets import config


# Utilities.

class ReplayGainError(Exception):
    """Raised when a local (to a track or an album) error occurs in one
    of the backends.
    """


class FatalReplayGainError(Exception):
    """Raised when a fatal error occurs in one of the backends.
    """


def call(args):
    """Execute the command and return its output or raise a
    ReplayGainError on failure.
    """
    try:
        return command_output(args)
    except subprocess.CalledProcessError as e:
        raise ReplayGainError(
            "{0} exited with status {1}".format(args[0], e.returncode)
        )
    except UnicodeEncodeError:
        # Due to a bug in Python 2's subprocess on Windows, Unicode
        # filenames can fail to encode on that platform. See:
        # http://code.google.com/p/beets/issues/detail?id=499
        raise ReplayGainError("argument encoding failed")


# Backend base and plumbing classes.

Gain = collections.namedtuple("Gain", "gain peak")
AlbumGain = collections.namedtuple("AlbumGain", "album_gain track_gains")


class Backend(object):
    """An abstract class representing engine for calculating RG values.
    """

    def __init__(self, config, log):
        """Initialize the backend with the configuration view for the
        plugin.
        """
        self._log = log

    def compute_track_gain(self, items):
        raise NotImplementedError()

    def compute_album_gain(self, album):
        # TODO: implement album gain in terms of track gain of the
        # individual tracks which can be used for any backend.
        raise NotImplementedError()


# bsg1770gain backend
class Bs1770gainBackend(Backend):
    """bs1770gain is a loudness scanner compliant with ITU-R BS.1770 and
    its flavors EBU R128, ATSC A/85 and Replaygain 2.0.
    """

    def __init__(self, config, log):
        super(Bs1770gainBackend, self).__init__(config, log)
        config.add({
            'chunk_at': 5000,
            'method': 'replaygain',
        })
        self.chunk_at = config['chunk_at'].as_number()
        self.method = b'--' + bytes(config['method'].get(unicode))

        cmd = b'bs1770gain'
        try:
            call([cmd, self.method])
            self.command = cmd
        except OSError:
            raise FatalReplayGainError(
                'Is bs1770gain installed? Is your method in config correct?'
            )
        if not self.command:
            raise FatalReplayGainError(
                'no replaygain command found: install bs1770gain'
            )

    def compute_track_gain(self, items):
        """Computes the track gain of the given tracks, returns a list
        of TrackGain objects.
        """

        output = self.compute_gain(items, False)
        return output

    def compute_album_gain(self, album):
        """Computes the album gain of the given album, returns an
        AlbumGain object.
        """
        # TODO: What should be done when not all tracks in the album are
        # supported?

        supported_items = album.items()
        output = self.compute_gain(supported_items, True)

        return AlbumGain(output[-1], output[:-1])

    def isplitter(self, items, chunk_at):
        """Break an iterable into chunks of at most size `chunk_at`,
        generating lists for each chunk.
        """
        iterable = iter(items)
        while True:
            result = []
            for i in range(chunk_at):
                try:
                    a = next(iterable)
                except StopIteration:
                    break
                else:
                    result.append(a)
            if result:
                yield result
            else:
                break

    def compute_gain(self, items, is_album):
        """Computes the track or album gain of a list of items, returns
        a list of TrackGain objects.
        When computing album gain, the last TrackGain object returned is
        the album gain
        """

        if len(items) == 0:
            return []

        albumgaintot = 0.0
        albumpeaktot = 0.0
        returnchunks = []

        # In the case of very large sets of music, we break the tracks
        # into smaller chunks and process them one at a time. This
        # avoids running out of memory.
        if len(items) > self.chunk_at:
            i = 0
            for chunk in self.isplitter(items, self.chunk_at):
                i += 1
                returnchunk = self.compute_chunk_gain(chunk, is_album)
                albumgaintot += returnchunk[-1].gain
                albumpeaktot += returnchunk[-1].peak
                returnchunks = returnchunks + returnchunk[0:-1]
            returnchunks.append(Gain(albumgaintot / i, albumpeaktot / i))
            return returnchunks
        else:
            return self.compute_chunk_gain(items, is_album)

    def compute_chunk_gain(self, items, is_album):
        """Compute ReplayGain values and return a list of results
        dictionaries as given by `parse_tool_output`.
        """
        # Construct shell command.
        cmd = [self.command]
        cmd = cmd + [self.method]
        cmd = cmd + [b'-it']
        # workaround for windows MAX_PATH prefix, will get problems
        # when path is too long on windows
        try:
            output = call(cmd + [syspath(i.path) for i in items])
            if not output:
                output = call(cmd +
                              [syspath(i.path, prefix=False) for i in items])
        except:
            self._log.debug(u'bsgain1770 failed')
        self._log.debug(u'analysis finished:{0}', output)
        results = self.parse_tool_output(output,
                                         len(items) + is_album)
        self._log.debug(u'{0} items/ {1} results', len(items), len(results))
        return results

    def parse_tool_output(self, text, num_lines):
        """Given the  output from bs1770gain, parse the text and
        return a list of dictionaries
        containing information about each analyzed file.
        """
        out = []
        data = text.decode('utf8', errors='ignore')
        regex = re.compile(
            ur'(\s{2,2}\[\d+\/\d+\].*?|\[ALBUM\].*?\
                )(?=\s{2,2}\[\d+\/\d+\]|\s{2,2}\[ALBUM\]:|done\.\s\
                )', re.DOTALL | re.UNICODE)
        results = re.findall(regex, data)
        for parts in results[0:num_lines]:
            part = parts.split(b'\n')
            if len(part) == 0:
                self._log.debug(u'bad tool output: {0!r}', text)
                raise ReplayGainError('bs1770gain failed')<commit_msg>fix no-PREFIX for long window-paths<commit_after># This file is part of beets.
# Copyright 2015, <NAME>, <NAME>, and <NAME>.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.

from __future__ import (division, absolute_import, print_function,
                        unicode_literals)

import subprocess
import os
import collections
import itertools
import sys
import warnings
import re

from beets import logging
from beets import ui
from beets.plugins import BeetsPlugin
from beets.util import syspath, command_output, displayable_path
from beets import config


# Utilities.

class ReplayGainError(Exception):
    """Raised when a local (to a track or an album) error occurs in one
    of the backends.
    """


class FatalReplayGainError(Exception):
    """Raised when a fatal error occurs in one of the backends.
    """


def call(args):
    """Execute the command and return its output or raise a
    ReplayGainError on failure.
    """
    try:
        return command_output(args)
    except subprocess.CalledProcessError as e:
        raise ReplayGainError(
            "{0} exited with status {1}".format(args[0], e.returncode)
        )
    except UnicodeEncodeError:
        # Due to a bug in Python 2's subprocess on Windows, Unicode
        # filenames can fail to encode on that platform. See:
        # http://code.google.com/p/beets/issues/detail?id=499
        raise ReplayGainError("argument encoding failed")


# Backend base and plumbing classes.

Gain = collections.namedtuple("Gain", "gain peak")
AlbumGain = collections.namedtuple("AlbumGain", "album_gain track_gains")


class Backend(object):
    """An abstract class representing engine for calculating RG values.
    """

    def __init__(self, config, log):
        """Initialize the backend with the configuration view for the
        plugin.
        """
        self._log = log

    def compute_track_gain(self, items):
        raise NotImplementedError()

    def compute_album_gain(self, album):
        # TODO: implement album gain in terms of track gain of the
        # individual tracks which can be used for any backend.
        raise NotImplementedError()


# bsg1770gain backend
class Bs1770gainBackend(Backend):
    """bs1770gain is a loudness scanner compliant with ITU-R BS.1770 and
    its flavors EBU R128, ATSC A/85 and Replaygain 2.0.
    """

    def __init__(self, config, log):
        super(Bs1770gainBackend, self).__init__(config, log)
        config.add({
            'chunk_at': 5000,
            'method': 'replaygain',
        })
        self.chunk_at = config['chunk_at'].as_number()
        self.method = b'--' + bytes(config['method'].get(unicode))

        cmd = b'bs1770gain'
        try:
            call([cmd, self.method])
            self.command = cmd
        except OSError:
            raise FatalReplayGainError(
                'Is bs1770gain installed? Is your method in config correct?'
            )
        if not self.command:
            raise FatalReplayGainError(
                'no replaygain command found: install bs1770gain'
            )

    def compute_track_gain(self, items):
        """Computes the track gain of the given tracks, returns a list
        of TrackGain objects.
        """

        output = self.compute_gain(items, False)
        return output

    def compute_album_gain(self, album):
        """Computes the album gain of the given album, returns an
        AlbumGain object.
        """
        # TODO: What should be done when not all tracks in the album are
        # supported?

        supported_items = album.items()
        output = self.compute_gain(supported_items, True)

        return AlbumGain(output[-1], output[:-1])

    def isplitter(self, items, chunk_at):
        """Break an iterable into chunks of at most size `chunk_at`,
        generating lists for each chunk.
        """
        iterable = iter(items)
        while True:
            result = []
            for i in range(chunk_at):
                try:
                    a = next(iterable)
                except StopIteration:
                    break
                else:
                    result.append(a)
            if result:
                yield result
            else:
                break

    def compute_gain(self, items, is_album):
        """Computes the track or album gain of a list of items, returns
        a list of TrackGain objects.
        When computing album gain, the last TrackGain object returned is
        the album gain
        """

        if len(items) == 0:
            return []

        albumgaintot = 0.0
        albumpeaktot = 0.0
        returnchunks = []

        # In the case of very large sets of music, we break the tracks
        # into smaller chunks and process them one at a time. This
        # avoids running out of memory.
        if len(items) > self.chunk_at:
            i = 0
            for chunk in self.isplitter(items, self.chunk_at):
                i += 1
                returnchunk = self.compute_chunk_gain(chunk, is_album)
                albumgaintot += returnchunk[-1].gain
                albumpeaktot += returnchunk[-1].peak
                returnchunks = returnchunks + returnchunk[0:-1]
            returnchunks.append(Gain(albumgaintot / i, albumpeaktot / i))
            return returnchunks
        else:
            return self.compute_chunk_gain(items, is_album)

    def compute_chunk_gain(self, items, is_album):
        """Compute ReplayGain values and return a list of results
        dictionaries as given by `parse_tool_output`.
        """
        # Construct shell command.
        cmd = [self.command]
        cmd = cmd + [self.method]
        cmd = cmd + [b'-it']
        # workaround for windows MAX_PATH prefix, will get problems
        # when path is too long on windows
        try:
            output = call(cmd +
                          [syspath(i.path, prefix=False) for i in items])
        except:
            self._log.debug(u'bsgain1770 failed')
        self._log.debug(u'analysis finished:{0}', output)
        results = self.parse_tool_output(output,
                                         len(items) + is_album)
        self._log.debug(u'{0} items/ {1} results', len(items), len(results))
        return results

    def parse_tool_output(self, text, num_lines):
        """Given the  output from bs1770gain, parse the text and
        return a list of dictionaries
        containing information about each analyzed file.
        """
        out = []
        data = text.decode('utf8', errors='ignore')
        regex = re.compile(
            ur'(\s{2,2}\[\d+\/\d+\].*?|\[ALBUM\].*?\
                )(?=\s{2,2}\[\d+\/\d+\]|\s{2,2}\[ALBUM\]:|done\.\s\
                )', re.DOTALL | re.UNICODE)
        results = re.findall(regex, data)
        for parts in results[0:num_lines]:
            part = parts.split(b'\n')
            if len(part) == 0:
                self._log.debug(u'bad tool output: {0!r}', text)
                raise ReplayGainError('bs1770gain failed')
{'fraction_non_alphanumeric': 0.09047211430828998, 'fraction_numerical': 0.012115625205768091, 'mean_word_length': 2.1419114604882084, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7521566', 'n_tokens_mistral': 4349, 'n_tokens_neox': 4119, 'n_words': 1419}