



'''simple docstring'''




from typing import Optional

from torch import nn

from .transformer_ad import TransformeraDModel, TransformeraDModelOutput






class 		_UpperCamelCase						(							nn.Module					):


			'''simple docstring'''

			def __init__(							self  :				str							,  lowerCAmelCase__  :				int = 1_6							,  lowerCAmelCase__  :				int = 8_8							,  lowerCAmelCase__  :				Optional[int] = None							,  lowerCAmelCase__  :				int = 1							,  lowerCAmelCase__  :				float = 0.0							,  lowerCAmelCase__  :				int = 3_2							,  lowerCAmelCase__  :				Optional[int] = None							,  lowerCAmelCase__  :				bool = False							,  lowerCAmelCase__  :				Optional[int] = None							,  lowerCAmelCase__  :				Optional[int] = None							,  lowerCAmelCase__  :				str = "geglu"							,  lowerCAmelCase__  :				Optional[int] = None							,  ):





					"""simple docstring"""






					super().__init__()
					__SCREAMING_SNAKE_CASE :	Optional[int]					 =	nn.ModuleList(
					    [
					        TransformeraDModel(
					            num_attention_heads=lowerCAmelCase__							,  attention_head_dim=lowerCAmelCase__							,  in_channels=lowerCAmelCase__							,  num_layers=lowerCAmelCase__							,  dropout=lowerCAmelCase__							,  norm_num_groups=lowerCAmelCase__							,  cross_attention_dim=lowerCAmelCase__							,  attention_bias=lowerCAmelCase__							,  sample_size=lowerCAmelCase__							,  num_vector_embeds=lowerCAmelCase__							,  activation_fn=lowerCAmelCase__							,  num_embeds_ada_norm=lowerCAmelCase__							,  )
					        for _ in range(2		)
					    ]		)

					# Variables that can be set by a pipeline:

					# The ratio of transformer1 to transformer2's output states to be combined during inference
					__SCREAMING_SNAKE_CASE :	Union[str, Any]					 =	0.5

					# The shape of `encoder_hidden_states` is expected to be
					# `(batch_size, condition_lengths[0]+condition_lengths[1], num_features)`
					__SCREAMING_SNAKE_CASE :	int					 =	[7_7, 2_5_7]

					# Which transformer to use to encode which condition.
					# E.g. `(1, 0)` means that we'll use `transformers[1](conditions[0])` and `transformers[0](conditions[1])`
					__SCREAMING_SNAKE_CASE :	Tuple					 =	[1, 0]



			def UpperCamelCase__      (							self  :				Any							,  lowerCAmelCase__  :				Optional[int]							,  lowerCAmelCase__  :				List[str]							,  lowerCAmelCase__  :				Tuple=None							,  lowerCAmelCase__  :				Tuple=None							,  lowerCAmelCase__  :				List[str]=None							,  lowerCAmelCase__  :				bool = True							,  ):





					"""simple docstring"""






					__SCREAMING_SNAKE_CASE :	Optional[Any]					 =	hidden_states

					__SCREAMING_SNAKE_CASE :	List[Any]					 =	[]
					__SCREAMING_SNAKE_CASE :	str					 =	0
					# attention_mask is not used yet
					for i in range(2		):
							# for each of the two transformers, pass the corresponding condition tokens
							__SCREAMING_SNAKE_CASE :	Tuple					 =	encoder_hidden_states[:, tokens_start : tokens_start + self.condition_lengths[i]]
							__SCREAMING_SNAKE_CASE :	Optional[Any]					 =	self.transformer_index_for_condition[i]
							__SCREAMING_SNAKE_CASE :	Optional[int]					 =	self.transformers[transformer_index](
							    lowerCAmelCase__							,  encoder_hidden_states=lowerCAmelCase__							,  timestep=lowerCAmelCase__							,  cross_attention_kwargs=lowerCAmelCase__							,  return_dict=lowerCAmelCase__							,  )[0]
							encoded_states.append(encoded_state - input_states		)
							tokens_start += self.condition_lengths[i]

					__SCREAMING_SNAKE_CASE :	str					 =	encoded_states[0] * self.mix_ratio + encoded_states[1] * (1 - self.mix_ratio)
					__SCREAMING_SNAKE_CASE :	List[str]					 =	output_states + input_states

					if not return_dict:
							return (output_states,)

					return TransformeraDModelOutput(sample=lowerCAmelCase__		)







'''simple docstring'''




from unittest.mock import Mock, patch

from file_transfer.send_file import send_file


@patch("""socket.socket"""      )
@patch("""builtins.open"""      )
def        lowerCAmelCase_     (			_lowerCamelCase:     Dict ,   _lowerCamelCase:     List[Any]      ):
		# ===== initialization =====
		__SCREAMING_SNAKE_CASE :	Optional[int]					 =	Mock()
		__SCREAMING_SNAKE_CASE :	Optional[int]					 =	conn, Mock()
		__SCREAMING_SNAKE_CASE :	str					 =	iter([1, None]      )
		__SCREAMING_SNAKE_CASE :	Any					 =	lambda _lowerCamelCase      : next(_lowerCamelCase      )

		# ===== invoke =====
		send_file(filename="""mytext.txt""" ,   testing=_lowerCamelCase      )

		# ===== ensurance =====
		sock.assert_called_once()
		sock.return_value.bind.assert_called_once()
		sock.return_value.listen.assert_called_once()
		sock.return_value.accept.assert_called_once()
		conn.recv.assert_called_once()

		file.return_value.__enter__.assert_called_once()
		file.return_value.__enter__.return_value.read.assert_called()

		conn.send.assert_called_once()
		conn.close.assert_called_once()
		sock.return_value.shutdown.assert_called_once()
		sock.return_value.close.assert_called_once()