





import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
 from .features import FeatureType


_lowerCAmelCase       ,       _lowerCAmelCase       ,       _lowerCAmelCase				:   Optional[Any]						       =						False, False, False


@dataclass
class        __magic_name__	:





     """simple docstring"""



     __UpperCamelCase            =  None
     __UpperCamelCase            =  True
     __UpperCamelCase            =  True
     __UpperCamelCase            =  None
     # Automatically constructed
     __UpperCamelCase            =  "dict"
     __UpperCamelCase            =  pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}				)
     __UpperCamelCase            =  field(default='''Audio'''     ,					init=lowerCamelCase__     ,					repr=lowerCamelCase__				)



     def __call__(				self							:Dict	):





        '''simple docstring'''




        return self.pa_type



     def     SCREAMING_SNAKE_CASE		(				self							:List[Any]      ,   snake_case							:Union[str, bytes, dict]	):





        '''simple docstring'''




        try:
           import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
        except ImportError as err:
           raise ImportError("To support encoding audio data, please install 'soundfile'."	) from err
        if isinstance(snake_case      ,   snake_case	):
           return {"bytes": None, "path": value}
        elif isinstance(snake_case      ,   snake_case	):
           return {"bytes": value, "path": None}
        elif "array" in value:
           # convert the audio array to wav bytes
           A_       :					List[str]        =      BytesIO()
           sf.write(snake_case      ,   value["array"]      ,   value["sampling_rate"]      ,   format="wav"	)
           return {"bytes": buffer.getvalue(), "path": None}
        elif value.get("path"	) is not None and os.path.isfile(value["path"]	):
           # we set "bytes": None to not duplicate the data if they're already available locally
           if value["path"].endswith("pcm"	):
              # "PCM" only has raw audio bytes
              if value.get("sampling_rate"	) is None:
                 # At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
                 raise KeyError("To use PCM files, please specify a 'sampling_rate' in Audio object"	)
              if value.get("bytes"	):
                 # If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
                 A_       :					str        =      np.frombuffer(value["bytes"]      ,   dtype=np.intaa	).astype(np.floataa	) / 32_767
              else:
                 A_       :					List[str]        =      np.memmap(value["path"]      ,   dtype="h"      ,   mode="r"	).astype(np.floataa	) / 32_767

              A_       :					List[Any]        =      BytesIO(bytes()	)
              sf.write(snake_case      ,   snake_case      ,   value["sampling_rate"]      ,   format="wav"	)
              return {"bytes": buffer.getvalue(), "path": None}
           else:
              return {"bytes": None, "path": value.get("path"	)}
        elif value.get("bytes"	) is not None or value.get("path"	) is not None:
           # store the audio bytes, and path is used to infer the audio format using the file extension
           return {"bytes": value.get("bytes"	), "path": value.get("path"	)}
        else:
           raise ValueError(
               f"An audio sample should have one of 'path' or 'bytes' but they are missing or None in {value}."	)



     def     SCREAMING_SNAKE_CASE		(				self							:Optional[Any]      ,   snake_case							:dict      ,   snake_case							:Optional[Dict[str, Union[str, bool, None]]] = None	):





        '''simple docstring'''




        if not self.decode:
           raise RuntimeError("Decoding is disabled for this feature. Please use Audio(decode=True) instead."	)

        A_   ,    A_       :					Optional[int]        =      (value["path"], BytesIO(value["bytes"]	)) if value["bytes"] is not None else (value["path"], None)
        if path is None and file is None:
           raise ValueError(f"An audio sample should have one of 'path' or 'bytes' but both are None in {value}."	)

        try:
           import librosa
           import soundfile as sf
        except ImportError as err:
           raise ImportError("To support decoding audio files, please install 'librosa' and 'soundfile'."	) from err

        A_       :					Dict        =      xsplitext(snake_case	)[1][1:].lower() if path is not None else None
        if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
           raise RuntimeError(
               "Decoding 'opus' files requires system library 'libsndfile'>=1.0.31, "
               "You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. "	)
        elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
           raise RuntimeError(
               "Decoding 'mp3' files requires system library 'libsndfile'>=1.1.0, "
               "You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. "	)

        if file is None:
           A_       :					str        =      token_per_repo_id or {}
           A_       :					Optional[int]        =      path.split("::"	)[-1]
           try:
              A_       :					int        =      string_to_dict(snake_case      ,   config.HUB_DATASETS_URL	)["repo_id"]
              A_       :					int        =      token_per_repo_id[repo_id]
           except (ValueError, KeyError):
              A_       :					int        =      None

           with xopen(snake_case      ,   "rb"      ,   use_auth_token=snake_case	) as f:
              A_   ,    A_       :					Dict        =      sf.read(snake_case	)

        else:
           A_   ,    A_       :					List[Any]        =      sf.read(snake_case	)

        A_       :					Dict        =      array.T
        if self.mono:
           A_       :					str        =      librosa.to_mono(snake_case	)
        if self.sampling_rate and self.sampling_rate != sampling_rate:
           A_       :					Union[str, Any]        =      librosa.resample(snake_case      ,   orig_sr=snake_case      ,   target_sr=self.sampling_rate	)
           A_       :					str        =      self.sampling_rate

        return {"path": path, "array": array, "sampling_rate": sampling_rate}



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





        '''simple docstring'''




        from .features import Value

        if self.decode:
           raise ValueError("Cannot flatten a decoded Audio feature."	)
        return {
            "bytes": Value("binary"	),
            "path": Value("string"	),
        }



     def     SCREAMING_SNAKE_CASE		(				self							:str      ,   snake_case							:Union[pa.StringArray, pa.StructArray]	):





        '''simple docstring'''




        if pa.types.is_string(storage.type	):
           A_       :					List[str]        =      pa.array([None] * len(snake_case	)      ,   type=pa.binary()	)
           A_       :					Optional[Any]        =      pa.StructArray.from_arrays([bytes_array, storage]      ,   ["bytes", "path"]      ,   mask=storage.is_null()	)
        elif pa.types.is_binary(storage.type	):
           A_       :					Any        =      pa.array([None] * len(snake_case	)      ,   type=pa.string()	)
           A_       :					Optional[int]        =      pa.StructArray.from_arrays([storage, path_array]      ,   ["bytes", "path"]      ,   mask=storage.is_null()	)
        elif pa.types.is_struct(storage.type	) and storage.type.get_all_field_indices("array"	):
           A_       :					Union[str, Any]        =      pa.array([Audio().encode_example(snake_case	) if x is not None else None for x in storage.to_pylist()]	)
        elif pa.types.is_struct(storage.type	):
           if storage.type.get_field_index("bytes"	) >= 0:
              A_       :					List[str]        =      storage.field("bytes"	)
           else:
              A_       :					str        =      pa.array([None] * len(snake_case	)      ,   type=pa.binary()	)
           if storage.type.get_field_index("path"	) >= 0:
              A_       :					List[str]        =      storage.field("path"	)
           else:
              A_       :					List[Any]        =      pa.array([None] * len(snake_case	)      ,   type=pa.string()	)
           A_       :					Any        =      pa.StructArray.from_arrays([bytes_array, path_array]      ,   ["bytes", "path"]      ,   mask=storage.is_null()	)
        return array_cast(snake_case      ,   self.pa_type	)





     def     SCREAMING_SNAKE_CASE		(				self							:List[str]      ,   snake_case							:pa.StructArray	):





        '''simple docstring'''




        @no_op_if_value_is_null
        def path_to_bytes(snake_case							:List[str]	):
           with xopen(snake_case      ,   "rb"	) as f:
              A_       :					Union[str, Any]        =      f.read()
           return bytes_

        A_       :					Union[str, Any]        =      pa.array(
            [
                (path_to_bytes(x["path"]	) if x["bytes"] is None else x["bytes"]) if x is not None else None
                for x in storage.to_pylist()
            ]      ,   type=pa.binary()      ,   )
        A_       :					Any        =      pa.array(
            [os.path.basename(snake_case	) if path is not None else None for path in storage.field("path"	).to_pylist()]      ,   type=pa.string()      ,   )
        A_       :					int        =      pa.StructArray.from_arrays([bytes_array, path_array]      ,   ["bytes", "path"]      ,   mask=bytes_array.is_null()	)
        return array_cast(snake_case      ,   self.pa_type	)







