





import os
import tempfile
from functools import partial
from unittest import TestCase
from unittest.mock import patch

import datasets
import datasets.config

from .utils import require_beam






class 							snake_case_    (		datasets.BeamBasedBuilder						):






  def 				UpperCAmelCase__						(							self			:					int      )->Tuple:






   '''simple docstring'''

   return datasets.DatasetInfo(
       features=datasets.Features({"""content""": datasets.Value("""string"""      )}      )       ,  supervised_keys=_snake_case       ,  )






  def 				UpperCAmelCase__						(							self			:					str       ,  _snake_case			:					int       ,  _snake_case			:					int      )->Tuple:






   '''simple docstring'''

   return [datasets.SplitGenerator(name=datasets.Split.TRAIN       ,  gen_kwargs={"""examples""": get_test_dummy_examples()}      )]






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






   '''simple docstring'''

   import apache_beam as beam

   return pipeline | "Load Examples" >> beam.Create(_snake_case      )






class 							snake_case_    (		datasets.BeamBasedBuilder						):






  def 				UpperCAmelCase__						(							self			:					int      )->str:






   '''simple docstring'''

   return datasets.DatasetInfo(
       features=datasets.Features({"""a""": datasets.Sequence({"""b""": datasets.Value("""string"""      )}      )}      )       ,  supervised_keys=_snake_case       ,  )






  def 				UpperCAmelCase__						(							self			:					int       ,  _snake_case			:					Tuple       ,  _snake_case			:					Tuple      )->Dict:






   '''simple docstring'''

   return [
       datasets.SplitGenerator(name=datasets.Split.TRAIN       ,  gen_kwargs={"""examples""": get_test_nested_examples()}      )
   ]






  def 				UpperCAmelCase__						(							self			:					Optional[int]       ,  _snake_case			:					Any       ,  _snake_case			:					List[str]      )->Optional[Any]:






   '''simple docstring'''

   import apache_beam as beam

   return pipeline | "Load Examples" >> beam.Create(_snake_case      )






def 	_SCREAMING_SNAKE_CASE		(      )      ->   Optional[int]:
 return [(i, {"content": content}) for i, content in enumerate(["""foo""", """bar""", """foobar"""]   )]






def 	_SCREAMING_SNAKE_CASE		(      )      ->   str:
 return [(i, {"a": {"b": [content]}}) for i, content in enumerate(["""foo""", """bar""", """foobar"""]   )]







class 							snake_case_    (		__lowercase						):






  @require_beam
  def 				UpperCAmelCase__						(							self			:					Optional[Any]      )->Union[str, Any]:






   '''simple docstring'''

   __lowerCAmelCase					:	Union[str, Any]      =       len(get_test_dummy_examples()      )
   with tempfile.TemporaryDirectory() as tmp_cache_dir:
    __lowerCAmelCase					:	Union[str, Any]      =       DummyBeamDataset(cache_dir=_snake_case       ,  beam_runner="""DirectRunner"""      )
    builder.download_and_prepare()
    self.assertTrue(
        os.path.exists(
            os.path.join(_snake_case       ,  builder.name       ,  """default"""       ,  """0.0.0"""       ,  F'''{builder.name}-train.arrow'''      )      )      )
    self.assertDictEqual(builder.info.features       ,  datasets.Features({"""content""": datasets.Value("""string"""      )}      )      )
    __lowerCAmelCase					:	Optional[Any]      =       builder.as_dataset()
    self.assertEqual(dset["""train"""].num_rows       ,  _snake_case      )
    self.assertEqual(dset["""train"""].info.splits["""train"""].num_examples       ,  _snake_case      )
    self.assertDictEqual(dset["""train"""][0]       ,  get_test_dummy_examples()[0][1]      )
    self.assertDictEqual(
        dset["""train"""][expected_num_examples - 1]       ,  get_test_dummy_examples()[expected_num_examples - 1][1]      )
    self.assertTrue(
        os.path.exists(os.path.join(_snake_case       ,  builder.name       ,  """default"""       ,  """0.0.0"""       ,  """dataset_info.json"""      )      )      )
    del dset






  @require_beam
  def 				UpperCAmelCase__						(							self			:					List[str]      )->List[str]:






   '''simple docstring'''

   import apache_beam as beam

   __lowerCAmelCase					:	Optional[Any]      =       beam.io.parquetio.WriteToParquet

   __lowerCAmelCase					:	int      =       len(get_test_dummy_examples()      )
   with tempfile.TemporaryDirectory() as tmp_cache_dir:
    __lowerCAmelCase					:	List[str]      =       DummyBeamDataset(cache_dir=_snake_case       ,  beam_runner="""DirectRunner"""      )
    with patch("""apache_beam.io.parquetio.WriteToParquet"""      ) as write_parquet_mock:
     __lowerCAmelCase					:	Any      =       partial(_snake_case       ,  num_shards=2      )
     builder.download_and_prepare()
    self.assertTrue(
        os.path.exists(
            os.path.join(
                _snake_case       ,  builder.name       ,  """default"""       ,  """0.0.0"""       ,  F'''{builder.name}-train-00000-of-00002.arrow'''      )      )      )
    self.assertTrue(
        os.path.exists(
            os.path.join(
                _snake_case       ,  builder.name       ,  """default"""       ,  """0.0.0"""       ,  F'''{builder.name}-train-00000-of-00002.arrow'''      )      )      )
    self.assertDictEqual(builder.info.features       ,  datasets.Features({"""content""": datasets.Value("""string"""      )}      )      )
    __lowerCAmelCase					:	Any      =       builder.as_dataset()
    self.assertEqual(dset["""train"""].num_rows       ,  _snake_case      )
    self.assertEqual(dset["""train"""].info.splits["""train"""].num_examples       ,  _snake_case      )
    # Order is not preserved when sharding, so we just check that all the elements are there
    self.assertListEqual(sorted(dset["""train"""]["""content"""]      )       ,  sorted(["""foo""", """bar""", """foobar"""]      )      )
    self.assertTrue(
        os.path.exists(os.path.join(_snake_case       ,  builder.name       ,  """default"""       ,  """0.0.0"""       ,  """dataset_info.json"""      )      )      )
    del dset






  @require_beam
  def 				UpperCAmelCase__						(							self			:					Union[str, Any]      )->Optional[Any]:






   '''simple docstring'''

   with tempfile.TemporaryDirectory() as tmp_cache_dir:
    __lowerCAmelCase					:	Any      =       DummyBeamDataset(cache_dir=_snake_case      )
    self.assertRaises(datasets.builder.MissingBeamOptions       ,  builder.download_and_prepare      )






  @require_beam
  def 				UpperCAmelCase__						(							self			:					Any      )->List[str]:






   '''simple docstring'''

   __lowerCAmelCase					:	Union[str, Any]      =       len(get_test_nested_examples()      )
   with tempfile.TemporaryDirectory() as tmp_cache_dir:
    __lowerCAmelCase					:	int      =       NestedBeamDataset(cache_dir=_snake_case       ,  beam_runner="""DirectRunner"""      )
    builder.download_and_prepare()
    self.assertTrue(
        os.path.exists(
            os.path.join(_snake_case       ,  builder.name       ,  """default"""       ,  """0.0.0"""       ,  F'''{builder.name}-train.arrow'''      )      )      )
    self.assertDictEqual(
        builder.info.features       ,  datasets.Features({"""a""": datasets.Sequence({"""b""": datasets.Value("""string"""      )}      )}      )      )
    __lowerCAmelCase					:	Dict      =       builder.as_dataset()
    self.assertEqual(dset["""train"""].num_rows       ,  _snake_case      )
    self.assertEqual(dset["""train"""].info.splits["""train"""].num_examples       ,  _snake_case      )
    self.assertDictEqual(dset["""train"""][0]       ,  get_test_nested_examples()[0][1]      )
    self.assertDictEqual(
        dset["""train"""][expected_num_examples - 1]       ,  get_test_nested_examples()[expected_num_examples - 1][1]      )
    self.assertTrue(
        os.path.exists(os.path.join(_snake_case       ,  builder.name       ,  """default"""       ,  """0.0.0"""       ,  """dataset_info.json"""      )      )      )
    del dset






