
"""simple docstring"""



import argparse
import random

import joblib
import numpy as np
import torch
from igf.igf import (
    SecondaryLearner,
    collect_objective_set,
    compute_perplexity,
    generate_datasets,
    load_gpta,
    recopy_gpta,
    set_seed,
    train_secondary_learner,
)
from torch.utils.data import DataLoader, RandomSampler

from transformers import GPTaLMHeadModel


def  _UpperCamelCase	(						_A=3_2					,      _A=1_0					,      _A=1_0_0					,      _A=1_0_2_6					,      _A=True					,      _A="data/tokenized_stories_train_wikitext103.jbl"					,      _A="igf_context_pairs.jbl"					,      )    ->  List[str]:





    """simple docstring"""

    set_seed(3      )
    # generate train_data and objective_set
    _UpperCAmelCase    ,_UpperCAmelCase          =   generate_datasets(
        _A					,      _A					,      number=_A					,      min_len=1_0_2_6					,      trim=_A      )
    # keeps model same across runs
    set_seed(4      )
    # model, lm_optimizer, lm_scheduler = recopy_gpt2(model, device, max_steps) # store original model weights
    # can we train on GPU?
    _UpperCAmelCase          =   torch.device("""cuda:0""" if torch.cuda.is_available() else """cpu"""      )

    # load pretrained model
    _UpperCAmelCase          =   load_gpta("""gpt2"""      ).to(_A      )
    print("""computing perplexity on objective set"""      )
    _UpperCAmelCase          =   compute_perplexity(_A					,      _A					,      _A      ).item()
    print("""perplexity on objective set:"""					,      _A      )

    # collect igf pairs and save to file demo.jbl
    collect_objective_set(_A					,      _A					,      _A					,      _A					,      _A					,      _A					,      _A					,      _A      )

    # clean up, delete model and data we don't need anymore
    del model, train_data, objective_set
    torch.cuda.empty_cache()


def  _UpperCamelCase	(						_A					,      _A=1_5					,      _A=1_2_8					,      _A=1_0_0					,      _A="igf_model.pt"					,      )    ->  Any:





    """simple docstring"""

    set_seed(4_2      )

    # Load pre-trained model
    _UpperCAmelCase          =   GPTaLMHeadModel.from_pretrained("""gpt2"""      )

    # Initialize secondary learner to use embedding weights of model
    _UpperCAmelCase          =   SecondaryLearner(_A      )

    # Train secondary learner
    _UpperCAmelCase          =   train_secondary_learner(
        _A					,      _A					,      max_epochs=_A					,      batch_size=_A					,      eval_freq=1_0_0					,      igf_model_path=_A					,      )

    del model, secondary_learner_train_data
    torch.cuda.empty_cache()

    return secondary_learner


