






import collections
import gzip
import os
import urllib

import numpy
from tensorflow.python.framework import dtypes, random_seed
from tensorflow.python.platform import gfile
from tensorflow.python.util.deprecation import deprecated

__magic_name__:					str							=     collections.namedtuple("_Datasets", ["train", "validation", "test"])

# CVDF mirror of http://yann.lecun.com/exdb/mnist/
__magic_name__:					Dict							=     "https://storage.googleapis.com/cvdf-datasets/mnist/"

def    UpperCamelCase						(	_A			):






						"""simple docstring"""







						__magic_name__      :       Optional[int]       						= numpy.dtype(numpy.uintaa			).newbyteorder(""">"""			)
						return numpy.frombuffer(bytestream.read(4			),   dtype=_A			)[0]

@deprecated(_A,   """Please use tf.data to implement this functionality."""			)
def    UpperCamelCase						(	_A			):






						"""simple docstring"""







						print("""Extracting""",   f.name			)
						with gzip.GzipFile(fileobj=_A			) as bytestream:
												__magic_name__      :       Optional[Any]       						= _readaa(_A			)
												if magic != 2051:
																		raise ValueError(
																		    """Invalid magic number %d in MNIST image file: %s""" % (magic, f.name)			)
												__magic_name__      :       int       						= _readaa(_A			)
												__magic_name__      :       Tuple       						= _readaa(_A			)
												__magic_name__      :       str       						= _readaa(_A			)
												__magic_name__      :       Any       						= bytestream.read(rows * cols * num_images			)
												__magic_name__      :       List[str]       						= numpy.frombuffer(_A,   dtype=numpy.uinta			)
												__magic_name__      :       List[str]       						= data.reshape(_A,   _A,   _A,   1			)
												return data

@deprecated(_A,   """Please use tf.one_hot on tensors."""			)
def    UpperCamelCase						(	_A,   _A			):






						"""simple docstring"""







						__magic_name__      :       Any       						= labels_dense.shape[0]
						__magic_name__      :       Union[str, Any]       						= numpy.arange(_A			) * num_classes
						__magic_name__      :       str       						= numpy.zeros((num_labels, num_classes)			)
						__magic_name__      :       Optional[Any]       						= 1
						return labels_one_hot

@deprecated(_A,   """Please use tf.data to implement this functionality."""			)
def    UpperCamelCase						(	_A,   _A=False,   _A=10			):






						"""simple docstring"""







						print("""Extracting""",   f.name			)
						with gzip.GzipFile(fileobj=_A			) as bytestream:
												__magic_name__      :       List[str]       						= _readaa(_A			)
												if magic != 2049:
																		raise ValueError(
																		    """Invalid magic number %d in MNIST label file: %s""" % (magic, f.name)			)
												__magic_name__      :       Tuple       						= _readaa(_A			)
												__magic_name__      :       Union[str, Any]       						= bytestream.read(_A			)
												__magic_name__      :       Dict       						= numpy.frombuffer(_A,   dtype=numpy.uinta			)
												if one_hot:
																		return _dense_to_one_hot(_A,   _A			)
												return labels





