code: 
  
  
  
  
  
  '''simple docstring'''
  
  import unittest
  
  from transformers import SpeechaTextaConfig
  from transformers.testing_utils import is_torch_available, require_torch, torch_device
  
  from ...generation.test_utils import GenerationTesterMixin
  from ...test_configuration_common import ConfigTester
  from ...test_modeling_common import ModelTesterMixin, ids_tensor
  from ...test_pipeline_mixin import PipelineTesterMixin
  
  
  if is_torch_available():
    import torch
  
    from transformers.models.speech_to_text_a.modeling_speech_to_text_a import (
        SpeechaTextaDecoder,
        SpeechaTextaForCausalLM,
    )
  
  
  
  
  
  
  @require_torch
  class      UpperCAmelCase			:
   def __init__(       self   :Any		, lowercase_   :List[Any]		, lowercase_   :Dict=99		, lowercase_   :List[str]=13		, lowercase_   :List[Any]=16		, lowercase_   :Dict=7		, lowercase_   :Any=True		, lowercase_   :List[str]=True		, lowercase_   :int=True		, lowercase_   :str=False		, lowercase_   :Optional[int]=True		, lowercase_   :Optional[int]=2		, lowercase_   :int=32		, lowercase_   :Optional[int]=4		, lowercase_   :Dict=4		, lowercase_   :str=30		, lowercase_   :Any=0		, lowercase_   :List[Any]=1		, lowercase_   :int=2		, lowercase_   :List[str]=None		, )->							Optional[Any]:
       A__       =    parent
       A__       =    batch_size
       A__       =    decoder_seq_length
       # For common tests
       A__       =    self.decoder_seq_length
       A__       =    is_training
       A__       =    use_attention_mask
       A__       =    use_labels
  
       A__       =    vocab_size
       A__       =    d_model
       A__       =    d_model
       A__       =    decoder_layers
       A__       =    decoder_layers
       A__       =    decoder_ffn_dim
       A__       =    decoder_attention_heads
       A__       =    decoder_attention_heads
       A__       =    eos_token_id
       A__       =    bos_token_id
       A__       =    pad_token_id
       A__       =    decoder_start_token_id
       A__       =    use_cache
       A__       =    max_position_embeddings
  
       A__       =    None
       A__       =    decoder_seq_length
       A__       =    2
       A__       =    1
   def   UpperCAmelCase_							(       self   :Any							)->							Optional[Any]:
       A__       =    ids_tensor([self.batch_size, self.decoder_seq_length]		, self.vocab_size							)
  
       A__       =    None
       if self.use_attention_mask:
           A__       =    ids_tensor([self.batch_size, self.decoder_seq_length]		, vocab_size=2							)
  
       A__       =    None
       if self.use_labels:
           A__       =    ids_tensor([self.batch_size, self.decoder_seq_length]		, self.vocab_size							)
  
       A__       =    SpeechaTextaConfig(
           vocab_size=self.vocab_size		, d_model=self.d_model		, decoder_layers=self.decoder_layers		, decoder_ffn_dim=self.decoder_ffn_dim		, decoder_attention_heads=self.decoder_attention_heads		, eos_token_id=self.eos_token_id		, bos_token_id=self.bos_token_id		, use_cache=self.use_cache		, pad_token_id=self.pad_token_id		, decoder_start_token_id=self.decoder_start_token_id		, max_position_embeddings=self.max_position_embeddings		, )
  
       return (
           config,
           input_ids,
           attention_mask,
           lm_labels,
       )
   def   UpperCAmelCase_							(       self   :Tuple		, lowercase_   :Dict		, lowercase_   :List[Any]		, lowercase_   :Tuple		, lowercase_   :Tuple		, )->							Any:
       A__       =    True
       A__       =    SpeechaTextaDecoder(config=lowercase_							).to(lowercase_							).eval()
       A__       =    input_ids[:2]
  
       input_ids[input_ids == 0] += 1
       # first forward pass
       A__       =    model(lowercase_		, use_cache=lowercase_							)
       A__       =    model(lowercase_							)
       A__       =    model(lowercase_		, use_cache=lowercase_							)
  
       self.parent.assertTrue(len(lowercase_							) == len(lowercase_							)							)
       self.parent.assertTrue(len(lowercase_							) == len(lowercase_							) + 1							)
  
       A__       =    outputs["past_key_values"]
  
       # create hypothetical next token and extent to next_input_ids
       A__       =    ids_tensor((2, 1)		, config.vocab_size - 1							) + 1
  
       # append to next input_ids and
       A__       =    torch.cat([input_ids, next_tokens]		, dim=-1							)
  
       print(lowercase_							)
  
       A__       =    model(lowercase_							)["last_hidden_state"]
       A__       =    model(lowercase_		, past_key_values=lowercase_							)["last_hidden_state"]
  
       # select random slice
       A__       =    ids_tensor((1,)		, output_from_past.shape[-1]							).item()
       A__       =    output_from_no_past[:, next_input_ids.shape[-1] - 1, random_slice_idx].detach()
       A__       =    output_from_past[:, 0, random_slice_idx].detach()
  
       # test that outputs are equal for slice
       assert torch.allclose(lowercase_		, lowercase_		, atol=1E-3							)
  
  
  
   def   UpperCAmelCase_							(       self   :Union[str, Any]							)->							Union[str, Any]:
       A__       =    self.prepare_config_and_inputs()
       (
           (
           A__
       ),			(
           A__
       ),			(
           A__
       ),			(
           A__
       ),			
       )       =    config_and_inputs
  
       A__       =    {
           "input_ids": input_ids,
           "attention_mask": attention_mask,
       }
       return config, inputs_dict
  
  
  
  
  
  
  @require_torch
  class      UpperCAmelCase			( UpperCamelCase__    , UpperCamelCase__    , UpperCamelCase__    , unittest.TestCase       ):
   __lowercase     	=       (SpeechaTextaDecoder, SpeechaTextaForCausalLM) if is_torch_available() else ()
   __lowercase     	=       (SpeechaTextaForCausalLM,) if is_torch_available() else ()
   __lowercase     	=       {"""text-generation""": SpeechaTextaForCausalLM} if is_torch_available() else {}
   __lowercase     	=       True
   __lowercase     	=       False
   def   UpperCAmelCase_							(       self   :Tuple		, )->							Any:
       A__       =    SpeechaTextaStandaloneDecoderModelTester(self		, is_training=lowercase_							)
       A__       =    ConfigTester(self		, config_class=lowercase_							)
   def   UpperCAmelCase_							(       self   :Any							)->							Dict:
       pass
   def   UpperCAmelCase_							(       self   :List[Any]							)->							Optional[int]:
       pass
   def   UpperCAmelCase_							(       self   :Union[str, Any]							)->							str:
       pass
   def   UpperCAmelCase_							(       self   :List[Any]							)->							List[str]:
       self.config_tester.run_common_tests()
   def   UpperCAmelCase_							(       self   :int							)->							Dict:
       A__       =    self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_decoder_model_past(*lowercase_							)
  
  
  
   def   UpperCAmelCase_							(       self   :Optional[int]							)->							List[Any]:
       return
  
  
  
  
