






"""simple docstring"""





import argparse
import os
from functools import reduce

import fairseq
import torch
from datasets import load_dataset

from transformers import WavaVecaProcessor, logging
from transformers.models.dataavec.configuration_dataavec_audio import DataaVecAudioConfig

# Copied from https://github.com/pytorch/fairseq/blob/main/examples/data2vec/models/data2vec_audio.py
from transformers.models.dataavec.dataavec_audio import DataaVecAudioModel as Dummy  # noqa: F401
from transformers.models.dataavec.modeling_dataavec_audio import DataaVecAudioForCTC, DataaVecAudioModel


logging.set_verbosity_info()
UpperCamelCase_						:					Optional[int]      				=			logging.get_logger(__name__)

UpperCamelCase_						:					Optional[Any]      				=			{
    """post_extract_proj""": """feature_projection.projection""",
    """models.0.layer_norm""": """feature_projection.layer_norm""",
    """self_attn.k_proj""": """encoder.layers.*.attention.k_proj""",
    """self_attn.v_proj""": """encoder.layers.*.attention.v_proj""",
    """self_attn.q_proj""": """encoder.layers.*.attention.q_proj""",
    """self_attn.out_proj""": """encoder.layers.*.attention.out_proj""",
    """self_attn_layer_norm""": """encoder.layers.*.layer_norm""",
    """fc1""": """encoder.layers.*.feed_forward.intermediate_dense""",
    """fc2""": """encoder.layers.*.feed_forward.output_dense""",
    """final_layer_norm""": """encoder.layers.*.final_layer_norm""",
    """encoder.layer_norm""": """encoder.layer_norm""",
    """w2v_model.layer_norm""": """feature_projection.layer_norm""",
    """w2v_encoder.proj""": """lm_head""",
    """mask_emb""": """masked_spec_embed""",
}
UpperCamelCase_						:					List[Any]      				=			[
    """lm_head""",
]



def      __lowercase       (     a			:					Dict							,		a			:					Optional[Any]							,		a			:					List[str]							,		a			:					Dict							,		a			:					Optional[Any]			)       ->       List[str]:
      for attribute in key.split('''.'''			):
            __snake_case   :     Optional[int]	      =getattr(a							,		a			)

      if weight_type is not None:
            __snake_case   :     str	      =getattr(a							,		a			).shape
      else:
            __snake_case   :     List[str]	      =hf_pointer.shape

      if hf_shape != value.shape:
            raise ValueError(
                f'''Shape of hf {key + "." + weight_type if weight_type is not None else ""} is {hf_shape}, but should be'''
                f''' {value.shape} for {full_name}'''			)

      if weight_type == "weight":
            __snake_case   :     Optional[Any]	      =value
      elif weight_type == "weight_g":
            __snake_case   :     Any	      =value
      elif weight_type == "weight_v":
            __snake_case   :     Dict	      =value
      elif weight_type == "bias":
            __snake_case   :     Optional[int]	      =value
      else:
            __snake_case   :     Optional[Any]	      =value

      logger.info(f'''{key + "." + weight_type if weight_type is not None else ""} was initialized from {full_name}.'''			)



def      __lowercase       (     a			:					int							,		a			:					Optional[int]							,		a			:					List[str]			)       ->       str:
      __snake_case   :     Optional[Any]	      =[]
      __snake_case   :     Optional[Any]	      =fairseq_model.state_dict()

      if not is_headless:
            __snake_case   :     List[Any]	      =hf_model.dataavec_audio.feature_extractor
            __snake_case   :     List[str]	      =hf_model.dataavec_audio.encoder.pos_conv_embed

      else:
            __snake_case   :     Dict	      =hf_model.feature_extractor
            __snake_case   :     Tuple	      =hf_model.encoder.pos_conv_embed

      for name, value in fairseq_dict.items():
            __snake_case   :     Union[str, Any]	      =False
            if "conv_layers" in name:
                  load_conv_layer(
                      a							,		a							,		a							,		a							,		)
                  __snake_case   :     List[Any]	      =True
            elif "pos_conv" in name:
                  load_pos_conv_layer(
                      a							,		a							,		a							,		a							,		)
                  __snake_case   :     Optional[Any]	      =True
            else:
                  for key, mapped_key in MAPPING.items():
                        if not is_headless:
                              __snake_case   :     Tuple	      ='''data2vec_audio.''' + mapped_key if mapped_key not in TOP_LEVEL_KEYS else mapped_key
                        if key in name or key.split('''w2v_model.'''			)[-1] == name.split('''.'''			)[0]:
                              __snake_case   :     Dict	      =True
                              if "*" in mapped_key:
                                    __snake_case   :     str	      =name.split(a			)[0].split('''.'''			)[-2]
                                    __snake_case   :     Any	      =mapped_key.replace('''*'''							,		a			)
                              if "weight_g" in name:
                                    __snake_case   :     Any	      ='''weight_g'''
                              elif "weight_v" in name:
                                    __snake_case   :     Optional[Any]	      ='''weight_v'''
                              elif "bias" in name:
                                    __snake_case   :     Union[str, Any]	      ='''bias'''
                              elif "weight" in name:
                                    # TODO: don't match quantizer.weight_proj
                                    __snake_case   :     Union[str, Any]	      ='''weight'''
                              else:
                                    __snake_case   :     Union[str, Any]	      =None
                              set_recursively(a							,		a							,		a							,		a							,		a			)
                        continue
            if not is_used:
                  unused_weights.append(a			)

      logger.warning(f'''Unused weights: {unused_weights}'''			)