class 	snake_case__ :
	@deprecated(
	    lowerCAmelCase__      ,       """Please use alternatives such as official/mnist/_DataSet.py"""
	    """ from tensorflow/models."""      ,       )
	def __init__(    self      ,       lowerCAmelCase__      ,       lowerCAmelCase__      ,       lowerCAmelCase__=False      ,       lowerCAmelCase__=False      ,       lowerCAmelCase__=dtypes.floataa      ,       lowerCAmelCase__=True      ,       lowerCAmelCase__=None      ,       )							->				int:
							__magic_name__						,__magic_name__      :       Dict       						= random_seed.get_seed(lowerCAmelCase__			)
							# If op level seed is not set, use whatever graph level seed is returned
							numpy.random.seed(seeda if seed is None else seeda			)
							__magic_name__      :       int       						= dtypes.as_dtype(lowerCAmelCase__			).base_dtype
							if dtype not in (dtypes.uinta, dtypes.floataa):
													raise TypeError("""Invalid image dtype %r, expected uint8 or float32""" % dtype			)
							if fake_data:
													__magic_name__      :       Optional[Any]       						= 1_00_00
													__magic_name__      :       str       						= one_hot
							else:
													assert (
													    images.shape[0] == labels.shape[0]
													), F'images.shape: {images.shape} labels.shape: {labels.shape}'
													__magic_name__      :       Dict       						= images.shape[0]

													# Convert shape from [num examples, rows, columns, depth]
													# to [num examples, rows*columns] (assuming depth == 1)
													if reshape:
																			assert images.shape[3] == 1
																			__magic_name__      :       List[str]       						= images.reshape(
																			    images.shape[0]      ,       images.shape[1] * images.shape[2]			)
													if dtype == dtypes.floataa:
																			# Convert from [0, 255] -> [0.0, 1.0].
																			__magic_name__      :       List[str]       						= images.astype(numpy.floataa			)
																			__magic_name__      :       int       						= numpy.multiply(lowerCAmelCase__      ,       1.0 / 2_5_5.0			)
							__magic_name__      :       Optional[Any]       						= images
							__magic_name__      :       Optional[Any]       						= labels
							__magic_name__      :       str       						= 0
							__magic_name__      :       Union[str, Any]       						= 0
	@property
	def       __magic_name__    (    self			)							->				Dict:
							return self._images
	@property
	def       __magic_name__    (    self			)							->				Any:
							return self._labels
	@property
	def       __magic_name__    (    self			)							->				str:
							return self._num_examples
	@property
	def       __magic_name__    (    self			)							->				Any:
							return self._epochs_completed







	def       __magic_name__    (    self      ,       lowerCAmelCase__      ,       lowerCAmelCase__=False      ,       lowerCAmelCase__=True			)							->				int:
							if fake_data:
													__magic_name__      :       Tuple       						= [1] * 7_84
													__magic_name__      :       Dict       						= [1] + [0] * 9 if self.one_hot else 0
													return (
													    [fake_image for _ in range(lowerCAmelCase__			)],
													    [fake_label for _ in range(lowerCAmelCase__			)],
													)
							__magic_name__      :       Any       						= self._index_in_epoch
							# Shuffle for the first epoch
							if self._epochs_completed == 0 and start == 0 and shuffle:
													__magic_name__      :       Union[str, Any]       						= numpy.arange(self._num_examples			)
													numpy.random.shuffle(lowerCAmelCase__			)
													__magic_name__      :       List[Any]       						= self.images[perma]
													__magic_name__      :       int       						= self.labels[perma]
							# Go to the next epoch
							if start + batch_size > self._num_examples:
													# Finished epoch
													self._epochs_completed += 1
													# Get the rest examples in this epoch
													__magic_name__      :       Any       						= self._num_examples - start
													__magic_name__      :       List[Any]       						= self._images[start : self._num_examples]
													__magic_name__      :       Optional[Any]       						= self._labels[start : self._num_examples]
													# Shuffle the data
													if shuffle:
																			__magic_name__      :       Optional[Any]       						= numpy.arange(self._num_examples			)
																			numpy.random.shuffle(lowerCAmelCase__			)
																			__magic_name__      :       int       						= self.images[perm]
																			__magic_name__      :       Dict       						= self.labels[perm]
													# Start next epoch
													__magic_name__      :       Tuple       						= 0
													__magic_name__      :       Optional[int]       						= batch_size - rest_num_examples
													__magic_name__      :       List[str]       						= self._index_in_epoch
													__magic_name__      :       Dict       						= self._images[start:end]
													__magic_name__      :       int       						= self._labels[start:end]
													return (
													    numpy.concatenate((images_rest_part, images_new_part)      ,       axis=0			),
													    numpy.concatenate((labels_rest_part, labels_new_part)      ,       axis=0			),
													)
							else:
													self._index_in_epoch += batch_size
													__magic_name__      :       str       						= self._index_in_epoch
													return self._images[start:end], self._labels[start:end]