def  _UpperCamelCase	(						_A					,      _A					,      _A					,      _A=3_2					,      _A=1_0_0_0					,      _A=1_6					,      _A=1.0					,      _A=recopy_gpta					,      _A=None					,      _A=1_0					,      _A="gpt2_finetuned.pt"					,      )    ->  List[Any]:





    """simple docstring"""

    _UpperCAmelCase          =   torch.device("""cuda:0""" if torch.cuda.is_available() else """cpu"""      )
    _UpperCAmelCase          =   RandomSampler(_A      )
    _UpperCAmelCase          =   DataLoader(_A					,      sampler=_A      )

    _UpperCAmelCase          =   max_steps // (len(_A      )) + 1
    _UpperCAmelCase          =   0
    _UpperCAmelCase          =   torch.zeros((1, context_len)					,      dtype=torch.long					,      device=_A      )
    _UpperCAmelCase    ,_UpperCAmelCase    ,_UpperCAmelCase          =   recopy_model(_A					,      _A					,      _A      )

    model.train()
    if secondary_learner is not None:
        secondary_learner.to(_A      )
        secondary_learner.eval()
    _UpperCAmelCase          =   []
    _UpperCAmelCase          =   0

    _UpperCAmelCase          =   []
    _UpperCAmelCase          =   []

    # Compute the performance of the transformer model at the beginning
    _UpperCAmelCase          =   compute_perplexity(_A					,      _A					,      _A      )
    test_perps.append(_A      )
    print("""Test perplexity, step"""					,      _A					,      """:"""					,      _A      )
    for epoch in range(int(_A      )      ):
        for step, example in enumerate(_A      ):
            torch.cuda.empty_cache()
            _UpperCAmelCase          =   random.randint(0					,      example.size(2      ) - context_len - 1      )
            _UpperCAmelCase          =   example[0, 0, start : start + context_len]
            lm_optimizer.zero_grad()
            _UpperCAmelCase          =   model(_A					,      labels=_A      )
            _UpperCAmelCase          =   True

            if secondary_learner is not None:
                _UpperCAmelCase          =   secondary_learner.forward(
                    torch.tensor(_A					,      dtype=torch.long					,      device=_A      ).unsqueeze(0      )      )[0].item()
                observed_qs.append(float(_A      )      )

                # Here we implement the simple non-constant threshold for the predicted IG(X) value
                # We will decay the selectivity of our secondary learner filter from
                # 1 standard deviation above average to 1 below average after 10 batches.

                if global_step == 1_0:
                    _UpperCAmelCase          =   -1
                if predicted_q < threshold:
                    _UpperCAmelCase          =   False

            # If we passed the filter, add the context to the batch!
            if do_backprop:
                contexts.append(np.array(context.cpu()      )      )
                _UpperCAmelCase          =   outputs[0]
                lm_loss.backward()
                examples += 1

            del outputs

            # Once the batch is filled with enough contexts, backprop on the batch.
            if examples == batch_size:
                torch.cuda.empty_cache()
                _UpperCAmelCase          =   0
                # Do LM backprop
                torch.nn.utils.clip_grad_norm_(model.parameters()					,      3.0      )
                lm_optimizer.step()
                lm_scheduler.step()  # Update learning rate schedule
                global_step += 1
                # Compute the performance of the transformer model at this batch
                if global_step % eval_interval == 0:
                    _UpperCAmelCase          =   compute_perplexity(_A					,      _A					,      _A      )
                    test_perps.append(_A      )

                    print("""Test perplexity, step"""					,      _A					,      """:"""					,      _A      )
            # Break out of the loop after 60 batches
            if max_steps > 0 and global_step > 6_0:
                break
        if max_steps > 0 and global_step > 6_0:
            break

    # save finetuned transformer model
    torch.save(model.state_dict()					,      _A      )
    torch.cuda.empty_cache()
    # Do some cleaning up so we can reinitialize for the next run of this function
    del lm_optimizer
    del lm_scheduler
    return model


