





import logging
import math
from functools import partial
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union

import torch

from .tensor_utils import tensor_tree_map, tree_map


def 							__snake_case							(    lowerCAmelCase_	)				->						List[Tuple[int, ...]]:
    SCREAMING_SNAKE_CASE__											=			[]
    if isinstance(lowerCAmelCase_ ,							lowerCAmelCase_	):
        for v in tree.values():
            shapes.extend(_fetch_dims(lowerCAmelCase_	)	)
    elif isinstance(lowerCAmelCase_ ,							(list, tuple)	):
        for t in tree:
            shapes.extend(_fetch_dims(lowerCAmelCase_	)	)
    elif isinstance(lowerCAmelCase_ ,							torch.Tensor	):
        shapes.append(tree.shape	)
    else:
        raise ValueError('''Not supported'''	)

    return shapes


@torch.jit.ignore
def 							__snake_case							(    lowerCAmelCase_ ,							lowerCAmelCase_	)				->						Tuple[int, ...]:
    SCREAMING_SNAKE_CASE__											=			[]
    for d in reversed(lowerCAmelCase_	):
        idx.append(flat_idx % d	)
        SCREAMING_SNAKE_CASE__											=			flat_idx // d

    return tuple(reversed(lowerCAmelCase_	)	)


@torch.jit.ignore
def 							__snake_case							(    lowerCAmelCase_ ,							lowerCAmelCase_ ,							lowerCAmelCase_ ,							lowerCAmelCase_ = None ,							lowerCAmelCase_ = None ,							)				->						List[Tuple[slice, ...]]:

    # start_edges and end_edges both indicate whether, starting from any given
    # dimension, the start/end index is at the top/bottom edge of the
    # corresponding tensor, modeled as a tree
    def reduce_edge_list(lowerCAmelCase_	) -> None:
        SCREAMING_SNAKE_CASE__											=			True
        for i in range(len(lowerCAmelCase_	)	):
            SCREAMING_SNAKE_CASE__											=			-1 * (i + 1)
            l[reversed_idx] &= tally
            SCREAMING_SNAKE_CASE__											=			l[reversed_idx]

    if start_edges is None:
        SCREAMING_SNAKE_CASE__											=			[s == 0 for s in start]
        reduce_edge_list(lowerCAmelCase_	)
    if end_edges is None:
        SCREAMING_SNAKE_CASE__											=			[e == (d - 1) for e, d in zip(lowerCAmelCase_ ,							lowerCAmelCase_	)]
        reduce_edge_list(lowerCAmelCase_	)

    # Base cases. Either start/end are empty and we're done, or the final,
    # one-dimensional tensor can be simply sliced
    if len(lowerCAmelCase_	) == 0:
        return [()]
    elif len(lowerCAmelCase_	) == 1:
        return [(slice(start[0] ,							end[0] + 1	),)]

    SCREAMING_SNAKE_CASE__											=			[]
    SCREAMING_SNAKE_CASE__											=			[]

    # Dimensions common to start and end can be selected directly
    for s, e in zip(lowerCAmelCase_ ,							lowerCAmelCase_	):
        if s == e:
            path_list.append(slice(lowerCAmelCase_ ,							s + 1	)	)
        else:
            break

    SCREAMING_SNAKE_CASE__											=			tuple(lowerCAmelCase_	)
    SCREAMING_SNAKE_CASE__											=			len(lowerCAmelCase_	)

    # start == end, and we're done
    if divergence_idx == len(lowerCAmelCase_	):
        return [path]

    def upper() -> Tuple[Tuple[slice, ...], ...]:
        assert start_edges is not None
        assert end_edges is not None

        SCREAMING_SNAKE_CASE__											=			start[divergence_idx]
        return tuple(
            path + (slice(lowerCAmelCase_ ,							sdi + 1	),) + s
            for s in _get_minimal_slice_set(
                start[divergence_idx + 1 :] ,							[d - 1 for d in dims[divergence_idx + 1 :]] ,							dims[divergence_idx + 1 :] ,							start_edges=start_edges[divergence_idx + 1 :] ,							end_edges=[True for _ in end_edges[divergence_idx + 1 :]] ,							)	)

    def lower() -> Tuple[Tuple[slice, ...], ...]:
        assert start_edges is not None
        assert end_edges is not None

        SCREAMING_SNAKE_CASE__											=			end[divergence_idx]
        return tuple(
            path + (slice(lowerCAmelCase_ ,							edi + 1	),) + s
            for s in _get_minimal_slice_set(
                [0 for _ in start[divergence_idx + 1 :]] ,							end[divergence_idx + 1 :] ,							dims[divergence_idx + 1 :] ,							start_edges=[True for _ in start_edges[divergence_idx + 1 :]] ,							end_edges=end_edges[divergence_idx + 1 :] ,							)	)

    # If both start and end are at the edges of the subtree rooted at
    # divergence_idx, we can just select the whole subtree at once
    if start_edges[divergence_idx] and end_edges[divergence_idx]:
        slices.append(path + (slice(start[divergence_idx] ,							end[divergence_idx] + 1	),)	)
    # If just start is at the edge, we can grab almost all of the subtree,
    # treating only the ragged bottom edge as an edge case
    elif start_edges[divergence_idx]:
        slices.append(path + (slice(start[divergence_idx] ,							end[divergence_idx]	),)	)
        slices.extend(lower()	)
    # Analogous to the previous case, but the top is ragged this time
    elif end_edges[divergence_idx]:
        slices.extend(upper()	)
        slices.append(path + (slice(start[divergence_idx] + 1 ,							end[divergence_idx] + 1	),)	)
    # If both sides of the range are ragged, we need to handle both sides
    # separately. If there's contiguous meat in between them, we can index it
    # in one big chunk
    else:
        slices.extend(upper()	)
        SCREAMING_SNAKE_CASE__											=			end[divergence_idx] - start[divergence_idx]
        if middle_ground > 1:
            slices.append(path + (slice(start[divergence_idx] + 1 ,							end[divergence_idx]	),)	)
        slices.extend(lower()	)

    return slices


