





'''simple docstring'''



import logging
import math
from functools import partial
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union

import torch

from .tensor_utils import tensor_tree_map, tree_map




def 				snake_case_      (		SCREAMING_SNAKE_CASE__       ):







  '''simple docstring'''



  _snake_case     		=				[]
  if isinstance(SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       ):
    for v in tree.values():
      shapes.extend(_fetch_dims(SCREAMING_SNAKE_CASE__       )       )
  elif isinstance(SCREAMING_SNAKE_CASE__					,	(list, tuple)       ):
    for t in tree:
      shapes.extend(_fetch_dims(SCREAMING_SNAKE_CASE__       )       )
  elif isinstance(SCREAMING_SNAKE_CASE__					,	torch.Tensor       ):
    shapes.append(tree.shape       )
  else:
    raise ValueError("Not supported"       )

  return shapes




@torch.jit.ignore
def 				snake_case_      (		SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       ):







  '''simple docstring'''



  _snake_case     		=				[]
  for d in reversed(SCREAMING_SNAKE_CASE__       ):
    idx.append(flat_idx % d       )
    _snake_case     		=				flat_idx // d

  return tuple(reversed(SCREAMING_SNAKE_CASE__       )       )




@torch.jit.ignore
def 				snake_case_      (		SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__ = None					,	SCREAMING_SNAKE_CASE__ = None					,	):







  '''simple docstring'''



  def reduce_edge_list(SCREAMING_SNAKE_CASE__       ) -> None:
    _snake_case     		=				True
    for i in range(len(SCREAMING_SNAKE_CASE__       )       ):
      _snake_case     		=				-1 * (i + 1)
      l[reversed_idx] &= tally
      _snake_case     		=				l[reversed_idx]

  if start_edges is None:
    _snake_case     		=				[s == 0 for s in start]
    reduce_edge_list(SCREAMING_SNAKE_CASE__       )
  if end_edges is None:
    _snake_case     		=				[e == (d - 1) for e, d in zip(SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       )]
    reduce_edge_list(SCREAMING_SNAKE_CASE__       )

  # Base cases. Either start/end are empty and we're done, or the final,
  # one-dimensional tensor can be simply sliced
  if len(SCREAMING_SNAKE_CASE__       ) == 0:
    return [()]
  elif len(SCREAMING_SNAKE_CASE__       ) == 1:
    return [(slice(start[0]					,	end[0] + 1       ),)]

  _snake_case     		=				[]
  _snake_case     		=				[]

  # Dimensions common to start and end can be selected directly
  for s, e in zip(SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       ):
    if s == e:
      path_list.append(slice(SCREAMING_SNAKE_CASE__					,	s + 1       )       )
    else:
      break

  _snake_case     		=				tuple(SCREAMING_SNAKE_CASE__       )
  _snake_case     		=				len(SCREAMING_SNAKE_CASE__       )

  # start == end, and we're done
  if divergence_idx == len(SCREAMING_SNAKE_CASE__       ):
    return [path]

  def upper() -> Tuple[Tuple[slice, ...], ...]:
    assert start_edges is not None
    assert end_edges is not None

    _snake_case     		=				start[divergence_idx]
    return tuple(
        path + (slice(SCREAMING_SNAKE_CASE__					,	sdi + 1       ),) + s
        for s in _get_minimal_slice_set(
            start[divergence_idx + 1 :]					,	[d - 1 for d in dims[divergence_idx + 1 :]]					,	dims[divergence_idx + 1 :]					,	start_edges=start_edges[divergence_idx + 1 :]					,	end_edges=[True for _ in end_edges[divergence_idx + 1 :]]					,	)       )

  def lower() -> Tuple[Tuple[slice, ...], ...]:
    assert start_edges is not None
    assert end_edges is not None

    _snake_case     		=				end[divergence_idx]
    return tuple(
        path + (slice(SCREAMING_SNAKE_CASE__					,	edi + 1       ),) + s
        for s in _get_minimal_slice_set(
            [0 for _ in start[divergence_idx + 1 :]]					,	end[divergence_idx + 1 :]					,	dims[divergence_idx + 1 :]					,	start_edges=[True for _ in start_edges[divergence_idx + 1 :]]					,	end_edges=end_edges[divergence_idx + 1 :]					,	)       )

  # If both start and end are at the edges of the subtree rooted at
  # divergence_idx, we can just select the whole subtree at once
  if start_edges[divergence_idx] and end_edges[divergence_idx]:
    slices.append(path + (slice(start[divergence_idx]					,	end[divergence_idx] + 1       ),)       )
  # If just start is at the edge, we can grab almost all of the subtree,
  # treating only the ragged bottom edge as an edge case
  elif start_edges[divergence_idx]:
    slices.append(path + (slice(start[divergence_idx]					,	end[divergence_idx]       ),)       )
    slices.extend(lower()       )
  # Analogous to the previous case, but the top is ragged this time
  elif end_edges[divergence_idx]:
    slices.extend(upper()       )
    slices.append(path + (slice(start[divergence_idx] + 1					,	end[divergence_idx] + 1       ),)       )
  # If both sides of the range are ragged, we need to handle both sides
  # separately. If there's contiguous meat in between them, we can index it
  # in one big chunk
  else:
    slices.extend(upper()       )
    _snake_case     		=				end[divergence_idx] - start[divergence_idx]
    if middle_ground > 1:
      slices.append(path + (slice(start[divergence_idx] + 1					,	end[divergence_idx]       ),)       )
    slices.extend(lower()       )

  return slices