def  _UpperCamelCase	(						)    ->  Optional[int]:





    """simple docstring"""

    _UpperCAmelCase          =   argparse.ArgumentParser(description="""Fine-tune a transformer model with IGF on a language modeling task"""      )

    # Required parameters
    parser.add_argument(
        """--data_dir"""					,      default=_A					,      type=_A					,      required=_A					,      help="""The input data dir. Should contain data files for WikiText."""					,      )
    parser.add_argument(
        """--model_name_or_path"""					,      default=_A					,      type=_A					,      required=_A					,      help="""Path to pretrained model or model identifier from huggingface.co/models"""					,      )
    parser.add_argument(
        """--data_file"""					,      type=_A					,      default=_A					,      help=(
            """A jbl file containing tokenized data which can be split as objective dataset, """
            """train_dataset and test_dataset."""
        )					,      )

    parser.add_argument(
        """--igf_data_file"""					,      type=_A					,      default=_A					,      help="""A jbl file containing the context and information gain pairs to train secondary learner."""					,      )

    parser.add_argument(
        """--output_dir"""					,      default=_A					,      type=_A					,      required=_A					,      help="""The output directory where the final fine-tuned model is stored."""					,      )

    parser.add_argument(
        """--tokenizer_name"""					,      default=_A					,      type=_A					,      help="""Pretrained tokenizer name or path if not the same as model_name"""					,      )
    parser.add_argument("""--seed"""					,      type=_A					,      default=_A					,      help="""A seed for reproducible training."""      )

    parser.add_argument(
        """--context_len"""					,      default=3_2					,      type=_A					,      help=(
            """The maximum total input sequence length after tokenization. Sequences longer """
            """than this will be truncated, sequences shorter will be padded."""
        )					,      )

    parser.add_argument(
        """--size_objective_set"""					,      default=1_0_0					,      type=_A					,      help="""number of articles that are long enough to be used as our objective set"""					,      )
    parser.add_argument(
        """--eval_freq"""					,      default=1_0_0					,      type=_A					,      help="""secondary model evaluation is triggered at eval_freq"""      )

    parser.add_argument("""--max_steps"""					,      default=1_0_0_0					,      type=_A					,      help="""To calculate training epochs"""      )

    parser.add_argument(
        """--secondary_learner_batch_size"""					,      default=1_2_8					,      type=_A					,      help="""batch size of training data for secondary learner"""					,      )

    parser.add_argument(
        """--batch_size"""					,      default=1_6					,      type=_A					,      help="""batch size of training data of language model(gpt2) """      )

    parser.add_argument(
        """--eval_interval"""					,      default=1_0					,      type=_A					,      help=(
            """decay the selectivity of our secondary learner filter from"""
            """1 standard deviation above average to 1 below average after 10 batches"""
        )					,      )

    parser.add_argument(
        """--number"""					,      default=1_0_0					,      type=_A					,      help="""The number of examples split to be used as objective_set/test_data"""      )

    parser.add_argument(
        """--min_len"""					,      default=1_0_2_6					,      type=_A					,      help="""The minimum length of the article to be used as objective set"""      )

    parser.add_argument(
        """--secondary_learner_max_epochs"""					,      default=1_5					,      type=_A					,      help="""number of epochs to train secondary learner"""      )

    parser.add_argument("""--trim"""					,      default=_A					,      type=_A					,      help="""truncate the example if it exceeds context length"""      )

    parser.add_argument(
        """--threshold"""					,      default=1.0					,      type=_A					,      help=(
            """The threshold value used by secondary learner to filter the train_data and allow only"""
            """ informative data as input to the model"""
        )					,      )

    parser.add_argument("""--finetuned_model_name"""					,      default="""gpt2_finetuned.pt"""					,      type=_A					,      help="""finetuned_model_name"""      )

    parser.add_argument(
        """--recopy_model"""					,      default=_A					,      type=_A					,      help="""Reset the model to the original pretrained GPT-2 weights after each iteration"""					,      )

    # function calls
    # Collecting *n* pairs of context and information gain(X, IG(X)) for training the secondary learner
    generate_n_pairs(
        context_len=3_2					,      max_steps=1_0					,      size_objective_set=1_0_0					,      min_len=1_0_2_6					,      trim=_A					,      data_file="""data/tokenized_stories_train_wikitext103.jbl"""					,      igf_data_file="""igf_context_pairs.jbl"""					,      )

    # Load train data for secondary learner
    _UpperCAmelCase          =   joblib.load("""data/IGF_values.jbl"""      )

    # Train secondary learner
    _UpperCAmelCase          =   training_secondary_learner(
        _A					,      secondary_learner_max_epochs=1_5					,      secondary_learner_batch_size=1_2_8					,      eval_freq=1_0_0					,      igf_model_path="""igf_model.pt"""					,      )

    # load pretrained gpt2 model
    _UpperCAmelCase          =   GPTaLMHeadModel.from_pretrained("""gpt2"""      )
    set_seed(4_2      )

    # Generate train and test data to train and evaluate gpt2 model
    _UpperCAmelCase    ,_UpperCAmelCase          =   generate_datasets(
        context_len=3_2					,      file="""data/tokenized_stories_train_wikitext103.jbl"""					,      number=1_0_0					,      min_len=1_0_2_6					,      trim=_A      )

    # fine-tuning of the gpt2 model using igf (Information Gain Filtration)
    finetune(
        _A					,      _A					,      _A					,      context_len=3_2					,      max_steps=1_0_0_0					,      batch_size=1_6					,      threshold=1.0					,      recopy_model=_A					,      secondary_learner=_A					,      eval_interval=1_0					,      finetuned_model_name="""gpt2_finetuned.pt"""					,      )


if __name__ == "__main__":
 main()

"""simple docstring"""