@torch.jit.ignore
def 							__snake_case							(    lowerCAmelCase_ ,							lowerCAmelCase_ ,							lowerCAmelCase_ ,							lowerCAmelCase_	)				->						torch.Tensor:
    SCREAMING_SNAKE_CASE__											=			t.shape[:no_batch_dims]
    SCREAMING_SNAKE_CASE__											=			list(_flat_idx_to_idx(lowerCAmelCase_ ,							lowerCAmelCase_	)	)
    # _get_minimal_slice_set is inclusive
    SCREAMING_SNAKE_CASE__											=			list(_flat_idx_to_idx(flat_end - 1 ,							lowerCAmelCase_	)	)

    # Get an ordered list of slices to perform
    SCREAMING_SNAKE_CASE__											=			_get_minimal_slice_set(
        lowerCAmelCase_ ,							lowerCAmelCase_ ,							lowerCAmelCase_ ,							)

    SCREAMING_SNAKE_CASE__											=			[t[s] for s in slices]

    return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]	) for s in sliced_tensors]	)


def 							__snake_case							(    lowerCAmelCase_ ,							lowerCAmelCase_ ,							lowerCAmelCase_ ,							lowerCAmelCase_ ,							lowerCAmelCase_ = False ,							lowerCAmelCase_ = None ,							lowerCAmelCase_ = False ,							)				->						Any:
    if not (len(lowerCAmelCase_	) > 0):
        raise ValueError('''Must provide at least one input'''	)

    SCREAMING_SNAKE_CASE__											=			[shape[:no_batch_dims] for shape in _fetch_dims(lowerCAmelCase_	)]
    SCREAMING_SNAKE_CASE__											=			tuple([max(lowerCAmelCase_	) for s in zip(*lowerCAmelCase_	)]	)

    def _prep_inputs(lowerCAmelCase_	) -> torch.Tensor:
        if not low_mem:
            if not sum(t.shape[:no_batch_dims]	) == no_batch_dims:
                SCREAMING_SNAKE_CASE__											=			t.expand(orig_batch_dims + t.shape[no_batch_dims:]	)
            SCREAMING_SNAKE_CASE__											=			t.reshape(-1 ,							*t.shape[no_batch_dims:]	)
        else:
            SCREAMING_SNAKE_CASE__											=			t.expand(orig_batch_dims + t.shape[no_batch_dims:]	)
        return t

    SCREAMING_SNAKE_CASE__											=			tensor_tree_map(_prep_inputs ,							lowerCAmelCase_	)
    SCREAMING_SNAKE_CASE__											=			None
    if _out is not None:
        SCREAMING_SNAKE_CASE__											=			tensor_tree_map(lambda lowerCAmelCase_	: t.view([-1] + list(t.shape[no_batch_dims:]	)	) ,							_out	)

    SCREAMING_SNAKE_CASE__											=			1
    for d in orig_batch_dims:
        flat_batch_dim *= d

    SCREAMING_SNAKE_CASE__											=			flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

    def _select_chunk(lowerCAmelCase_	) -> torch.Tensor:
        return t[i : i + chunk_size] if t.shape[0] != 1 else t

    SCREAMING_SNAKE_CASE__											=			0
    SCREAMING_SNAKE_CASE__											=			prepped_outputs
    for _ in range(lowerCAmelCase_	):
        # Chunk the input
        if not low_mem:
            SCREAMING_SNAKE_CASE__											=			_select_chunk
        else:
            SCREAMING_SNAKE_CASE__											=			partial(
                _chunk_slice ,							flat_start=lowerCAmelCase_ ,							flat_end=min(lowerCAmelCase_ ,							i + chunk_size	) ,							no_batch_dims=len(lowerCAmelCase_	) ,							)

        SCREAMING_SNAKE_CASE__											=			tensor_tree_map(lowerCAmelCase_ ,							lowerCAmelCase_	)

        # Run the layer on the chunk
        SCREAMING_SNAKE_CASE__											=			layer(**lowerCAmelCase_	)

        # Allocate space for the output
        if out is None:
            SCREAMING_SNAKE_CASE__											=			tensor_tree_map(lambda lowerCAmelCase_	: t.new_zeros((flat_batch_dim,) + t.shape[1:]	) ,							lowerCAmelCase_	)

        # Put the chunk in its pre-allocated space
        if isinstance(lowerCAmelCase_ ,							lowerCAmelCase_	):

            def assign(lowerCAmelCase_ ,							lowerCAmelCase_	) -> None:
                for k, v in da.items():
                    if isinstance(lowerCAmelCase_ ,							lowerCAmelCase_	):
                        assign(lowerCAmelCase_ ,							da[k]	)
                    else:
                        if _add_into_out:
                            v[i : i + chunk_size] += da[k]
                        else:
                            SCREAMING_SNAKE_CASE__											=			da[k]

            assign(lowerCAmelCase_ ,							lowerCAmelCase_	)
        elif isinstance(lowerCAmelCase_ ,							lowerCAmelCase_	):
            for xa, xa in zip(lowerCAmelCase_ ,							lowerCAmelCase_	):
                if _add_into_out:
                    xa[i : i + chunk_size] += xa
                else:
                    SCREAMING_SNAKE_CASE__											=			xa
        elif isinstance(lowerCAmelCase_ ,							torch.Tensor	):
            if _add_into_out:
                out[i : i + chunk_size] += output_chunk
            else:
                SCREAMING_SNAKE_CASE__											=			output_chunk
        else:
            raise ValueError('''Not supported'''	)

        i += chunk_size

    SCREAMING_SNAKE_CASE__											=			tensor_tree_map(lambda lowerCAmelCase_	: t.view(orig_batch_dims + t.shape[1:]	) ,							lowerCAmelCase_	)

    return out



