
'''simple docstring'''







import json
import os
import tempfile

import datasets
from utils import generate_example_dataset, get_duration


lowercase    :							Any      =					5_0000
lowercase    :							str      =					5000

lowercase      ,  lowercase    :							List[Any]      =					os.path.split(__file__)
lowercase    :							List[str]      =					os.path.join(RESULTS_BASEPATH, "results", RESULTS_FILENAME.replace(".py", ".json"))







@get_duration
def 				SCREAMING_SNAKE_CASE__	(   __A			,				__A	)       ->	Optional[Any]:
	for i in range(__A	):
		_snake_case   						=		dataset[i]







@get_duration
def 				SCREAMING_SNAKE_CASE__	(   __A			,				__A			,				__A	)       ->	str:
	for i in range(0			,				len(__A	)			,				__A	):
		_snake_case   						=		dataset[i : i + batch_size]







@get_duration
def 				SCREAMING_SNAKE_CASE__	(   __A			,				__A			,				__A	)       ->	Optional[int]:
	with dataset.formatted_as(type=__A	):
		for i in range(__A	):
			_snake_case   						=		dataset[i]







@get_duration
def 				SCREAMING_SNAKE_CASE__	(   __A			,				__A			,				__A			,				__A	)       ->	Dict:
	with dataset.formatted_as(type=__A	):
		for i in range(0			,				__A			,				__A	):
			_snake_case   						=		dataset[i : i + batch_size]







def 				SCREAMING_SNAKE_CASE__	(   )       ->	Union[str, Any]:
	_snake_case   						=		{'num examples': SPEED_TEST_N_EXAMPLES}
	_snake_case   						=		[
	    (read, {'length': SMALL_TEST}),
	    (read, {'length': SPEED_TEST_N_EXAMPLES}),
	    (read_batch, {'length': SPEED_TEST_N_EXAMPLES, 'batch_size': 10}),
	    (read_batch, {'length': SPEED_TEST_N_EXAMPLES, 'batch_size': 100}),
	    (read_batch, {'length': SPEED_TEST_N_EXAMPLES, 'batch_size': 1_000}),
	    (read_formatted, {'type': 'numpy', 'length': SMALL_TEST}),
	    (read_formatted, {'type': 'pandas', 'length': SMALL_TEST}),
	    (read_formatted, {'type': 'torch', 'length': SMALL_TEST}),
	    (read_formatted, {'type': 'tensorflow', 'length': SMALL_TEST}),
	    (read_formatted_batch, {'type': 'numpy', 'length': SMALL_TEST, 'batch_size': 10}),
	    (read_formatted_batch, {'type': 'numpy', 'length': SMALL_TEST, 'batch_size': 1_000}),
	]

	_snake_case   						=		[
	    (read, {'length': SMALL_TEST}),
	    (read, {'length': SPEED_TEST_N_EXAMPLES}),
	    (read_batch, {'length': SPEED_TEST_N_EXAMPLES, 'batch_size': 10}),
	    (read_batch, {'length': SPEED_TEST_N_EXAMPLES, 'batch_size': 100}),
	    (read_batch, {'length': SPEED_TEST_N_EXAMPLES, 'batch_size': 1_000}),
	    (read_formatted, {'type': 'numpy', 'length': SMALL_TEST}),
	    (read_formatted_batch, {'type': 'numpy', 'length': SMALL_TEST, 'batch_size': 10}),
	    (read_formatted_batch, {'type': 'numpy', 'length': SMALL_TEST, 'batch_size': 1_000}),
	]
	with tempfile.TemporaryDirectory() as tmp_dir:
		print('generating dataset'	)
		_snake_case   						=		datasets.Features(
		    {'list': datasets.Sequence(datasets.Value('float32'	)	), 'numbers': datasets.Value('float32'	)}	)
		_snake_case   						=		generate_example_dataset(
		    os.path.join(__A			,				'dataset.arrow'	)			,				__A			,				num_examples=__A			,				seq_shapes={'list': (100,)}			,				)
		print('first set of iterations'	)
		for func, kwargs in functions:
			print(func.__name__			,				str(__A	)	)
			_snake_case   						=		func(__A			,				**__A	)

		print('shuffling dataset'	)
		_snake_case   						=		dataset.shuffle()
		print('Second set of iterations (after shuffling'	)
		for func, kwargs in functions_shuffled:
			print('shuffled '			,				func.__name__			,				str(__A	)	)
			_snake_case   						=		func(
			    __A			,				**__A	)

	with open(__A			,				'wb'	) as f:
		f.write(json.dumps(__A	).encode('utf-8'	)	)