import copy
import os
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
import pyarrow as pa
import pyarrow.parquet as pq
import pytest

from datasets.arrow_writer import ArrowWriter, OptimizedTypedSequence, ParquetWriter, TypedSequence
from datasets.features import ArrayaD, ClassLabel, Features, Image, Value
from datasets.features.features import ArrayaDExtensionType, cast_to_python_objects
from datasets.keyhash import DuplicatedKeysError, InvalidKeyError

from .utils import require_pil




class 		a_						( _UpperCAmelCase	):







    def      _snake_case					(			self			:       str     )		->str:






        '''simple docstring'''






        _UpperCAmelCase          =   pa.array(TypedSequence([1, 2, 3]     )     )
        self.assertEqual(arr.type     ,					pa.intaa()     )







    def      _snake_case					(			self			:       Any     )		->List[str]:






        '''simple docstring'''






        with self.assertRaises(__UpperCamelCase     ):
            _UpperCAmelCase          =   pa.array(TypedSequence([1, 2, 3]     )     ,					type=pa.intaa()     )







    def      _snake_case					(			self			:       Optional[int]     )		->Tuple:






        '''simple docstring'''






        with self.assertRaises(__UpperCamelCase     ):
            _UpperCAmelCase          =   pa.array(TypedSequence([1, 2, 3]     ,					try_type=Value("""bool"""     )     ,					type=Value("""int64"""     )     )     )







    def      _snake_case					(			self			:       List[Any]     )		->Dict:






        '''simple docstring'''






        _UpperCAmelCase          =   pa.array(TypedSequence([1, 2, 3]     ,					type=Value("""int32"""     )     )     )
        self.assertEqual(arr.type     ,					pa.intaa()     )







    def      _snake_case					(			self			:       str     )		->Dict:






        '''simple docstring'''






        with self.assertRaises((TypeError, pa.lib.ArrowInvalid)     ):
            _UpperCAmelCase          =   pa.array(TypedSequence(["""foo""", """bar"""]     ,					type=Value("""int64"""     )     )     )







    def      _snake_case					(			self			:       Union[str, Any]     )		->str:






        '''simple docstring'''






        _UpperCAmelCase          =   pa.array(TypedSequence([1, 2, 3]     ,					try_type=Value("""int32"""     )     )     )
        self.assertEqual(arr.type     ,					pa.intaa()     )







    def      _snake_case					(			self			:       List[str]     )		->Any:






        '''simple docstring'''






        _UpperCAmelCase          =   pa.array(TypedSequence(["""foo""", """bar"""]     ,					try_type=Value("""int64"""     )     )     )
        self.assertEqual(arr.type     ,					pa.string()     )







    def      _snake_case					(			self			:       List[Any]     )		->List[Any]:






        '''simple docstring'''






        _UpperCAmelCase          =   pa.array(TypedSequence([[[1, 2, 3]]]     ,					type=ArrayaD((1, 3)     ,					"""int64"""     )     )     )
        self.assertEqual(arr.type     ,					ArrayaDExtensionType((1, 3)     ,					"""int64"""     )     )







    def      _snake_case					(			self			:       List[Any]     )		->Optional[Any]:






        '''simple docstring'''






        with self.assertRaises((TypeError, pa.lib.ArrowInvalid)     ):
            _UpperCAmelCase          =   pa.array(TypedSequence(["""foo""", """bar"""]     ,					type=ArrayaD((1, 3)     ,					"""int64"""     )     )     )







    def      _snake_case					(			self			:       List[str]     )		->int:






        '''simple docstring'''






        _UpperCAmelCase          =   pa.array(TypedSequence([[[1, 2, 3]]]     ,					try_type=ArrayaD((1, 3)     ,					"""int64"""     )     )     )
        self.assertEqual(arr.type     ,					ArrayaDExtensionType((1, 3)     ,					"""int64"""     )     )







    def      _snake_case					(			self			:       Optional[int]     )		->Dict:






        '''simple docstring'''






        _UpperCAmelCase          =   pa.array(TypedSequence(["""foo""", """bar"""]     ,					try_type=ArrayaD((1, 3)     ,					"""int64"""     )     )     )
        self.assertEqual(arr.type     ,					pa.string()     )







    @require_pil
    def      _snake_case					(			self			:       str     )		->Optional[Any]:






        '''simple docstring'''






        import PIL.Image

        _UpperCAmelCase          =   PIL.Image.fromarray(np.arange(10     ,					dtype=np.uinta     ).reshape(2     ,					5     )     )
        with patch(
            """datasets.arrow_writer.cast_to_python_objects"""     ,					side_effect=__UpperCamelCase     ) as mock_cast_to_python_objects:
            _UpperCAmelCase          =   pa.array(TypedSequence([{"""path""": None, """bytes""": b"""image_bytes"""}, pil_image]     ,					type=Image()     )     )
            _UpperCAmelCase    ,_UpperCAmelCase          =   mock_cast_to_python_objects.call_args_list[-1]
            self.assertIn("""optimize_list_casting"""     ,					__UpperCamelCase     )
            self.assertFalse(kwargs["""optimize_list_casting"""]     )