class 					__snake_case	:


    '''simple docstring'''







    def __init__(       self   ,				A_ = 5_12   ,				):





        '''simple docstring'''

        SCREAMING_SNAKE_CASE__											=			max_chunk_size
        SCREAMING_SNAKE_CASE__											=			None
        SCREAMING_SNAKE_CASE__											=			None



    def  lowercase_					(       self   ,				A_   ,				A_   ,				A_     ):





        '''simple docstring'''

        logging.info('''Tuning chunk size...'''     )

        if min_chunk_size >= self.max_chunk_size:
            return min_chunk_size

        SCREAMING_SNAKE_CASE__											=			[2**l for l in range(int(math.log(self.max_chunk_size   ,				2     )     ) + 1     )]
        SCREAMING_SNAKE_CASE__											=			[c for c in candidates if c > min_chunk_size]
        SCREAMING_SNAKE_CASE__											=			[min_chunk_size] + candidates
        candidates[-1] += 4

        def test_chunk_size(A_     ) -> bool:
            try:
                with torch.no_grad():
                    fn(*A_   ,				chunk_size=A_     )
                return True
            except RuntimeError:
                return False

        SCREAMING_SNAKE_CASE__											=			0
        SCREAMING_SNAKE_CASE__											=			len(A_     ) - 1
        while i > min_viable_chunk_size_index:
            SCREAMING_SNAKE_CASE__											=			test_chunk_size(candidates[i]     )
            if not viable:
                SCREAMING_SNAKE_CASE__											=			(min_viable_chunk_size_index + i) // 2
            else:
                SCREAMING_SNAKE_CASE__											=			i
                SCREAMING_SNAKE_CASE__											=			(i + len(A_     ) - 1) // 2

        return candidates[min_viable_chunk_size_index]



    def  lowercase_					(       self   ,				A_   ,				A_     ):





        '''simple docstring'''

        SCREAMING_SNAKE_CASE__											=			True
        for aa, aa in zip(A_   ,				A_     ):
            assert type(A_     ) == type(A_     )
            if isinstance(A_   ,				(list, tuple)     ):
                consistent &= self._compare_arg_caches(A_   ,				A_     )
            elif isinstance(A_   ,				A_     ):
                SCREAMING_SNAKE_CASE__											=			[v for _, v in sorted(aa.items()   ,				key=lambda A_     : x[0]     )]
                SCREAMING_SNAKE_CASE__											=			[v for _, v in sorted(aa.items()   ,				key=lambda A_     : x[0]     )]
                consistent &= self._compare_arg_caches(A_   ,				A_     )
            else:
                consistent &= aa == aa

        return consistent




    def  lowercase_					(       self   ,				A_   ,				A_   ,				A_   ,				):





        '''simple docstring'''

        SCREAMING_SNAKE_CASE__											=			True
        SCREAMING_SNAKE_CASE__											=			tree_map(lambda A_     : a.shape if isinstance(A_   ,				torch.Tensor     ) else a   ,				A_   ,				A_     )
        if self.cached_arg_data is not None:
            # If args have changed shape/value, we need to re-tune
            assert len(self.cached_arg_data     ) == len(A_     )
            SCREAMING_SNAKE_CASE__											=			self._compare_arg_caches(self.cached_arg_data   ,				A_     )
        else:
            # Otherwise, we can reuse the precomputed value
            SCREAMING_SNAKE_CASE__											=			False

        if not consistent:
            SCREAMING_SNAKE_CASE__											=			self._determine_favorable_chunk_size(
                A_   ,				A_   ,				A_   ,				)
            SCREAMING_SNAKE_CASE__											=			arg_data

        assert self.cached_chunk_size is not None

        return self.cached_chunk_size