from __future__ import annotations






def 	_SCREAMING_SNAKE_CASE		(      SCREAMING_SNAKE_CASE		:list[list[int]]   )      ->   bool:
 __lowerCAmelCase					:	Tuple      =       len(SCREAMING_SNAKE_CASE   )
 # We need to create solution object to save path.
 __lowerCAmelCase					:	str      =       [[0 for _ in range(SCREAMING_SNAKE_CASE   )] for _ in range(SCREAMING_SNAKE_CASE   )]
 __lowerCAmelCase					:	str      =       run_maze(SCREAMING_SNAKE_CASE   ,   0   ,   0   ,   SCREAMING_SNAKE_CASE   )
 if solved:
  print("""\n""".join(str(SCREAMING_SNAKE_CASE   ) for row in solutions   )   )
 else:
  print("""No solution exists!"""   )
 return solved






def 	_SCREAMING_SNAKE_CASE		(      SCREAMING_SNAKE_CASE		:list[list[int]]   ,   SCREAMING_SNAKE_CASE		:int   ,   SCREAMING_SNAKE_CASE		:int   ,   SCREAMING_SNAKE_CASE		:list[list[int]]   )      ->   bool:
 __lowerCAmelCase					:	List[str]      =       len(SCREAMING_SNAKE_CASE   )
 # Final check point.
 if i == j == (size - 1):
  __lowerCAmelCase					:	str      =       1
  return True

 __lowerCAmelCase					:	Optional[Any]      =       (not i < 0) and (not j < 0)  # Check lower bounds
 __lowerCAmelCase					:	Optional[Any]      =       (i < size) and (j < size)  # Check upper bounds

 if lower_flag and upper_flag:
  # check for already visited and block points.
  __lowerCAmelCase					:	int      =       (not solutions[i][j]) and (not maze[i][j])
  if block_flag:
   # check visited
   __lowerCAmelCase					:	Tuple      =       1

   # check for directions
   if (
       run_maze(SCREAMING_SNAKE_CASE   ,   i + 1   ,   SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   )
       or run_maze(SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   ,   j + 1   ,   SCREAMING_SNAKE_CASE   )
       or run_maze(SCREAMING_SNAKE_CASE   ,   i - 1   ,   SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   )
       or run_maze(SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   ,   j - 1   ,   SCREAMING_SNAKE_CASE   )
   ):
    return True

   __lowerCAmelCase					:	Tuple      =       0
   return False
 return False


if __name__ == "__main__":
      import doctest

      doctest.testmod()