def  _UpperCamelCase	(						_A					,      _A      )    ->  Any:





    """simple docstring"""

    _UpperCAmelCase          =   pa.BufferReader(_A      ) if isinstance(_A					,      pa.Buffer      ) else pa.memory_map(_A      )
    _UpperCAmelCase          =   pa.ipc.open_stream(_A      )
    _UpperCAmelCase          =   f.read_all()
    assert len(pa_table.to_batches()      ) == expected_num_chunks
    assert pa_table.to_pydict() == {"col_1": ["foo", "bar"], "col_2": [1, 2]}
    del pa_table


@pytest.mark.parametrize("""writer_batch_size"""					,      [None, 1, 1_0]      )
@pytest.mark.parametrize(
    """fields"""					,      [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]      )
def  _UpperCamelCase	(						_A					,      _A      )    ->  Optional[Any]:





    """simple docstring"""

    _UpperCAmelCase          =   pa.BufferOutputStream()
    _UpperCAmelCase          =   pa.schema(_A      ) if fields else None
    with ArrowWriter(stream=_A					,      schema=_A					,      writer_batch_size=_A      ) as writer:
        writer.write({"""col_1""": """foo""", """col_2""": 1}      )
        writer.write({"""col_1""": """bar""", """col_2""": 2}      )
        _UpperCAmelCase    ,_UpperCAmelCase          =   writer.finalize()
    assert num_examples == 2
    assert num_bytes > 0
    if not fields:
        _UpperCAmelCase          =   {"""col_1""": pa.string(), """col_2""": pa.intaa()}
    assert writer._schema == pa.schema(_A					,      metadata=writer._schema.metadata      )
    _check_output(output.getvalue()					,      expected_num_chunks=num_examples if writer_batch_size == 1 else 1      )


def  _UpperCamelCase	(						)    ->  Union[str, Any]:





    """simple docstring"""

    _UpperCAmelCase          =   pa.BufferOutputStream()
    _UpperCAmelCase          =   Features({"""labels""": ClassLabel(names=["""neg""", """pos"""]      )}      )
    with ArrowWriter(stream=_A					,      features=_A      ) as writer:
        writer.write({"""labels""": 0}      )
        writer.write({"""labels""": 1}      )
        _UpperCAmelCase    ,_UpperCAmelCase          =   writer.finalize()
    assert num_examples == 2
    assert num_bytes > 0
    assert writer._schema == features.arrow_schema
    assert writer._schema.metadata == features.arrow_schema.metadata
    _UpperCAmelCase          =   pa.BufferReader(output.getvalue()      )
    _UpperCAmelCase          =   pa.ipc.open_stream(_A      )
    _UpperCAmelCase          =   f.read_all()
    _UpperCAmelCase          =   pa_table.schema
    assert pa_table.num_rows == 2
    assert schema == features.arrow_schema
    assert schema.metadata == features.arrow_schema.metadata
    assert features == Features.from_arrow_schema(_A      )