@deprecated(_A,   """Please write your own downloading logic."""			)
def    UpperCamelCase						(	_A,   _A,   _A			):






						"""simple docstring"""







						if not gfile.Exists(_A			):
												gfile.MakeDirs(_A			)
						__magic_name__      :       Optional[int]       						= os.path.join(_A,   _A			)
						if not gfile.Exists(_A			):
												urllib.request.urlretrieve(_A,   _A			)  # noqa: S310
												with gfile.GFile(_A			) as f:
																		__magic_name__      :       Any       						= f.size()
												print("""Successfully downloaded""",   _A,   _A,   """bytes."""			)
						return filepath

@deprecated(
    _A,   """Please use alternatives such as:""" """ tensorflow_datasets.load('mnist')"""			)
def    UpperCamelCase						(	_A,   _A=False,   _A=False,   _A=dtypes.floataa,   _A=True,   _A=5000,   _A=None,   _A=DEFAULT_SOURCE_URL,   ):






						"""simple docstring"""







						if fake_data:

												def fake():
																		return _DataSet(
																		    [],   [],   fake_data=_A,   one_hot=_A,   dtype=_A,   seed=_A			)

												__magic_name__      :       Tuple       						= fake()
												__magic_name__      :       List[Any]       						= fake()
												__magic_name__      :       Optional[int]       						= fake()
												return _Datasets(train=_A,   validation=_A,   test=_A			)

						if not source_url:  # empty string check
												__magic_name__      :       str       						= DEFAULT_SOURCE_URL

						__magic_name__      :       Optional[Any]       						= """train-images-idx3-ubyte.gz"""
						__magic_name__      :       Any       						= """train-labels-idx1-ubyte.gz"""
						__magic_name__      :       Optional[Any]       						= """t10k-images-idx3-ubyte.gz"""
						__magic_name__      :       Union[str, Any]       						= """t10k-labels-idx1-ubyte.gz"""

						__magic_name__      :       str       						= _maybe_download(
						    _A,   _A,   source_url + train_images_file			)
						with gfile.Open(_A,   """rb"""			) as f:
												__magic_name__      :       Optional[int]       						= _extract_images(_A			)

						__magic_name__      :       Any       						= _maybe_download(
						    _A,   _A,   source_url + train_labels_file			)
						with gfile.Open(_A,   """rb"""			) as f:
												__magic_name__      :       Dict       						= _extract_labels(_A,   one_hot=_A			)

						__magic_name__      :       Optional[int]       						= _maybe_download(
						    _A,   _A,   source_url + test_images_file			)
						with gfile.Open(_A,   """rb"""			) as f:
												__magic_name__      :       str       						= _extract_images(_A			)

						__magic_name__      :       Optional[int]       						= _maybe_download(
						    _A,   _A,   source_url + test_labels_file			)
						with gfile.Open(_A,   """rb"""			) as f:
												__magic_name__      :       str       						= _extract_labels(_A,   one_hot=_A			)

						if not 0 <= validation_size <= len(_A			):
												__magic_name__      :       List[str]       						= (
												    """Validation size should be between 0 and """
												    f'{len(_A			)}. Received: {validation_size}.'
												)
												raise ValueError(_A			)

						__magic_name__      :       Tuple       						= train_images[:validation_size]
						__magic_name__      :       int       						= train_labels[:validation_size]
						__magic_name__      :       Any       						= train_images[validation_size:]
						__magic_name__      :       Union[str, Any]       						= train_labels[validation_size:]

						__magic_name__      :       List[Any]       						= {"""dtype""": dtype, """reshape""": reshape, """seed""": seed}

						__magic_name__      :       Dict       						= _DataSet(_A,   _A,   **_A			)
						__magic_name__      :       Optional[int]       						= _DataSet(_A,   _A,   **_A			)
						__magic_name__      :       Dict       						= _DataSet(_A,   _A,   **_A			)

						return _Datasets(train=_A,   validation=_A,   test=_A			)








