


"""simple docstring"""


from typing import Callable, List, Optional, Union

import PIL
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPSegForImageSegmentation,
    CLIPSegProcessor,
    CLIPTextModel,
    CLIPTokenizer,
)

from diffusers import DiffusionPipeline
from diffusers.configuration_utils import FrozenDict
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionInpaintPipeline
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import deprecate, is_accelerate_available, logging


A     :   str			    =			logging.get_logger(__name__)  # pylint: disable=invalid-name
class   _UpperCamelCase  (   lowerCAmelCase__		):
							'''simple docstring'''







							def __init__(     self    ,   __a    ,   __a    ,   __a    ,   __a    ,   __a    ,   __a    ,   __a    ,   __a    ,   __a    ,   ):
										super().__init__()

										if hasattr(scheduler.config    ,   "steps_offset"	) and scheduler.config.steps_offset != 1:
													__lowerCAmelCase				    =			(
													    f"The configuration file of this scheduler: {scheduler} is outdated. `steps_offset`"
													    f" should be set to 1 instead of {scheduler.config.steps_offset}. Please make sure "
													    "to update the config accordingly as leaving `steps_offset` might led to incorrect results"
													    " in future versions. If you have downloaded this checkpoint from the Hugging Face Hub,"
													    " it would be very nice if you could open a Pull request for the `scheduler/scheduler_config.json`"
													    " file"
													)
													deprecate("steps_offset!=1"    ,   "1.0.0"    ,   __a    ,   standard_warn=__a	)
													__lowerCAmelCase				    =			dict(scheduler.config	)
													__lowerCAmelCase				    =			1
													__lowerCAmelCase				    =			FrozenDict(__a	)

										if hasattr(scheduler.config    ,   "skip_prk_steps"	) and scheduler.config.skip_prk_steps is False:
													__lowerCAmelCase				    =			(
													    f"The configuration file of this scheduler: {scheduler} has not set the configuration"
													    " `skip_prk_steps`. `skip_prk_steps` should be set to True in the configuration file. Please make"
													    " sure to update the config accordingly as not setting `skip_prk_steps` in the config might lead to"
													    " incorrect results in future versions. If you have downloaded this checkpoint from the Hugging Face"
													    " Hub, it would be very nice if you could open a Pull request for the"
													    " `scheduler/scheduler_config.json` file"
													)
													deprecate("skip_prk_steps not set"    ,   "1.0.0"    ,   __a    ,   standard_warn=__a	)
													__lowerCAmelCase				    =			dict(scheduler.config	)
													__lowerCAmelCase				    =			True
													__lowerCAmelCase				    =			FrozenDict(__a	)

										if safety_checker is None:
													logger.warning(
													    f"You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure"
													    " that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered"
													    " results in services or applications open to the public. Both the diffusers team and Hugging Face"
													    " strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling"
													    " it only for use-cases that involve analyzing network behavior or auditing its results. For more"
													    " information, please have a look at https://github.com/huggingface/diffusers/pull/254 ."	)

										self.register_modules(
										    segmentation_model=__a    ,   segmentation_processor=__a    ,   vae=__a    ,   text_encoder=__a    ,   tokenizer=__a    ,   unet=__a    ,   scheduler=__a    ,   safety_checker=__a    ,   feature_extractor=__a    ,   )

							def 			snake_case				(     self    ,   __a = "auto"	):
										if slice_size == "auto":
													# half the attention head size is usually a good trade-off between
													# speed and memory
													__lowerCAmelCase				    =			self.unet.config.attention_head_dim // 2
										self.unet.set_attention_slice(__a	)

							def 			snake_case				(     self	):
										self.enable_attention_slicing(__a	)

							def 			snake_case				(     self	):
										if is_accelerate_available():
													from accelerate import cpu_offload
										else:
													raise ImportError("Please install accelerate via `pip install accelerate`"	)

										__lowerCAmelCase				    =			torch.device("cuda"	)

										for cpu_offloaded_model in [self.unet, self.text_encoder, self.vae, self.safety_checker]:
													if cpu_offloaded_model is not None:
																cpu_offload(__a    ,   __a	)

							@property
							# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
							def 			snake_case				(     self	):
										if self.device != torch.device("meta"	) or not hasattr(self.unet    ,   "_hf_hook"	):
													return self.device
										for module in self.unet.modules():
													if (
													    hasattr(__a    ,   "_hf_hook"	)
													    and hasattr(module._hf_hook    ,   "execution_device"	)
													    and module._hf_hook.execution_device is not None
													):
																return torch.device(module._hf_hook.execution_device	)
										return self.device



							@torch.no_grad()
							def __call__(     self    ,   __a    ,   __a    ,   __a    ,   __a = 5_12    ,   __a = 5_12    ,   __a = 50    ,   __a = 7.5    ,   __a = None    ,   __a = 1    ,   __a = 0.0    ,   __a = None    ,   __a = None    ,   __a = "pil"    ,   __a = True    ,   __a = None    ,   __a = 1    ,   **__a    ,   ):
										__lowerCAmelCase				    =			self.segmentation_processor(
										    text=[text]    ,   images=[image]    ,   padding="max_length"    ,   return_tensors="pt"	).to(self.device	)
										__lowerCAmelCase				    =			self.segmentation_model(**__a	)
										__lowerCAmelCase				    =			torch.sigmoid(outputs.logits	).cpu().detach().unsqueeze(-1	).numpy()
										__lowerCAmelCase				    =			self.numpy_to_pil(__a	)[0].resize(image.size	)

										# Run inpainting pipeline with the generated mask
										__lowerCAmelCase				    =			StableDiffusionInpaintPipeline(
										    vae=self.vae    ,   text_encoder=self.text_encoder    ,   tokenizer=self.tokenizer    ,   unet=self.unet    ,   scheduler=self.scheduler    ,   safety_checker=self.safety_checker    ,   feature_extractor=self.feature_extractor    ,   )
										return inpainting_pipeline(
										    prompt=__a    ,   image=__a    ,   mask_image=__a    ,   height=__a    ,   width=__a    ,   num_inference_steps=__a    ,   guidance_scale=__a    ,   negative_prompt=__a    ,   num_images_per_prompt=__a    ,   eta=__a    ,   generator=__a    ,   latents=__a    ,   output_type=__a    ,   return_dict=__a    ,   callback=__a    ,   callback_steps=__a    ,   )