if __name__ == "__main__":  # useful to run the profiler
					benchmark_iterating()



'''simple docstring'''







import argparse

import intel_extension_for_pytorch as ipex
import torch

from diffusers import DPMSolverMultistepScheduler, StableDiffusionPipeline


lowercase    :							Optional[Any]      =					argparse.ArgumentParser("Stable Diffusion script with intel optimization", add_help=False)
parser.add_argument("--dpm", action="store_true", help="Enable DPMSolver or not")
parser.add_argument("--steps", default=None, type=int, help="Num inference steps")
lowercase    :							Tuple      =					parser.parse_args()


lowercase    :							Optional[int]      =					"cpu"
lowercase    :							Optional[Any]      =					"a lovely <dicoo> in red dress and hat, in the snowly and brightly night, with many brighly buildings"

lowercase    :							Optional[int]      =					"path-to-your-trained-model"
lowercase    :							List[str]      =					StableDiffusionPipeline.from_pretrained(model_id)
if args.dpm:
					lowercase    :							str      =					DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
lowercase    :							Dict      =					pipe.to(device)

# to channels last
lowercase    :							Optional[Any]      =					pipe.unet.to(memory_format=torch.channels_last)
lowercase    :							int      =					pipe.vae.to(memory_format=torch.channels_last)
lowercase    :							Optional[Any]      =					pipe.text_encoder.to(memory_format=torch.channels_last)
if pipe.requires_safety_checker:
					lowercase    :							Optional[int]      =					pipe.safety_checker.to(memory_format=torch.channels_last)

# optimize with ipex
lowercase    :							Any      =					torch.randn(2, 4, 64, 64)
lowercase    :							Optional[int]      =					torch.rand(1) * 999
lowercase    :							Optional[Any]      =					torch.randn(2, 77, 768)
lowercase    :							Optional[Any]      =					(sample, timestep, encoder_hidden_status)
try:
					lowercase    :							List[Any]      =					ipex.optimize(pipe.unet.eval(), dtype=torch.bfloataa, inplace=True, sample_input=input_example)
except Exception:
					lowercase    :							List[str]      =					ipex.optimize(pipe.unet.eval(), dtype=torch.bfloataa, inplace=True)
lowercase    :							Tuple      =					ipex.optimize(pipe.vae.eval(), dtype=torch.bfloataa, inplace=True)
lowercase    :							Optional[Any]      =					ipex.optimize(pipe.text_encoder.eval(), dtype=torch.bfloataa, inplace=True)
if pipe.requires_safety_checker:
					lowercase    :							Tuple      =					ipex.optimize(pipe.safety_checker.eval(), dtype=torch.bfloataa, inplace=True)

# compute
lowercase    :							List[str]      =					666
lowercase    :							Tuple      =					torch.Generator(device).manual_seed(seed)
lowercase    :							Union[str, Any]      =					{"generator": generator}
if args.steps is not None:
					lowercase    :							Dict      =					args.steps

with torch.cpu.amp.autocast(enabled=True, dtype=torch.bfloataa):
					lowercase    :							List[str]      =					pipe(prompt, **generate_kwargs).images[0]

# save image
image.save("generated.png")