from jiwer import compute_measures

import datasets


A__      :     Tuple       =					'\\n@inproceedings{inproceedings,\n    author = {Morris, Andrew and Maier, Viktoria and Green, Phil},\n    year = {2004},\n    month = {01},\n    pages = {},\n    title = {From WER and RIL to MER and WIL: improved evaluation measures for connected speech recognition.}\n}\n'

A__      :     Optional[int]       =					'\\nWord error rate (WER) is a common metric of the performance of an automatic speech recognition system.\n\nThe general difficulty of measuring performance lies in the fact that the recognized word sequence can have a different length from the reference word sequence (supposedly the correct one). The WER is derived from the Levenshtein distance, working at the word level instead of the phoneme level. The WER is a valuable tool for comparing different systems as well as for evaluating improvements within one system. This kind of measurement, however, provides no details on the nature of translation errors and further work is therefore required to identify the main source(s) of error and to focus any research effort.\n\nThis problem is solved by first aligning the recognized word sequence with the reference (spoken) word sequence using dynamic string alignment. Examination of this issue is seen through a theory called the power law that states the correlation between perplexity and word error rate.\n\nWord error rate can then be computed as:\n\nWER = (S + D + I) / N = (S + D + I) / (S + D + C)\n\nwhere\n\nS is the number of substitutions,\nD is the number of deletions,\nI is the number of insertions,\nC is the number of correct words,\nN is the number of words in the reference (N=S+D+C).\n\nThis value indicates the average number of errors per reference word. The lower the value, the better the\nperformance of the ASR system with a WER of 0 being a perfect score.\n'