from collections import defaultdict


class        __magic_name__	:





     """simple docstring"""



     def __init__(				self							:int      ,   snake_case							:Dict      ,   snake_case							:Optional[Any]	):





        '''simple docstring'''




        A_       :					Optional[int]        =      total  # total no of tasks (N)

        # DP table will have a dimension of (2^M)*N
        # initially all values are set to -1
        A_       :					str        =      [
            [-1 for i in range(total + 1	)] for j in range(2 ** len(snake_case	)	)
        ]

        A_       :					Optional[int]        =      defaultdict(snake_case	)  # stores the list of persons for each task

        # final_mask is used to check if all persons are included by setting all bits
        # to 1
        A_       :					Dict        =      (1 << len(snake_case	)) - 1



     def     SCREAMING_SNAKE_CASE		(				self							:Union[str, Any]      ,   snake_case							:Union[str, Any]      ,   snake_case							:str	):





        '''simple docstring'''




        if mask == self.final_mask:
           return 1

        # if not everyone gets the task and no more tasks are available, return 0
        if task_no > self.total_tasks:
           return 0

        # if case already considered
        if self.dp[mask][task_no] != -1:
           return self.dp[mask][task_no]

        # Number of ways when we don't this task in the arrangement
        A_       :					Tuple        =      self.count_ways_until(snake_case      ,   task_no + 1	)

        # now assign the tasks one by one to all possible persons and recursively
        # assign for the remaining tasks.
        if task_no in self.task:
           for p in self.task[task_no]:
              # if p is already given a task
              if mask & (1 << p):
                 continue

              # assign this task to p and change the mask value. And recursively
              # assign tasks with the new mask value.
              total_ways_util += self.count_ways_until(mask | (1 << p)      ,   task_no + 1	)

        # save the value.
        A_       :					Any        =      total_ways_util

        return self.dp[mask][task_no]





     def     SCREAMING_SNAKE_CASE		(				self							:int      ,   snake_case							:List[str]	):





        '''simple docstring'''




        for i in range(len(snake_case	)	):
           for j in task_performed[i]:
              self.task[j].append(snake_case	)

        # call the function to fill the DP table, final answer is stored in dp[0][1]
        return self.count_ways_until(0      ,   1	)


if __name__ == "__main__":
 _lowerCAmelCase				:   Dict						       =						5  # total no of tasks (the value of N)

 # the list of tasks that can be done by M persons.
 _lowerCAmelCase				:   str						       =						[[1, 3, 4], [1, 2, 5], [3, 4]]
 print(
     AssignmentUsingBitmask(task_performed, total_tasks).count_no_of_ways(
         task_performed
     )
 )