@torch.jit.ignore
def 				snake_case_      (		SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       ):







  '''simple docstring'''



  _snake_case     		=				t.shape[:no_batch_dims]
  _snake_case     		=				list(_flat_idx_to_idx(SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       )       )
  # _get_minimal_slice_set is inclusive
  _snake_case     		=				list(_flat_idx_to_idx(flat_end - 1					,	SCREAMING_SNAKE_CASE__       )       )

  # Get an ordered list of slices to perform
  _snake_case     		=				_get_minimal_slice_set(
      SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__					,	)

  _snake_case     		=				[t[s] for s in slices]

  return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]       ) for s in sliced_tensors]       )




def 				snake_case_      (		SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__ = False					,	SCREAMING_SNAKE_CASE__ = None					,	SCREAMING_SNAKE_CASE__ = False					,	):







  '''simple docstring'''



  if not (len(SCREAMING_SNAKE_CASE__       ) > 0):
    raise ValueError("Must provide at least one input"       )

  _snake_case     		=				[shape[:no_batch_dims] for shape in _fetch_dims(SCREAMING_SNAKE_CASE__       )]
  _snake_case     		=				tuple([max(SCREAMING_SNAKE_CASE__       ) for s in zip(*SCREAMING_SNAKE_CASE__       )]       )

  def _prep_inputs(SCREAMING_SNAKE_CASE__       ) -> torch.Tensor:
    if not low_mem:
      if not sum(t.shape[:no_batch_dims]       ) == no_batch_dims:
        _snake_case     		=				t.expand(orig_batch_dims + t.shape[no_batch_dims:]       )
      _snake_case     		=				t.reshape(-1					,	*t.shape[no_batch_dims:]       )
    else:
      _snake_case     		=				t.expand(orig_batch_dims + t.shape[no_batch_dims:]       )
    return t

  _snake_case     		=				tensor_tree_map(_prep_inputs					,	SCREAMING_SNAKE_CASE__       )
  _snake_case     		=				None
  if _out is not None:
    _snake_case     		=				tensor_tree_map(lambda SCREAMING_SNAKE_CASE__       : t.view([-1] + list(t.shape[no_batch_dims:]       )       )					,	_out       )

  _snake_case     		=				1
  for d in orig_batch_dims:
    flat_batch_dim *= d

  _snake_case     		=				flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

  def _select_chunk(SCREAMING_SNAKE_CASE__       ) -> torch.Tensor:
    return t[i : i + chunk_size] if t.shape[0] != 1 else t

  _snake_case     		=				0
  _snake_case     		=				prepped_outputs
  for _ in range(SCREAMING_SNAKE_CASE__       ):
    # Chunk the input
    if not low_mem:
      _snake_case     		=				_select_chunk
    else:
      _snake_case     		=				partial(
          _chunk_slice					,	flat_start=SCREAMING_SNAKE_CASE__					,	flat_end=min(SCREAMING_SNAKE_CASE__					,	i + chunk_size       )					,	no_batch_dims=len(SCREAMING_SNAKE_CASE__       )					,	)

    _snake_case     		=				tensor_tree_map(SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       )

    # Run the layer on the chunk
    _snake_case     		=				layer(**SCREAMING_SNAKE_CASE__       )

    # Allocate space for the output
    if out is None:
      _snake_case     		=				tensor_tree_map(lambda SCREAMING_SNAKE_CASE__       : t.new_zeros((flat_batch_dim,) + t.shape[1:]       )					,	SCREAMING_SNAKE_CASE__       )

    # Put the chunk in its pre-allocated space
    if isinstance(SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       ):

      def assign(SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       ) -> None:
        for k, v in da.items():
          if isinstance(SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       ):
            assign(SCREAMING_SNAKE_CASE__					,	da[k]       )
          else:
            if _add_into_out:
              v[i : i + chunk_size] += da[k]
            else:
              _snake_case     		=				da[k]

      assign(SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       )
    elif isinstance(SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       ):
      for xa, xa in zip(SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       ):
        if _add_into_out:
          xa[i : i + chunk_size] += xa
        else:
          _snake_case     		=				xa
    elif isinstance(SCREAMING_SNAKE_CASE__					,	torch.Tensor       ):
      if _add_into_out:
        out[i : i + chunk_size] += output_chunk
      else:
        _snake_case     		=				output_chunk
    else:
      raise ValueError("Not supported"       )

    i += chunk_size

  _snake_case     		=				tensor_tree_map(lambda SCREAMING_SNAKE_CASE__       : t.view(orig_batch_dims + t.shape[1:]       )					,	SCREAMING_SNAKE_CASE__       )

  return out