A__      :     Dict       =					'\nCompute WER score of transcribed segments against references.\n\nArgs:\n    references: List of references for each speech input.\n    predictions: List of transcriptions to score.\n    concatenate_texts (bool, default=False): Whether to concatenate all input texts or compute WER iteratively.\n\nReturns:\n    (float): the word error rate\n\nExamples:\n\n    >>> predictions = ["this is the prediction", "there is an other sample"]\n    >>> references = ["this is the reference", "there is another one"]\n    >>> wer = datasets.load_metric("wer")\n    >>> wer_score = wer.compute(predictions=predictions, references=references)\n    >>> print(wer_score)\n    0.5\n'
@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION						,_KWARGS_DESCRIPTION      )
class  _UpperCAmelCase   (			datasets.Metric      ):

				"""simple docstring"""




				def 							lowercase__  (	self		:       Union[str, Any]    ):

								'''simple docstring'''
								return datasets.MetricInfo(
								    description=_DESCRIPTION,    citation=_CITATION,    inputs_description=_KWARGS_DESCRIPTION,    features=datasets.Features(
								        {
								            '''predictions''': datasets.Value('''string''',    id='''sequence'''    ),
								            '''references''': datasets.Value('''string''',    id='''sequence'''    ),
								        }    ),    codebase_urls=['''https://github.com/jitsi/jiwer/'''],    reference_urls=[
								        '''https://en.wikipedia.org/wiki/Word_error_rate''',
								    ],    )




				def 							lowercase__  (	self		:       Union[str, Any],    lowerCamelCase		:       str=None,    lowerCamelCase		:       Union[str, Any]=None,    lowerCamelCase		:       List[str]=False    ):

								'''simple docstring'''
								if concatenate_texts:
												return compute_measures(lowerCamelCase,    lowerCamelCase    )["wer"]
								else:
												lowercase__  					=     0
												lowercase__  					=     0
												for prediction, reference in zip(lowerCamelCase,    lowerCamelCase    ):
																lowercase__  					=     compute_measures(lowerCamelCase,    lowerCamelCase    )
																incorrect += measures["substitutions"] + measures["deletions"] + measures["insertions"]
																total += measures["substitutions"] + measures["deletions"] + measures["hits"]
												return incorrect / total