import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModel,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableUnCLIPImgaImgPipeline, UNetaDConditionModel
from diffusers.pipelines.pipeline_utils import DiffusionPipeline
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    floats_tensor,
    load_image,
    load_numpy,
    require_torch_gpu,
    skip_mps,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()





class 	snake_case__ (							_lowerCAmelCase						,  _lowerCAmelCase						,  _lowerCAmelCase						,  unittest.TestCase				):
	lowercase__						:					str											=   StableUnCLIPImgaImgPipeline
	lowercase__						:					Any											=   TEXT_GUIDED_IMAGE_VARIATION_PARAMS
	lowercase__						:					Tuple											=   TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
	lowercase__						:					Union[str, Any]											=   frozenset(
	    []				)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
	lowercase__						:					Union[str, Any]											=   frozenset([]				)
	def       __magic_name__    (    self			)							->				Tuple:
							__magic_name__      :       Any       						= 32
							__magic_name__      :       Union[str, Any]       						= embedder_hidden_size

							# image encoding components

							__magic_name__      :       Optional[int]       						= CLIPImageProcessor(crop_size=32      ,       size=32			)

							torch.manual_seed(0			)
							__magic_name__      :       Optional[Any]       						= CLIPVisionModelWithProjection(
							    CLIPVisionConfig(
							        hidden_size=lowerCAmelCase__      ,       projection_dim=lowerCAmelCase__      ,       num_hidden_layers=5      ,       num_attention_heads=4      ,       image_size=32      ,       intermediate_size=37      ,       patch_size=1      ,       )			)

							# regular denoising components

							torch.manual_seed(0			)
							__magic_name__      :       Any       						= StableUnCLIPImageNormalizer(embedding_dim=lowerCAmelCase__			)
							__magic_name__      :       int       						= DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""			)

							torch.manual_seed(0			)
							__magic_name__      :       Tuple       						= CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""			)

							torch.manual_seed(0			)
							__magic_name__      :       Optional[Any]       						= CLIPTextModel(
							    CLIPTextConfig(
							        bos_token_id=0      ,       eos_token_id=2      ,       hidden_size=lowerCAmelCase__      ,       projection_dim=32      ,       intermediate_size=37      ,       layer_norm_eps=1e-0_5      ,       num_attention_heads=4      ,       num_hidden_layers=5      ,       pad_token_id=1      ,       vocab_size=10_00      ,       )			)

							torch.manual_seed(0			)
							__magic_name__      :       str       						= UNetaDConditionModel(
							    sample_size=32      ,       in_channels=4      ,       out_channels=4      ,       down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D""")      ,       up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D""")      ,       block_out_channels=(32, 64)      ,       attention_head_dim=(2, 4)      ,       class_embed_type="""projection"""      ,       projection_class_embeddings_input_dim=embedder_projection_dim * 2      ,       cross_attention_dim=lowerCAmelCase__      ,       layers_per_block=1      ,       upcast_attention=lowerCAmelCase__      ,       use_linear_projection=lowerCAmelCase__      ,       )

							torch.manual_seed(0			)
							__magic_name__      :       Optional[Any]       						= DDIMScheduler(
							    beta_schedule="""scaled_linear"""      ,       beta_start=0.0_0_0_8_5      ,       beta_end=0.0_1_2      ,       prediction_type="""v_prediction"""      ,       set_alpha_to_one=lowerCAmelCase__      ,       steps_offset=1      ,       )

							torch.manual_seed(0			)
							__magic_name__      :       List[str]       						= AutoencoderKL()

							__magic_name__      :       List[str]       						= {
							    # image encoding components
							    """feature_extractor""": feature_extractor,
							    """image_encoder""": image_encoder.eval(),
							    # image noising components
							    """image_normalizer""": image_normalizer.eval(),
							    """image_noising_scheduler""": image_noising_scheduler,
							    # regular denoising components
							    """tokenizer""": tokenizer,
							    """text_encoder""": text_encoder.eval(),
							    """unet""": unet.eval(),
							    """scheduler""": scheduler,
							    """vae""": vae.eval(),
							}

							return components
	def       __magic_name__    (    self      ,       lowerCAmelCase__      ,       lowerCAmelCase__=0      ,       lowerCAmelCase__=True			)							->				List[Any]:
							if str(lowerCAmelCase__			).startswith("""mps"""			):
													__magic_name__      :       Optional[int]       						= torch.manual_seed(lowerCAmelCase__			)
							else:
													__magic_name__      :       Optional[int]       						= torch.Generator(device=lowerCAmelCase__			).manual_seed(lowerCAmelCase__			)

							__magic_name__      :       Any       						= floats_tensor((1, 3, 32, 32)      ,       rng=random.Random(lowerCAmelCase__			)			).to(lowerCAmelCase__			)

							if pil_image:
													__magic_name__      :       Optional[Any]       						= input_image * 0.5 + 0.5
													__magic_name__      :       int       						= input_image.clamp(0      ,       1			)
													__magic_name__      :       Union[str, Any]       						= input_image.cpu().permute(0      ,       2      ,       3      ,       1			).float().numpy()
													__magic_name__      :       Optional[int]       						= DiffusionPipeline.numpy_to_pil(lowerCAmelCase__			)[0]

							return {
							    "prompt": "An anime racoon running a marathon",
							    "image": input_image,
							    "generator": generator,
							    "num_inference_steps": 2,
							    "output_type": "np",
							}
	@skip_mps
	def       __magic_name__    (    self			)							->				List[Any]:
							__magic_name__      :       Optional[Any]       						= """cpu"""  # ensure determinism for the device-dependent torch.Generator
							__magic_name__      :       List[str]       						= self.get_dummy_components()
							__magic_name__      :       int       						= StableUnCLIPImgaImgPipeline(**lowerCAmelCase__			)
							__magic_name__      :       List[str]       						= sd_pipe.to(lowerCAmelCase__			)
							sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__			)

							__magic_name__      :       Any       						= self.get_dummy_inputs(lowerCAmelCase__			)
							inputs.update({"""image_embeds""": None}			)
							__magic_name__      :       List[str]       						= sd_pipe(**lowerCAmelCase__			).images
							__magic_name__      :       Dict       						= image[0, -3:, -3:, -1]

							assert image.shape == (1, 32, 32, 3)
							__magic_name__      :       int       						= np.array([0.3_8_7_2, 0.7_2_2_4, 0.5_6_0_1, 0.4_7_4_1, 0.6_8_7_2, 0.5_8_1_4, 0.4_6_3_6, 0.3_8_6_7, 0.5_0_7_8]			)

							assert np.abs(image_slice.flatten() - expected_slice			).max() < 1e-3
	def       __magic_name__    (    self			)							->				Dict:
							__magic_name__      :       int       						= torch_device in ["""cpu""", """mps"""]

							self._test_attention_slicing_forward_pass(test_max_difference=lowerCAmelCase__			)
	def       __magic_name__    (    self			)							->				Union[str, Any]:
							__magic_name__      :       Tuple       						= torch_device in ["""cpu""", """mps"""]

							self._test_inference_batch_single_identical(test_max_difference=lowerCAmelCase__			)







	@unittest.skipIf(
	    torch_device != """cuda""" or not is_xformers_available()      ,       reason="""XFormers attention is only available with CUDA and `xformers` installed"""      ,       )
	def       __magic_name__    (    self			)							->				str:
							self._test_xformers_attention_forwardGenerator_pass(test_max_difference=lowerCAmelCase__			)