def      __lowercase       (     a			:					List[str]							,		a			:					Optional[Any]			)       ->       Dict:
      __snake_case   :     Optional[int]	      =path.split('''.'''			)
      return reduce(a							,		a							,		a			)



def      __lowercase       (     a			:					Optional[int]							,		a			:					List[Any]							,		a			:					Tuple							,		a			:					List[str]			)       ->       Optional[Any]:
      __snake_case   :     Optional[Any]	      =access_by_string(a							,		a			)
      __snake_case   :     Any	      =hf_weight.data

      if fsq_value.shape != hf_value.shape:
            raise ValueError(f'''{full_name} has size {fsq_value.shape}, but {hf_value.shape} was found.'''			)
      __snake_case   :     Optional[Any]	      =fsq_value
      logger.info(f'''{full_name} was correctly initialized from {hf_weight_path}.'''			)



def      __lowercase       (     a			:					int							,		a			:					Optional[Any]							,		a			:					Union[str, Any]							,		a			:					Tuple			)       ->       int:
      __snake_case   :     int	      =full_name.split('''conv_layers.'''			)[-1]
      __snake_case   :     Optional[Any]	      =name.split('''.'''			)
      __snake_case   :     List[str]	      =int(items[0]			)
      __snake_case   :     List[str]	      =int(items[1]			)

      __snake_case   :     Dict	      =name.split('''.'''			)[-1]
      if type_id == 0:
            __snake_case   :     Any	      ='''conv'''
      elif type_id == 2:
            __snake_case   :     str	      ='''layer_norm'''
      else:
            unused_weights.append(a			)
            return

      set_weights(a							,		a							,		a							,		f'''conv_layers.{layer_id}.{layer_type}.{weight_type}'''			)



def      __lowercase       (     a			:					Optional[int]							,		a			:					List[Any]							,		a			:					int							,		a			:					Dict			)       ->       Optional[Any]:
      __snake_case   :     List[str]	      =full_name.split('''pos_conv.'''			)[-1]
      __snake_case   :     List[Any]	      =name.split('''.'''			)
      __snake_case   :     Tuple	      =int(items[0]			)
      __snake_case   :     Any	      =int(items[1]			)

      __snake_case   :     Tuple	      =name.split('''.'''			)[-1]
      if type_id != 0:
            unused_weights.append(a			)
            return
      else:
            __snake_case   :     Tuple	      ='''conv'''

      set_weights(a							,		a							,		a							,		f'''layers.{layer_id}.{layer_type}.{weight_type}'''			)