def 					__lowerCamelCase	( ):







							'''simple docstring'''






							return [list(range(1000 - i   ,							-1000 - i   ,							-1      )      ) for i in range(1000      )]


_UpperCAmelCase		:   Union[str, Any]										=       generate_large_matrix()
_UpperCAmelCase		:   Tuple										=       (
    [[4, 3, 2, -1], [3, 2, 1, -1], [1, 1, -1, -2], [-1, -1, -2, -3]],
    [[3, 2], [1, 0]],
    [[7, 7, 6]],
    [[7, 7, 6], [-1, -2, -3]],
    grid,
)





def 					__lowerCamelCase	( UpperCamelCase__      ):







							'''simple docstring'''






							assert all(row == sorted(UpperCamelCase__   ,							reverse=UpperCamelCase__      ) for row in grid      )
							assert all(list(UpperCamelCase__      ) == sorted(UpperCamelCase__   ,							reverse=UpperCamelCase__      ) for col in zip(*UpperCamelCase__      )      )





def 					__lowerCamelCase	( UpperCamelCase__      ):







							'''simple docstring'''






							snake_case_							=    0
							snake_case_							=    len(UpperCamelCase__      ) - 1

							# Edge cases such as no values or all numbers are negative.
							if not array or array[0] < 0:
														return 0

							while right + 1 > left:
														snake_case_							=    (left + right) // 2
														snake_case_							=    array[mid]

														# Num must be negative and the index must be greater than or equal to 0.
														if num < 0 and array[mid - 1] >= 0:
																					return mid

														if num >= 0:
																					snake_case_							=    mid + 1
														else:
																					snake_case_							=    mid - 1
    # No negative numbers so return the last index of the array + 1 which is the length.
							return len(UpperCamelCase__      )





def 					__lowerCamelCase	( UpperCamelCase__      ):







							'''simple docstring'''






							snake_case_							=    0
							snake_case_							=    len(grid[0]      )

							for i in range(len(UpperCamelCase__      )      ):
														snake_case_							=    find_negative_index(grid[i][:bound]      )
														total += bound
							return (len(UpperCamelCase__      ) * len(grid[0]      )) - total





def 					__lowerCamelCase	( UpperCamelCase__      ):







							'''simple docstring'''






							return len([number for row in grid for number in row if number < 0]      )





def 					__lowerCamelCase	( UpperCamelCase__      ):







							'''simple docstring'''






							snake_case_							=    0
							for row in grid:
														for i, number in enumerate(UpperCamelCase__      ):
																					if number < 0:
																												total += len(UpperCamelCase__      ) - i
																												break
							return total





def 					__lowerCamelCase	( ):







							'''simple docstring'''






							from timeit import timeit

							print('Running benchmarks'      )
							snake_case_							=    (
							    'from __main__ import count_negatives_binary_search, '
							    'count_negatives_brute_force, count_negatives_brute_force_with_break, grid'
							)
							for func in (
							    "count_negatives_binary_search",  # took 0.7727 seconds
							    "count_negatives_brute_force_with_break",  # took 4.6505 seconds
							    "count_negatives_brute_force",  # took 12.8160 seconds
							):
														snake_case_							=    timeit(F'''{func}(grid=grid)'''   ,							setup=UpperCamelCase__   ,							number=500      )
														print(F'''{func}() took {time:0.4f} seconds'''      )


if __name__ == "__main__":
					import doctest

					doctest.testmod()
					benchmark()