@pytest.mark.parametrize("""writer_batch_size"""					,      [None, 1, 1_0]      )
def  _UpperCamelCase	(						_A      )    ->  Optional[int]:





    """simple docstring"""

    _UpperCAmelCase          =   pa.BufferOutputStream()
    with ArrowWriter(
        stream=_A					,      writer_batch_size=_A					,      hash_salt="""split_name"""					,      check_duplicates=_A					,      ) as writer:
        with pytest.raises(_A      ):
            writer.write({"""col_1""": """foo""", """col_2""": 1}					,      key=[1, 2]      )
            _UpperCAmelCase    ,_UpperCAmelCase          =   writer.finalize()


@pytest.mark.parametrize("""writer_batch_size"""					,      [None, 2, 1_0]      )
def  _UpperCamelCase	(						_A      )    ->  List[Any]:





    """simple docstring"""

    _UpperCAmelCase          =   pa.BufferOutputStream()
    with ArrowWriter(
        stream=_A					,      writer_batch_size=_A					,      hash_salt="""split_name"""					,      check_duplicates=_A					,      ) as writer:
        with pytest.raises(_A      ):
            writer.write({"""col_1""": """foo""", """col_2""": 1}					,      key=1_0      )
            writer.write({"""col_1""": """bar""", """col_2""": 2}					,      key=1_0      )
            _UpperCAmelCase    ,_UpperCAmelCase          =   writer.finalize()


@pytest.mark.parametrize("""writer_batch_size"""					,      [None, 2, 1_0]      )
def  _UpperCamelCase	(						_A      )    ->  Tuple:





    """simple docstring"""

    _UpperCAmelCase          =   pa.BufferOutputStream()
    with ArrowWriter(
        stream=_A					,      writer_batch_size=_A					,      hash_salt="""split_name"""					,      check_duplicates=_A					,      ) as writer:
        writer.write({"""col_1""": """foo""", """col_2""": 1}					,      key=1      )
        writer.write({"""col_1""": """bar""", """col_2""": 2}					,      key=2      )
        _UpperCAmelCase    ,_UpperCAmelCase          =   writer.finalize()
    assert num_examples == 2
    assert num_bytes > 0
    _check_output(output.getvalue()					,      expected_num_chunks=num_examples if writer_batch_size == 1 else 1      )


@pytest.mark.parametrize("""writer_batch_size"""					,      [None, 1, 1_0]      )
@pytest.mark.parametrize(
    """fields"""					,      [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]      )
def  _UpperCamelCase	(						_A					,      _A      )    ->  List[Any]:





    """simple docstring"""

    _UpperCAmelCase          =   pa.BufferOutputStream()
    _UpperCAmelCase          =   pa.schema(_A      ) if fields else None
    with ArrowWriter(stream=_A					,      schema=_A					,      writer_batch_size=_A      ) as writer:
        writer.write_batch({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}      )
        writer.write_batch({"""col_1""": [], """col_2""": []}      )
        _UpperCAmelCase    ,_UpperCAmelCase          =   writer.finalize()
    assert num_examples == 2
    assert num_bytes > 0
    if not fields:
        _UpperCAmelCase          =   {"""col_1""": pa.string(), """col_2""": pa.intaa()}
    assert writer._schema == pa.schema(_A					,      metadata=writer._schema.metadata      )
    _check_output(output.getvalue()					,      expected_num_chunks=num_examples if writer_batch_size == 1 else 1      )


@pytest.mark.parametrize("""writer_batch_size"""					,      [None, 1, 1_0]      )
@pytest.mark.parametrize(
    """fields"""					,      [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]      )
def  _UpperCamelCase	(						_A					,      _A      )    ->  Any:





    """simple docstring"""

    _UpperCAmelCase          =   pa.BufferOutputStream()
    _UpperCAmelCase          =   pa.schema(_A      ) if fields else None
    with ArrowWriter(stream=_A					,      schema=_A					,      writer_batch_size=_A      ) as writer:
        writer.write_table(pa.Table.from_pydict({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}      )      )
        _UpperCAmelCase    ,_UpperCAmelCase          =   writer.finalize()
    assert num_examples == 2
    assert num_bytes > 0
    if not fields:
        _UpperCAmelCase          =   {"""col_1""": pa.string(), """col_2""": pa.intaa()}
    assert writer._schema == pa.schema(_A					,      metadata=writer._schema.metadata      )
    _check_output(output.getvalue()					,      expected_num_chunks=num_examples if writer_batch_size == 1 else 1      )