class       __SCREAMING_SNAKE_CASE	:



     '''simple docstring'''






     def __init__(					self						,   lowerCamelCase = 512						,   ):
       _snake_case     		=				max_chunk_size
       _snake_case     		=				None
       _snake_case     		=				None

     def 	UpperCamelCase(					self						,   lowerCamelCase						,   lowerCamelCase						,   lowerCamelCase	):
       logging.info("Tuning chunk size..."	)

       if min_chunk_size >= self.max_chunk_size:
         return min_chunk_size

       _snake_case     		=				[2**l for l in range(int(math.log(self.max_chunk_size						,   2	)	) + 1	)]
       _snake_case     		=				[c for c in candidates if c > min_chunk_size]
       _snake_case     		=				[min_chunk_size] + candidates
       candidates[-1] += 4

       def test_chunk_size(lowerCamelCase	) -> bool:
         try:
           with torch.no_grad():
             fn(*lowerCamelCase						,   chunk_size=lowerCamelCase	)
           return True
         except RuntimeError:
           return False

       _snake_case     		=				0
       _snake_case     		=				len(lowerCamelCase	) - 1
       while i > min_viable_chunk_size_index:
         _snake_case     		=				test_chunk_size(candidates[i]	)
         if not viable:
           _snake_case     		=				(min_viable_chunk_size_index + i) // 2
         else:
           _snake_case     		=				i
           _snake_case     		=				(i + len(lowerCamelCase	) - 1) // 2

       return candidates[min_viable_chunk_size_index]

     def 	UpperCamelCase(					self						,   lowerCamelCase						,   lowerCamelCase	):
       _snake_case     		=				True
       for aa, aa in zip(lowerCamelCase						,   lowerCamelCase	):
         assert type(lowerCamelCase	) == type(lowerCamelCase	)
         if isinstance(lowerCamelCase						,   (list, tuple)	):
           consistent &= self._compare_arg_caches(lowerCamelCase						,   lowerCamelCase	)
         elif isinstance(lowerCamelCase						,   lowerCamelCase	):
           _snake_case     		=				[v for _, v in sorted(aa.items()						,   key=lambda lowerCamelCase	: x[0]	)]
           _snake_case     		=				[v for _, v in sorted(aa.items()						,   key=lambda lowerCamelCase	: x[0]	)]
           consistent &= self._compare_arg_caches(lowerCamelCase						,   lowerCamelCase	)
         else:
           consistent &= aa == aa

       return consistent






     def 	UpperCamelCase(					self						,   lowerCamelCase						,   lowerCamelCase						,   lowerCamelCase						,   ):
       _snake_case     		=				True
       _snake_case     		=				tree_map(lambda lowerCamelCase	: a.shape if isinstance(lowerCamelCase						,   torch.Tensor	) else a						,   lowerCamelCase						,   lowerCamelCase	)
       if self.cached_arg_data is not None:
         # If args have changed shape/value, we need to re-tune
         assert len(self.cached_arg_data	) == len(lowerCamelCase	)
         _snake_case     		=				self._compare_arg_caches(self.cached_arg_data						,   lowerCamelCase	)
       else:
         # Otherwise, we can reuse the precomputed value
         _snake_case     		=				False

       if not consistent:
         _snake_case     		=				self._determine_favorable_chunk_size(
             lowerCamelCase						,   lowerCamelCase						,   lowerCamelCase						,   )
         _snake_case     		=				arg_data

       assert self.cached_chunk_size is not None

       return self.cached_chunk_size