@torch.no_grad()
def      __lowercase       (     a			:					List[Any]							,		a			:					str							,		a			:					str=None							,		a			:					Optional[int]=None							,		a			:					List[Any]=True			)       ->       str:
      if config_path is not None:
            __snake_case   :     List[Any]	      =DataaVecAudioConfig.from_pretrained(a			)
      else:
            __snake_case   :     Any	      =DataaVecAudioConfig()

      if not is_finetuned:
            # Modify final_proj layer name
            __snake_case   :     Optional[int]	      =DataaVecAudioModel(a			)
            __snake_case   :     List[str]	      =os.path.dirname(a			)

            __snake_case   :     Optional[int]	      =torch.load(a			)
            __snake_case   :     List[Any]	      =state_dict['''model'''].pop('''final_proj.0.weight'''			)
            __snake_case   :     str	      =state_dict['''model'''].pop('''final_proj.0.bias'''			)
            __snake_case   :     Union[str, Any]	      =os.path.join(a							,		'''converted.pt'''			)
            torch.save(a							,		a			)
      else:
            __snake_case   :     Optional[Any]	      =DataaVecAudioForCTC(a			)
            __snake_case   :     Union[str, Any]	      =checkpoint_path

      def load_dataavec(a			:					Union[str, Any]			):
            __snake_case      ,					__snake_case      ,					__snake_case   :     str	      =fairseq.checkpoint_utils.load_model_ensemble_and_task([path]			)
            return model[0].eval()

      __snake_case   :     str	      =load_dataavec(a			)

      recursively_load_weights(a							,		a							,		not is_finetuned			)

      __snake_case   :     Optional[int]	      =WavaVecaProcessor.from_pretrained('''facebook/wav2vec2-large-lv60'''			)

      __snake_case   :     Optional[int]	      =load_dataset('''patrickvonplaten/librispeech_asr_dummy'''							,		'''clean'''							,		split='''validation'''			)
      __snake_case   :     Optional[int]	      =[x['''array'''] for x in ds[:4]['''audio''']]

      __snake_case   :     Optional[Any]	      =processor(a							,		return_tensors='''pt'''							,		padding=a			)

      __snake_case   :     Union[str, Any]	      =inputs.input_values
      __snake_case   :     Optional[Any]	      =inputs.attention_mask
      #    input_values = inputs.input_values[:, :-1]
      #    attention_mask = inputs.attention_mask[:, :-1]

      hf_wavavec.eval()
      model.eval()
      if is_finetuned:
            __snake_case   :     List[Any]	      =model(source=a							,		padding_mask=(1 - attention_mask)							,		mask=a							,		features_only=a			)[
                '''encoder_out'''
            ].transpose(0							,		1			)
            __snake_case   :     str	      =hf_wavavec(a							,		attention_mask=a			)['''logits''']

            __snake_case   :     Optional[int]	      =torch.argmax(a							,		dim=-1			)
            __snake_case   :     Optional[int]	      =processor.batch_decode(a			)

            print(f'''Expected Output: {ds[:4]["text"]}, Pred: {output_string}'''			)
      else:
            __snake_case   :     Dict	      =model(source=a							,		padding_mask=(1 - attention_mask)							,		mask=a							,		features_only=a			)[
                '''layer_results'''
            ][-1][0].transpose(0							,		1			)
            __snake_case   :     Tuple	      =hf_wavavec(a							,		attention_mask=a			)['''last_hidden_state''']

      print(our_output.shape							,		their_output.shape			)
      __snake_case   :     Dict	      =torch.max(torch.abs(our_output - their_output			)			).item()
      print(f'''max_absolute_diff = {max_absolute_diff}'''			)  # ~ 1e-7
      __snake_case   :     Dict	      =torch.allclose(a							,		a							,		atol=1e-3			)
      print('''Do both models output the same tensors?'''							,		'''🔥''' if success else '''💩'''			)
      if not success:
            raise Exception('''Something went wRoNg'''			)

      hf_wavavec.save_pretrained(a			)

      if is_finetuned:
            processor.save_pretrained(a			)
      else:
            processor.feature_extractor.save_pretrained(a			)


if __name__ == "__main__":
    UpperCamelCase_						:					Tuple      				=			argparse.ArgumentParser()
    parser.add_argument("""--pytorch_dump_folder_path""", default=None, type=str, help="""Path to the output PyTorch model.""")
    parser.add_argument("""--checkpoint_path""", default=None, type=str, help="""Path to fairseq checkpoint""")
    parser.add_argument("""--dict_path""", default=None, type=str, help="""Path to dict of fine-tuned model""")
    parser.add_argument("""--config_path""", default=None, type=str, help="""Path to hf config.json of model to convert""")
    parser.add_argument(
        """--not_finetuned""", action="""store_true""", help="""Whether the model to convert is a fine-tuned model or not"""
    )
    UpperCamelCase_						:					str      				=			parser.parse_args()
    convert_wavaveca_checkpoint(
        args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
    )










"""simple docstring"""





from __future__ import annotations

from PIL import Image

# Define the first generation of cells
# fmt: off
UpperCamelCase_						:					int      				=			[[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1,
          0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]]



def      __lowercase       (     a			:					int			)       ->       list[int]:
      return [int(a			) for c in f'''{ruleset:08}'''[:8]]



def      __lowercase       (     a			:					list[list[int]]							,		a			:					list[int]							,		a			:					int			)       ->       list[int]:
      __snake_case   :     Optional[Any]	      =len(cells[0]			)  # 31
      __snake_case   :     List[str]	      =[]
      for i in range(a			):
            # Get the neighbors of each cell
            # Handle neighbours outside bounds by using 0 as their value
            __snake_case   :     Optional[Any]	      =0 if i == 0 else cells[time][i - 1]
            __snake_case   :     Dict	      =0 if i == population - 1 else cells[time][i + 1]
            # Define a new cell and add it to the new generation
            __snake_case   :     Optional[int]	      =7 - int(f'''{left_neighbor}{cells[time][i]}{right_neighbor}'''							,		2			)
            next_generation.append(rule[situation]			)
      return next_generation



def      __lowercase       (     a			:					list[list[int]]			)       ->       Image.Image:
      __snake_case   :     int	      =Image.new('''RGB'''							,		(len(cells[0]			), len(a			))			)
      __snake_case   :     List[Any]	      =img.load()
      # Generates image
      for w in range(img.width			):
            for h in range(img.height			):
                  __snake_case   :     int	      =255 - int(255 * cells[h][w]			)
                  __snake_case   :     List[str]	      =(color, color, color)
      return img


if __name__ == "__main__":
    UpperCamelCase_						:					Optional[Any]      				=			bin(int(input("""Rule:\n""").strip()))[2:]
    UpperCamelCase_						:					List[str]      				=			format_ruleset(int(rule_num))
    for time in range(16):
        CELLS.append(new_generation(CELLS, rule, time))
    UpperCamelCase_						:					List[Any]      				=			generate_image(CELLS)
    # Uncomment to save the image
    # img.save(f"rule_{rule_num}.png")
    img.show()