@pytest.mark.parametrize("""writer_batch_size"""					,      [None, 1, 1_0]      )
@pytest.mark.parametrize(
    """fields"""					,      [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]      )
def  _UpperCamelCase	(						_A					,      _A      )    ->  List[Any]:





    """simple docstring"""

    _UpperCAmelCase          =   pa.BufferOutputStream()
    _UpperCAmelCase          =   pa.schema(_A      ) if fields else None
    with ArrowWriter(stream=_A					,      schema=_A					,      writer_batch_size=_A      ) as writer:
        writer.write_row(pa.Table.from_pydict({"""col_1""": ["""foo"""], """col_2""": [1]}      )      )
        writer.write_row(pa.Table.from_pydict({"""col_1""": ["""bar"""], """col_2""": [2]}      )      )
        _UpperCAmelCase    ,_UpperCAmelCase          =   writer.finalize()
    assert num_examples == 2
    assert num_bytes > 0
    if not fields:
        _UpperCAmelCase          =   {"""col_1""": pa.string(), """col_2""": pa.intaa()}
    assert writer._schema == pa.schema(_A					,      metadata=writer._schema.metadata      )
    _check_output(output.getvalue()					,      expected_num_chunks=num_examples if writer_batch_size == 1 else 1      )


def  _UpperCamelCase	(						)    ->  str:





    """simple docstring"""

    with tempfile.TemporaryDirectory() as tmp_dir:
        _UpperCAmelCase          =   {"""col_1""": pa.string(), """col_2""": pa.intaa()}
        _UpperCAmelCase          =   os.path.join(_A					,      """test.arrow"""      )
        with ArrowWriter(path=_A					,      schema=pa.schema(_A      )      ) as writer:
            writer.write_batch({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}      )
            _UpperCAmelCase    ,_UpperCAmelCase          =   writer.finalize()
        assert num_examples == 2
        assert num_bytes > 0
        assert writer._schema == pa.schema(_A					,      metadata=writer._schema.metadata      )
        _check_output(_A					,      1      )


def  _UpperCamelCase	(						_A      )    ->  int:





    """simple docstring"""

    if pa.types.is_list(_A      ):
        return get_base_dtype(arr_type.value_type      )
    else:
        return arr_type


def  _UpperCamelCase	(						_A					,      _A      )    ->  Any:





    """simple docstring"""

    if isinstance(lst[0]					,      _A      ):
        change_first_primitive_element_in_list(lst[0]					,      _A      )
    else:
        _UpperCAmelCase          =   value


@pytest.mark.parametrize("""optimized_int_type, expected_dtype"""					,      [(None, pa.intaa()), (Value("""int32"""      ), pa.intaa())]      )
@pytest.mark.parametrize("""sequence"""					,      [[1, 2, 3], [[1, 2, 3]], [[[1, 2, 3]]]]      )
def  _UpperCamelCase	(						_A					,      _A					,      _A      )    ->  Dict:





    """simple docstring"""

    _UpperCAmelCase          =   pa.array(TypedSequence(_A					,      optimized_int_type=_A      )      )
    assert get_base_dtype(arr.type      ) == expected_dtype


@pytest.mark.parametrize(
    """col, expected_dtype"""					,      [
        ("""attention_mask""", pa.inta()),
        ("""special_tokens_mask""", pa.inta()),
        ("""token_type_ids""", pa.inta()),
        ("""input_ids""", pa.intaa()),
        ("""other""", pa.intaa()),
    ]					,      )
@pytest.mark.parametrize("""sequence"""					,      [[1, 2, 3], [[1, 2, 3]], [[[1, 2, 3]]]]      )
def  _UpperCamelCase	(						_A					,      _A					,      _A      )    ->  str:





    """simple docstring"""

    _UpperCAmelCase          =   pa.array(OptimizedTypedSequence(_A					,      col=_A      )      )
    assert get_base_dtype(arr.type      ) == expected_dtype

    # not in range
    if col != "other":
        # avoids errors due to in-place modifications
        _UpperCAmelCase          =   copy.deepcopy(_A      )
        _UpperCAmelCase          =   np.iinfo(expected_dtype.to_pandas_dtype()      ).max + 1
        change_first_primitive_element_in_list(_A					,      _A      )
        _UpperCAmelCase          =   pa.array(OptimizedTypedSequence(_A					,      col=_A      )      )
        assert get_base_dtype(arr.type      ) == pa.intaa()