@slow
@require_torch_gpu
class 	snake_case__ (							unittest.TestCase				):
	def       __magic_name__    (    self			)							->				int:
							# clean up the VRAM after each test
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()
	def       __magic_name__    (    self			)							->				List[Any]:
							__magic_name__      :       Optional[Any]       						= load_image(
							    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""			)

							__magic_name__      :       int       						= load_numpy(
							    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_img2img_anime_turtle_fp16.npy"""			)

							__magic_name__      :       List[Any]       						= StableUnCLIPImgaImgPipeline.from_pretrained(
							    """fusing/stable-unclip-2-1-l-img2img"""      ,       torch_dtype=torch.floataa			)
							pipe.to(lowerCAmelCase__			)
							pipe.set_progress_bar_config(disable=lowerCAmelCase__			)
							# stable unclip will oom when integration tests are run on a V100,
							# so turn on memory savings
							pipe.enable_attention_slicing()
							pipe.enable_sequential_cpu_offload()

							__magic_name__      :       Union[str, Any]       						= torch.Generator(device="""cpu"""			).manual_seed(0			)
							__magic_name__      :       Union[str, Any]       						= pipe(lowerCAmelCase__      ,       """anime turle"""      ,       generator=lowerCAmelCase__      ,       output_type="""np"""			)

							__magic_name__      :       Optional[Any]       						= output.images[0]

							assert image.shape == (7_68, 7_68, 3)

							assert_mean_pixel_difference(lowerCAmelCase__      ,       lowerCAmelCase__			)
	def       __magic_name__    (    self			)							->				List[Any]:
							__magic_name__      :       Union[str, Any]       						= load_image(
							    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""			)

							__magic_name__      :       Dict       						= load_numpy(
							    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_h_img2img_anime_turtle_fp16.npy"""			)

							__magic_name__      :       Union[str, Any]       						= StableUnCLIPImgaImgPipeline.from_pretrained(
							    """fusing/stable-unclip-2-1-h-img2img"""      ,       torch_dtype=torch.floataa			)
							pipe.to(lowerCAmelCase__			)
							pipe.set_progress_bar_config(disable=lowerCAmelCase__			)
							# stable unclip will oom when integration tests are run on a V100,
							# so turn on memory savings
							pipe.enable_attention_slicing()
							pipe.enable_sequential_cpu_offload()

							__magic_name__      :       List[str]       						= torch.Generator(device="""cpu"""			).manual_seed(0			)
							__magic_name__      :       Any       						= pipe(lowerCAmelCase__      ,       """anime turle"""      ,       generator=lowerCAmelCase__      ,       output_type="""np"""			)

							__magic_name__      :       Any       						= output.images[0]

							assert image.shape == (7_68, 7_68, 3)

							assert_mean_pixel_difference(lowerCAmelCase__      ,       lowerCAmelCase__			)







	def       __magic_name__    (    self			)							->				str:
							__magic_name__      :       Any       						= load_image(
							    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""			)

							torch.cuda.empty_cache()
							torch.cuda.reset_max_memory_allocated()
							torch.cuda.reset_peak_memory_stats()

							__magic_name__      :       Union[str, Any]       						= StableUnCLIPImgaImgPipeline.from_pretrained(
							    """fusing/stable-unclip-2-1-h-img2img"""      ,       torch_dtype=torch.floataa			)
							__magic_name__      :       int       						= pipe.to(lowerCAmelCase__			)
							pipe.set_progress_bar_config(disable=lowerCAmelCase__			)
							pipe.enable_attention_slicing()
							pipe.enable_sequential_cpu_offload()

							__magic_name__      :       List[Any]       						= pipe(
							    lowerCAmelCase__      ,       """anime turtle"""      ,       num_inference_steps=2      ,       output_type="""np"""      ,       )

							__magic_name__      :       List[str]       						= torch.cuda.max_memory_allocated()
							# make sure that less than 7 GB is allocated
							assert mem_bytes < 7 * 10**9