style_context: 
  
  
  
  
  
  '''simple docstring'''
  
  import getopt
  import json
  import os
  
  # import numpy as np
  import sys
  from collections import OrderedDict
  
  import datasets
  import numpy as np
  import torch
  from modeling_frcnn import GeneralizedRCNN
  from processing_image import Preprocess
  
  from utils import Config
  __lowerCAmelCase							:			str				=False
  __lowerCAmelCase							:			Union[str, Any]				=Config.from_pretrained("unc-nlp/frcnn-vg-finetuned")
  __lowerCAmelCase							:			int				=datasets.Features(
      OrderedDict(
          {
              "attr_ids": datasets.Sequence(length=CONFIG.MAX_DETECTIONS, feature=datasets.Value("float32")),
              "attr_probs": datasets.Sequence(length=CONFIG.MAX_DETECTIONS, feature=datasets.Value("float32")),
              "boxes": datasets.ArrayaD((CONFIG.MAX_DETECTIONS, 4), dtype="float32"),
              "img_id": datasets.Value("int32"),
              "obj_ids": datasets.Sequence(length=CONFIG.MAX_DETECTIONS, feature=datasets.Value("float32")),
              "obj_probs": datasets.Sequence(length=CONFIG.MAX_DETECTIONS, feature=datasets.Value("float32")),
              "roi_features": datasets.ArrayaD((CONFIG.MAX_DETECTIONS, 2048), dtype="float32"),
              "sizes": datasets.Sequence(length=2, feature=datasets.Value("float32")),
              "preds_per_image": datasets.Value(dtype="int32"),
          }
      )
  )
  
  
  
  
  
  
  class      UpperCAmelCase			:
   def __init__(       self   :List[str]		, lowercase_   :List[Any]=sys.argv[1:]							)->							Dict:
       A__       =    None
       A__       =    None
       A__       =    None
       A__       =    1
       A__,			A__       =    getopt.getopt(lowercase_		, "i:o:b:s"		, ["inputdir=", "outfile=", "batch_size=", "subset_list="]							)
       for opt, arg in opts:
           if opt in ("-i", "--inputdir"):
               A__       =    arg
           elif opt in ("-o", "--outfile"):
               A__       =    arg
           elif opt in ("-b", "--batch_size"):
               A__       =    int(lowercase_							)
           elif opt in ("-s", "--subset_list"):
               A__       =    arg
  
       assert inputdir is not None  # and os.path.isdir(inputdir), f"{inputdir}"
       assert outputfile is not None and not os.path.isfile(lowercase_							), F"{outputfile}"
       if subset_list is not None:
           with open(os.path.realpath(lowercase_							)							) as f:
               A__       =    {self._vqa_file_split()[0] for x in tryload(lowercase_							)}
       else:
           A__       =    None
  
       A__       =    CONFIG
       if torch.cuda.is_available():
           A__       =    "cuda"
       A__       =    os.path.realpath(lowercase_							)
       A__       =    os.path.realpath(lowercase_							)
       A__       =    Preprocess(self.config							)
       A__       =    GeneralizedRCNN.from_pretrained("unc-nlp/frcnn-vg-finetuned"		, config=self.config							)
       A__       =    batch_size if batch_size != 0 else 1
       A__       =    DEFAULT_SCHEMA
   def   UpperCAmelCase_							(       self   :Optional[Any]		, lowercase_   :Optional[Any]							)->							Dict:
       A__       =    int(file.split("."							)[0].split("_"							)[-1]							)
       A__       =    os.path.join(self.inputdir		, lowercase_							)
       return (img_id, filepath)
   @property
   def   UpperCAmelCase_							(       self   :Union[str, Any]							)->							Optional[int]:
       A__       =    []
       for i, file in enumerate(os.listdir(self.inputdir							)							):
           if self.subset_list is not None and i not in self.subset_list:
               continue
           batch.append(self._vqa_file_split(lowercase_							)							)
           if len(lowercase_							) == self.batch:
               A__       =    batch
               A__       =    []
               yield list(map(lowercase_		, zip(*lowercase_							)							)							)
  
       for i in range(1							):
           yield list(map(lowercase_		, zip(*lowercase_							)							)							)
  
  
  
   def __call__(       self   :Optional[int]							)->							List[str]:
       # make writer
       if not TEST:
           A__       =    datasets.ArrowWriter(features=self.schema		, path=self.outputfile							)
       # do file generator
       for i, (img_ids, filepaths) in enumerate(self.file_generator							):
           A__,			A__,			A__       =    self.preprocess(lowercase_							)
           A__       =    self.model(
               lowercase_		, lowercase_		, scales_yx=lowercase_		, padding="max_detections"		, max_detections=self.config.MAX_DETECTIONS		, pad_value=0		, return_tensors="np"		, location="cpu"		, )
           A__       =    output_dict.pop("normalized_boxes"							)
           if not TEST:
               A__       =    np.array(lowercase_							)
               A__       =    self.schema.encode_batch(lowercase_							)
               writer.write_batch(lowercase_							)
           if TEST:
               break
           # finalizer the writer
       if not TEST:
           A__,			A__       =    writer.finalize()
           print(F"Success! You wrote {num_examples} entry(s) and {num_bytes >> 20} mb"							)
  
  
  
  
  
  def 		UpperCamelCase       (							_lowerCamelCase      :      Optional[int] ):
      try:
          A__       =    json.load(_lowerCamelCase )
          try:
              A__       =    list(data.keys() )
          except Exception:
              A__       =    [d["img_id"] for d in data]
      except Exception:
          try:
              A__       =    eval(stream.read() )
          except Exception:
              A__       =    stream.read().split("\n" )
      return data
  
  
  if __name__ == "__main__":
    __lowerCAmelCase							:			int				=Extract(sys.argv[1:])
    extract()
    if not TEST:
      __lowerCAmelCase							:			Optional[Any]				=datasets.Dataset.from_file(extract.outputfile)
      # wala!
      # print(np.array(dataset[0:2]["roi_features"]).shape)
  
  
  
  