@pytest.mark.parametrize("""raise_exception"""					,      [False, True]      )
def  _UpperCamelCase	(						_A					,      _A      )    ->  Dict:





    """simple docstring"""

    _UpperCAmelCase          =   str(tmp_path / """dataset-train.arrow"""      )
    try:
        with ArrowWriter(path=_A      ) as writer:
            if raise_exception:
                raise pa.lib.ArrowInvalid()
            else:
                writer.stream.close()
    except pa.lib.ArrowInvalid:
        pass
    finally:
        assert writer.stream.closed


def  _UpperCamelCase	(						_A      )    ->  Dict:





    """simple docstring"""

    _UpperCAmelCase          =   """mock://dataset-train.arrow"""
    with ArrowWriter(path=_A					,      storage_options=mockfs.storage_options      ) as writer:
        assert isinstance(writer._fs					,      type(_A      )      )
        assert writer._fs.storage_options == mockfs.storage_options
        writer.write({"""col_1""": """foo""", """col_2""": 1}      )
        writer.write({"""col_1""": """bar""", """col_2""": 2}      )
        _UpperCAmelCase    ,_UpperCAmelCase          =   writer.finalize()
    assert num_examples == 2
    assert num_bytes > 0
    assert mockfs.exists(_A      )


def  _UpperCamelCase	(						)    ->  Dict:





    """simple docstring"""

    _UpperCAmelCase          =   pa.BufferOutputStream()
    with ParquetWriter(stream=_A      ) as writer:
        writer.write({"""col_1""": """foo""", """col_2""": 1}      )
        writer.write({"""col_1""": """bar""", """col_2""": 2}      )
        _UpperCAmelCase    ,_UpperCAmelCase          =   writer.finalize()
    assert num_examples == 2
    assert num_bytes > 0
    _UpperCAmelCase          =   pa.BufferReader(output.getvalue()      )
    _UpperCAmelCase          =   pq.read_table(_A      )
    assert pa_table.to_pydict() == {"col_1": ["foo", "bar"], "col_2": [1, 2]}


@require_pil
@pytest.mark.parametrize("""embed_local_files"""					,      [False, True]      )
def  _UpperCamelCase	(						_A					,      _A      )    ->  Any:





    """simple docstring"""

    import PIL.Image

    _UpperCAmelCase          =   str(tmp_path / """test_image_rgb.jpg"""      )
    PIL.Image.fromarray(np.zeros((5, 5)					,      dtype=np.uinta      )      ).save(_A					,      format="""png"""      )
    _UpperCAmelCase          =   pa.BufferOutputStream()
    with ParquetWriter(
        stream=_A					,      features=Features({"""image""": Image()}      )					,      embed_local_files=_A      ) as writer:
        writer.write({"""image""": image_path}      )
        writer.finalize()
    _UpperCAmelCase          =   pa.BufferReader(output.getvalue()      )
    _UpperCAmelCase          =   pq.read_table(_A      )
    _UpperCAmelCase          =   pa_table.to_pydict()
    if embed_local_files:
        assert isinstance(out["""image"""][0]["""path"""]					,      _A      )
        with open(_A					,      """rb"""      ) as f:
            assert out["image"][0]["bytes"] == f.read()
    else:
        assert out["image"][0]["path"] == image_path
        assert out["image"][0]["bytes"] is None







def  _UpperCamelCase	(						)    ->  int:





    """simple docstring"""

    _UpperCAmelCase          =   pa.schema([pa.field("""col_1"""					,      pa.string()					,      nullable=_A      )]      )
    _UpperCAmelCase          =   pa.BufferOutputStream()
    with ArrowWriter(stream=_A      ) as writer:
        writer._build_writer(inferred_schema=_A      )
        assert writer._schema == pa.schema([pa.field("""col_1"""					,      pa.string()      )]      )