
'''simple docstring'''
import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch







class   a_ (      UpperCamelCase__					):




			@require_torch
			def 					lowerCAmelCase__					(       self	):
				# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
				# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
				# while running an external program

				# python one-liner segments

				# this must be loaded before socket.socket is monkey-patched
				a_      				=						"""
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

				a_      				=						"""
mname = \"hf-internal-testing/tiny-random-bert\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task=\"fill-mask\", model=mname)
print(\"success\")
        """

				a_      				=						"""
import socket
def offline_socket(*args, **kwargs): raise RuntimeError(\"Offline mode is enabled, we shouldn't access internet\")
socket.socket = offline_socket
        """

				# Force fetching the files so that we can use the cache
				a_      				=						"""hf-internal-testing/tiny-random-bert"""
				BertConfig.from_pretrained(UpperCAmelCase	)
				BertModel.from_pretrained(UpperCAmelCase	)
				BertTokenizer.from_pretrained(UpperCAmelCase	)
				pipeline(task="""fill-mask"""					,   model=UpperCAmelCase	)

				# baseline - just load from_pretrained with normal network
				a_      				=						[sys.executable, """-c""", """\n""".join([load, run, mock]	)]

				# should succeed
				a_      				=						self.get_env()
				# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
				a_      				=						"""1"""
				a_      				=						subprocess.run(UpperCAmelCase					,   env=UpperCAmelCase					,   check=UpperCAmelCase					,   capture_output=UpperCAmelCase	)
				self.assertEqual(result.returncode					,   0					,   result.stderr	)
				self.assertIn("""success"""					,   result.stdout.decode()	)




			@require_torch
			def 					lowerCAmelCase__					(       self	):
				# python one-liner segments
				# this must be loaded before socket.socket is monkey-patched
				a_      				=						"""
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

				a_      				=						"""
mname = \"hf-internal-testing/tiny-random-bert\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task=\"fill-mask\", model=mname)
print(\"success\")
        """

				a_      				=						"""
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Faking flaky internet\")
socket.socket = offline_socket
        """

				# Force fetching the files so that we can use the cache
				a_      				=						"""hf-internal-testing/tiny-random-bert"""
				BertConfig.from_pretrained(UpperCAmelCase	)
				BertModel.from_pretrained(UpperCAmelCase	)
				BertTokenizer.from_pretrained(UpperCAmelCase	)
				pipeline(task="""fill-mask"""					,   model=UpperCAmelCase	)

				# baseline - just load from_pretrained with normal network
				a_      				=						[sys.executable, """-c""", """\n""".join([load, run, mock]	)]

				# should succeed
				a_      				=						self.get_env()
				a_      				=						subprocess.run(UpperCAmelCase					,   env=UpperCAmelCase					,   check=UpperCAmelCase					,   capture_output=UpperCAmelCase	)
				self.assertEqual(result.returncode					,   0					,   result.stderr	)
				self.assertIn("""success"""					,   result.stdout.decode()	)




			@require_torch
			def 					lowerCAmelCase__					(       self	):
				# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
				# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
				# while running an external program

				# python one-liner segments

				# this must be loaded before socket.socket is monkey-patched
				a_      				=						"""
from transformers import BertConfig, BertModel, BertTokenizer
        """

				a_      				=						"""
mname = \"hf-internal-testing/tiny-random-bert-sharded\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
print(\"success\")
        """

				a_      				=						"""
import socket
def offline_socket(*args, **kwargs): raise ValueError(\"Offline mode is enabled\")
socket.socket = offline_socket
        """

				# baseline - just load from_pretrained with normal network
				a_      				=						[sys.executable, """-c""", """\n""".join([load, run]	)]

				# should succeed
				a_      				=						self.get_env()
				a_      				=						subprocess.run(UpperCAmelCase					,   env=UpperCAmelCase					,   check=UpperCAmelCase					,   capture_output=UpperCAmelCase	)
				self.assertEqual(result.returncode					,   0					,   result.stderr	)
				self.assertIn("""success"""					,   result.stdout.decode()	)

				# next emulate no network
				a_      				=						[sys.executable, """-c""", """\n""".join([load, mock, run]	)]

				# Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
				# env["TRANSFORMERS_OFFLINE"] = "0"
				# result = subprocess.run(cmd, env=env, check=False, capture_output=True)
				# self.assertEqual(result.returncode, 1, result.stderr)

				# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
				a_      				=						"""1"""
				a_      				=						subprocess.run(UpperCAmelCase					,   env=UpperCAmelCase					,   check=UpperCAmelCase					,   capture_output=UpperCAmelCase	)
				self.assertEqual(result.returncode					,   0					,   result.stderr	)
				self.assertIn("""success"""					,   result.stdout.decode()	)




			@require_torch
			def 					lowerCAmelCase__					(       self	):
				a_      				=						"""
from transformers import pipeline
        """
				a_      				=						"""
mname = \"hf-internal-testing/tiny-random-bert\"
pipe = pipeline(model=mname)
        """

				a_      				=						"""
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")
socket.socket = offline_socket
        """
				a_      				=						self.get_env()
				a_      				=						"""1"""
				a_      				=						[sys.executable, """-c""", """\n""".join([load, mock, run]	)]
				a_      				=						subprocess.run(UpperCAmelCase					,   env=UpperCAmelCase					,   check=UpperCAmelCase					,   capture_output=UpperCAmelCase	)
				self.assertEqual(result.returncode					,   1					,   result.stderr	)
				self.assertIn(
				    """You cannot infer task automatically within `pipeline` when using offline mode"""					,   result.stderr.decode().replace("""\n"""					,   """"""	)					,   )






			@require_torch
			def 					lowerCAmelCase__					(       self	):
				a_      				=						"""
from transformers import AutoModel
        """
				a_      				=						"""
mname = \"hf-internal-testing/test_dynamic_model\"
AutoModel.from_pretrained(mname, trust_remote_code=True)
print(\"success\")
        """

				# baseline - just load from_pretrained with normal network
				a_      				=						[sys.executable, """-c""", """\n""".join([load, run]	)]

				# should succeed
				a_      				=						self.get_env()
				a_      				=						subprocess.run(UpperCAmelCase					,   env=UpperCAmelCase					,   check=UpperCAmelCase					,   capture_output=UpperCAmelCase	)
				self.assertEqual(result.returncode					,   0					,   result.stderr	)
				self.assertIn("""success"""					,   result.stdout.decode()	)

				# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
				a_      				=						"""1"""
				a_      				=						subprocess.run(UpperCAmelCase					,   env=UpperCAmelCase					,   check=UpperCAmelCase					,   capture_output=UpperCAmelCase	)
				self.assertEqual(result.returncode					,   0					,   result.stderr	)
				self.assertIn("""success"""					,   result.stdout.decode()	)


'''simple docstring'''
import unittest

import torch

from diffusers import DDIMScheduler, DDPMScheduler, UNetaDModel
from diffusers.training_utils import set_seed
from diffusers.utils.testing_utils import slow


lowercase__											=False







class   a_ (      unittest.TestCase					):




			def 					lowerCAmelCase__					(       self					,   UpperCAmelCase=32	):
				set_seed(0	)
				a_      				=						UNetaDModel(sample_size=UpperCAmelCase					,   in_channels=3					,   out_channels=3	)
				a_      				=						torch.optim.SGD(model.parameters()					,   lr=0.00_01	)
				return model, optimizer






			@slow
			def 					lowerCAmelCase__					(       self	):
				a_      				=						"""cpu"""  # ensure full determinism without setting the CUBLAS_WORKSPACE_CONFIG env variable
				a_      				=						DDPMScheduler(
				    num_train_timesteps=10_00					,   beta_start=0.00_01					,   beta_end=0.02					,   beta_schedule="""linear"""					,   clip_sample=UpperCAmelCase					,   )
				a_      				=						DDIMScheduler(
				    num_train_timesteps=10_00					,   beta_start=0.00_01					,   beta_end=0.02					,   beta_schedule="""linear"""					,   clip_sample=UpperCAmelCase					,   )

				assert ddpm_scheduler.config.num_train_timesteps == ddim_scheduler.config.num_train_timesteps

				# shared batches for DDPM and DDIM
				set_seed(0	)
				a_      				=						[torch.randn((4, 3, 32, 32)	).clip(-1					,   1	).to(UpperCAmelCase	) for _ in range(4	)]
				a_      				=						[torch.randn((4, 3, 32, 32)	).to(UpperCAmelCase	) for _ in range(4	)]
				a_      				=						[torch.randint(0					,   10_00					,   (4,)	).long().to(UpperCAmelCase	) for _ in range(4	)]

				# train with a DDPM scheduler
				a_				,  a_      				=						self.get_model_optimizer(resolution=32	)
				model.train().to(UpperCAmelCase	)
				for i in range(4	):
					optimizer.zero_grad()
					a_      				=						ddpm_scheduler.add_noise(clean_images[i]					,   noise[i]					,   timesteps[i]	)
					a_      				=						model(UpperCAmelCase					,   timesteps[i]	).sample
					a_      				=						torch.nn.functional.mse_loss(UpperCAmelCase					,   noise[i]	)
					loss.backward()
					optimizer.step()
				del model, optimizer

				# recreate the model and optimizer, and retry with DDIM
				a_				,  a_      				=						self.get_model_optimizer(resolution=32	)
				model.train().to(UpperCAmelCase	)
				for i in range(4	):
					optimizer.zero_grad()
					a_      				=						ddim_scheduler.add_noise(clean_images[i]					,   noise[i]					,   timesteps[i]	)
					a_      				=						model(UpperCAmelCase					,   timesteps[i]	).sample
					a_      				=						torch.nn.functional.mse_loss(UpperCAmelCase					,   noise[i]	)
					loss.backward()
					optimizer.step()
				del model, optimizer

				self.assertTrue(torch.allclose(UpperCAmelCase					,   UpperCAmelCase					,   atol=1e-5	)	)
				self.assertTrue(torch.allclose(UpperCAmelCase					,   UpperCAmelCase					,   atol=1e-5	)	)