'''simple docstring'''



import math


class       __SCREAMING_SNAKE_CASE	:



     '''simple docstring'''






     def 	UpperCamelCase(					self						,   lowerCamelCase						,   lowerCamelCase	):
       _snake_case     		=				0.0
       _snake_case     		=				0.0
       for i in range(len(lowerCamelCase	)	):
         da += math.pow((sample[i] - weights[0][i])						,   2	)
         da += math.pow((sample[i] - weights[1][i])						,   2	)
         return 0 if da > da else 1
       return 0






     def 	UpperCamelCase(					self						,   lowerCamelCase						,   lowerCamelCase						,   lowerCamelCase						,   lowerCamelCase	):
       for i in range(len(lowerCamelCase	)	):
         weights[j][i] += alpha * (sample[i] - weights[j][i])
       return weights







def 				snake_case_      (		):







  '''simple docstring'''



  _snake_case     		=				[[1, 1, 0, 0], [0, 0, 0, 1], [1, 0, 0, 0], [0, 0, 1, 1]]

  # weight initialization ( n, C )
  _snake_case     		=				[[0.2, 0.6, 0.5, 0.9], [0.8, 0.4, 0.7, 0.3]]

  # training
  _snake_case     		=				SelfOrganizingMap()
  _snake_case     		=				3
  _snake_case     		=				0.5

  for _ in range(SCREAMING_SNAKE_CASE__       ):
    for j in range(len(SCREAMING_SNAKE_CASE__       )       ):
      # training sample
      _snake_case     		=				training_samples[j]

      # Compute the winning vector
      _snake_case     		=				self_organizing_map.get_winner(SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       )

      # Update the winning vector
      _snake_case     		=				self_organizing_map.update(SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       )

    # classify test sample
  _snake_case     		=				[0, 0, 0, 1]
  _snake_case     		=				self_organizing_map.get_winner(SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__       )

  # results
  print(f'''Clusters that the test sample belongs to : {winner}'''       )
  print(f'''Weights that have been trained : {weights}'''       )


# running the main() function
if __name__ == "__main__":
 main()


