"""simple docstring"""


import unittest
from typing import Dict, List, Optional, Union

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
    import torch

if is_vision_available():
    from PIL import Image

    from transformers import BridgeTowerImageProcessor


class        a				(		unittest.TestCase   ):


    def __init__(					self	:  int   ,				__lowerCAmelCase	:  Dict   ,				__lowerCAmelCase	:  bool = True   ,				__lowerCAmelCase	:  Dict[str, int] = None   ,				__lowerCAmelCase	:  int = 32   ,				__lowerCAmelCase	:  bool = True   ,				__lowerCAmelCase	:  Union[int, float] = 1 / 255   ,				__lowerCAmelCase	:  bool = True   ,				__lowerCAmelCase	:  bool = True   ,				__lowerCAmelCase	:  Optional[Union[float, List[float]]] = [0.48_145_466, 0.4_578_275, 0.40_821_073]   ,				__lowerCAmelCase	:  Optional[Union[float, List[float]]] = [0.26_862_954, 0.26_130_258, 0.27_577_711]   ,				__lowerCAmelCase	:  bool = True   ,				__lowerCAmelCase	:  Union[str, Any]=7   ,				__lowerCAmelCase	:  Optional[int]=30   ,				__lowerCAmelCase	:  Union[str, Any]=400   ,				__lowerCAmelCase	:  List[Any]=3   ,				):
      _UpperCAmelCase						 =      parent
      _UpperCAmelCase						 =      do_resize
      _UpperCAmelCase						 =      size if size is not None else {"""shortest_edge""": 288}
      _UpperCAmelCase						 =      size_divisor
      _UpperCAmelCase						 =      do_rescale
      _UpperCAmelCase						 =      rescale_factor
      _UpperCAmelCase						 =      do_normalize
      _UpperCAmelCase						 =      do_center_crop
      _UpperCAmelCase						 =      image_mean
      _UpperCAmelCase						 =      image_std
      _UpperCAmelCase						 =      do_pad
      _UpperCAmelCase						 =      batch_size
      _UpperCAmelCase						 =      num_channels
      _UpperCAmelCase						 =      min_resolution
      _UpperCAmelCase						 =      max_resolution


    def      lowerCAmelCase_  (					self	:  Tuple					):
      return {
          "image_mean": self.image_mean,
          "image_std": self.image_std,
          "do_normalize": self.do_normalize,
          "do_resize": self.do_resize,
          "size": self.size,
          "size_divisor": self.size_divisor,
      }


    def      lowerCAmelCase_  (					self	:  Optional[int]   ,				__lowerCAmelCase	:  List[str]   ,				__lowerCAmelCase	:  Dict=False					):
      if not batched:
        _UpperCAmelCase						 =      self.size["""shortest_edge"""]
        _UpperCAmelCase						 =      image_inputs[0]
        if isinstance(UpperCAmelCase__   ,				Image.Image					):
          _UpperCAmelCase ,				_UpperCAmelCase						 =      image.size
        else:
          _UpperCAmelCase ,				_UpperCAmelCase						 =      image.shape[1], image.shape[2]
        _UpperCAmelCase						 =      size / min(UpperCAmelCase__   ,				UpperCAmelCase__					)
        if h < w:
          _UpperCAmelCase ,				_UpperCAmelCase						 =      size, scale * w
        else:
          _UpperCAmelCase ,				_UpperCAmelCase						 =      scale * h, size

        _UpperCAmelCase						 =      int((1333 / 800) * size					)
        if max(UpperCAmelCase__   ,				UpperCAmelCase__					) > max_size:
          _UpperCAmelCase						 =      max_size / max(UpperCAmelCase__   ,				UpperCAmelCase__					)
          _UpperCAmelCase						 =      newh * scale
          _UpperCAmelCase						 =      neww * scale

        _UpperCAmelCase ,				_UpperCAmelCase						 =      int(newh + 0.5					), int(neww + 0.5					)
        _UpperCAmelCase ,				_UpperCAmelCase						 =      (
            newh // self.size_divisor * self.size_divisor,
            neww // self.size_divisor * self.size_divisor,
        )

      else:
        _UpperCAmelCase						 =      []
        for image in image_inputs:
          _UpperCAmelCase ,				_UpperCAmelCase						 =      self.get_expected_values([image]					)
          expected_values.append((expected_height, expected_width)					)
        _UpperCAmelCase						 =      max(UpperCAmelCase__   ,				key=lambda __lowerCAmelCase					: item[0]					)[0]
        _UpperCAmelCase						 =      max(UpperCAmelCase__   ,				key=lambda __lowerCAmelCase					: item[1]					)[1]

      return expected_height, expected_width



@require_torch
@require_vision
class        a				(		lowerCAmelCase_	,							unittest.TestCase   ):
    _snake_case     :  List[Any]     							=							BridgeTowerImageProcessor if is_vision_available() else None


    def      lowerCAmelCase_  (					self	:  Optional[int]					):
      _UpperCAmelCase						 =      BridgeTowerImageProcessingTester(self					)


    @property
    def      lowerCAmelCase_  (					self	:  List[str]					):
      return self.image_processor_tester.prepare_image_processor_dict()


    def      lowerCAmelCase_  (					self	:  Optional[int]					):
      _UpperCAmelCase						 =      self.image_processing_class(**self.image_processor_dict					)
      self.assertTrue(hasattr(UpperCAmelCase__   ,				"""image_mean"""					)					)
      self.assertTrue(hasattr(UpperCAmelCase__   ,				"""image_std"""					)					)
      self.assertTrue(hasattr(UpperCAmelCase__   ,				"""do_normalize"""					)					)
      self.assertTrue(hasattr(UpperCAmelCase__   ,				"""do_resize"""					)					)
      self.assertTrue(hasattr(UpperCAmelCase__   ,				"""size"""					)					)
      self.assertTrue(hasattr(UpperCAmelCase__   ,				"""size_divisor"""					)					)


    def      lowerCAmelCase_  (					self	:  Optional[int]					):
      pass


    def      lowerCAmelCase_  (					self	:  List[str]					):
      # Initialize image processor
      _UpperCAmelCase						 =      self.image_processing_class(**self.image_processor_dict					)
      # create random PIL images
      _UpperCAmelCase						 =      prepare_image_inputs(self.image_processor_tester   ,				equal_resolution=UpperCAmelCase__					)
      for image in image_inputs:
        self.assertIsInstance(UpperCAmelCase__   ,				Image.Image					)

      # Test not batched input
      _UpperCAmelCase						 =      image_processing(image_inputs[0]   ,				return_tensors="""pt"""					).pixel_values

      _UpperCAmelCase ,				_UpperCAmelCase						 =      self.image_processor_tester.get_expected_values(UpperCAmelCase__					)
      self.assertEqual(
          encoded_images.shape   ,				(1, self.image_processor_tester.num_channels, expected_height, expected_width)   ,				)

      # Test batched
      _UpperCAmelCase						 =      image_processing(UpperCAmelCase__   ,				return_tensors="""pt"""					).pixel_values

      _UpperCAmelCase ,				_UpperCAmelCase						 =      self.image_processor_tester.get_expected_values(UpperCAmelCase__   ,				batched=UpperCAmelCase__					)
      self.assertEqual(
          encoded_images.shape   ,				(
              self.image_processor_tester.batch_size,
              self.image_processor_tester.num_channels,
              expected_height,
              expected_width,
          )   ,				)


    def      lowerCAmelCase_  (					self	:  Any					):
      # Initialize image processor
      _UpperCAmelCase						 =      self.image_processing_class(**self.image_processor_dict					)
      # create random numpy tensors
      _UpperCAmelCase						 =      prepare_image_inputs(self.image_processor_tester   ,				equal_resolution=UpperCAmelCase__   ,				numpify=UpperCAmelCase__					)
      for image in image_inputs:
        self.assertIsInstance(UpperCAmelCase__   ,				np.ndarray					)

      # Test not batched input
      _UpperCAmelCase						 =      image_processing(image_inputs[0]   ,				return_tensors="""pt"""					).pixel_values

      _UpperCAmelCase ,				_UpperCAmelCase						 =      self.image_processor_tester.get_expected_values(UpperCAmelCase__					)
      self.assertEqual(
          encoded_images.shape   ,				(1, self.image_processor_tester.num_channels, expected_height, expected_width)   ,				)

      # Test batched
      _UpperCAmelCase						 =      image_processing(UpperCAmelCase__   ,				return_tensors="""pt"""					).pixel_values

      _UpperCAmelCase ,				_UpperCAmelCase						 =      self.image_processor_tester.get_expected_values(UpperCAmelCase__   ,				batched=UpperCAmelCase__					)
      self.assertEqual(
          encoded_images.shape   ,				(
              self.image_processor_tester.batch_size,
              self.image_processor_tester.num_channels,
              expected_height,
              expected_width,
          )   ,				)


    def      lowerCAmelCase_  (					self	:  int					):
      # Initialize image processor
      _UpperCAmelCase						 =      self.image_processing_class(**self.image_processor_dict					)
      # create random PyTorch tensors
      _UpperCAmelCase						 =      prepare_image_inputs(self.image_processor_tester   ,				equal_resolution=UpperCAmelCase__   ,				torchify=UpperCAmelCase__					)
      for image in image_inputs:
        self.assertIsInstance(UpperCAmelCase__   ,				torch.Tensor					)

      # Test not batched input
      _UpperCAmelCase						 =      image_processing(image_inputs[0]   ,				return_tensors="""pt"""					).pixel_values

      _UpperCAmelCase ,				_UpperCAmelCase						 =      self.image_processor_tester.get_expected_values(UpperCAmelCase__					)
      self.assertEqual(
          encoded_images.shape   ,				(1, self.image_processor_tester.num_channels, expected_height, expected_width)   ,				)

      # Test batched
      _UpperCAmelCase						 =      image_processing(UpperCAmelCase__   ,				return_tensors="""pt"""					).pixel_values

      _UpperCAmelCase ,				_UpperCAmelCase						 =      self.image_processor_tester.get_expected_values(UpperCAmelCase__   ,				batched=UpperCAmelCase__					)
      self.assertEqual(
          encoded_images.shape   ,				(
              self.image_processor_tester.batch_size,
              self.image_processor_tester.num_channels,
              expected_height,
              expected_width,
          )   ,				)




"""simple docstring"""


from argparse import ArgumentParser, Namespace
from typing import Any, List, Optional

from ..pipelines import Pipeline, get_supported_tasks, pipeline
from ..utils import logging
from . import BaseTransformersCLICommand


try:
    from fastapi import Body, FastAPI, HTTPException
    from fastapi.routing import APIRoute
    from pydantic import BaseModel
    from starlette.responses import JSONResponse
    from uvicorn import run

    UpperCAmelCase__  		=						True
except (ImportError, AttributeError):
    UpperCAmelCase__  		=						object






    def     __UpperCAmelCase					(			*lowercase   ,**lowercase		):







      """simple docstring"""
      pass

    UpperCAmelCase__  		=						False


UpperCAmelCase__  		=						logging.get_logger("""transformers-cli/serving""")






def     __UpperCAmelCase					(			lowercase		):







  """simple docstring"""
  _UpperCAmelCase						 =      pipeline(
      task=args.task   ,model=args.model if args.model else None   ,config=args.config   ,tokenizer=args.tokenizer   ,device=args.device   ,)
  return ServeCommand(lowercase   ,args.host   ,args.port   ,args.workers		)


class        a				(		lowerCAmelCase_   ):
    _snake_case     :  dict



class        a				(		lowerCAmelCase_   ):
    _snake_case     :  List[str]
    _snake_case     :  Optional[List[int]]



class        a				(		lowerCAmelCase_   ):
    _snake_case     :  str



class        a				(		lowerCAmelCase_   ):
    _snake_case     :  Any



class        a				(		lowerCAmelCase_   ):


    @staticmethod
    def      lowerCAmelCase_  (					__lowerCAmelCase	:  ArgumentParser					):
      _UpperCAmelCase						 =      parser.add_parser(
          """serve"""   ,				help="""CLI tool to run inference requests through REST and GraphQL endpoints."""					)
      serve_parser.add_argument(
          """--task"""   ,				type=__lowerCAmelCase   ,				choices=get_supported_tasks()   ,				help="""The task to run the pipeline on"""   ,				)
      serve_parser.add_argument("""--host"""   ,				type=__lowerCAmelCase   ,				default="""localhost"""   ,				help="""Interface the server will listen on."""					)
      serve_parser.add_argument("""--port"""   ,				type=__lowerCAmelCase   ,				default=8888   ,				help="""Port the serving will listen to."""					)
      serve_parser.add_argument("""--workers"""   ,				type=__lowerCAmelCase   ,				default=1   ,				help="""Number of http workers"""					)
      serve_parser.add_argument("""--model"""   ,				type=__lowerCAmelCase   ,				help="""Model's name or path to stored model."""					)
      serve_parser.add_argument("""--config"""   ,				type=__lowerCAmelCase   ,				help="""Model's config name or path to stored model."""					)
      serve_parser.add_argument("""--tokenizer"""   ,				type=__lowerCAmelCase   ,				help="""Tokenizer name to use."""					)
      serve_parser.add_argument(
          """--device"""   ,				type=__lowerCAmelCase   ,				default=-1   ,				help="""Indicate the device to run onto, -1 indicates CPU, >= 0 indicates GPU (default: -1)"""   ,				)
      serve_parser.set_defaults(func=__lowerCAmelCase					)


    def __init__(					self	:  Tuple   ,				__lowerCAmelCase	:  Pipeline   ,				__lowerCAmelCase	:  str   ,				__lowerCAmelCase	:  int   ,				__lowerCAmelCase	:  int					):
      _UpperCAmelCase						 =      pipeline

      _UpperCAmelCase						 =      host
      _UpperCAmelCase						 =      port
      _UpperCAmelCase						 =      workers

      if not _serve_dependencies_installed:
        raise RuntimeError(
            """Using serve command requires FastAPI and uvicorn. """
            """Please install transformers with [serving]: pip install \"transformers[serving]\"."""
            """Or install FastAPI and uvicorn separately."""					)
      else:
        logger.info(f'''Serving model over {host}:{port}'''					)
        _UpperCAmelCase						 =      FastAPI(
            routes=[
                APIRoute(
                    """/"""   ,				self.model_info   ,				response_model=__lowerCAmelCase   ,				response_class=__lowerCAmelCase   ,				methods=["""GET"""]   ,				),
                APIRoute(
                    """/tokenize"""   ,				self.tokenize   ,				response_model=__lowerCAmelCase   ,				response_class=__lowerCAmelCase   ,				methods=["""POST"""]   ,				),
                APIRoute(
                    """/detokenize"""   ,				self.detokenize   ,				response_model=__lowerCAmelCase   ,				response_class=__lowerCAmelCase   ,				methods=["""POST"""]   ,				),
                APIRoute(
                    """/forward"""   ,				self.forward   ,				response_model=__lowerCAmelCase   ,				response_class=__lowerCAmelCase   ,				methods=["""POST"""]   ,				),
            ]   ,				timeout=600   ,				)


    def      lowerCAmelCase_  (					self	:  Optional[int]					):
      run(self._app   ,				host=self.host   ,				port=self.port   ,				workers=self.workers					)


    def      lowerCAmelCase_  (					self	:  str					):
      return ServeModelInfoResult(infos=vars(self._pipeline.model.config					)					)


    def      lowerCAmelCase_  (					self	:  Tuple   ,				__lowerCAmelCase	:  str = Body(__lowerCAmelCase   ,				embed=__lowerCAmelCase					)   ,				__lowerCAmelCase	:  bool = Body(__lowerCAmelCase   ,				embed=__lowerCAmelCase					)					):
      try:
        _UpperCAmelCase						 =      self._pipeline.tokenizer.tokenize(__lowerCAmelCase					)

        if return_ids:
          _UpperCAmelCase						 =      self._pipeline.tokenizer.convert_tokens_to_ids(__lowerCAmelCase					)
          return ServeTokenizeResult(tokens=__lowerCAmelCase   ,				tokens_ids=__lowerCAmelCase					)
        else:
          return ServeTokenizeResult(tokens=__lowerCAmelCase					)

      except Exception as e:
        raise HTTPException(status_code=500   ,				detail={"""model""": """""", """error""": str(__lowerCAmelCase					)}					)


    def      lowerCAmelCase_  (					self	:  int   ,				__lowerCAmelCase	:  List[int] = Body(__lowerCAmelCase   ,				embed=__lowerCAmelCase					)   ,				__lowerCAmelCase	:  bool = Body(__lowerCAmelCase   ,				embed=__lowerCAmelCase					)   ,				__lowerCAmelCase	:  bool = Body(__lowerCAmelCase   ,				embed=__lowerCAmelCase					)   ,				):
      try:
        _UpperCAmelCase						 =      self._pipeline.tokenizer.decode(__lowerCAmelCase   ,				__lowerCAmelCase   ,				__lowerCAmelCase					)
        return ServeDeTokenizeResult(model=""""""   ,				text=__lowerCAmelCase					)
      except Exception as e:
        raise HTTPException(status_code=500   ,				detail={"""model""": """""", """error""": str(__lowerCAmelCase					)}					)


    async def      lowerCAmelCase_  (					self	:  str   ,				__lowerCAmelCase	:  List[Any]=Body(__lowerCAmelCase   ,				embed=__lowerCAmelCase					)					):

      # Check we don't have empty string
      if len(__lowerCAmelCase					) == 0:
        return ServeForwardResult(output=[]   ,				attention=[]					)

      try:
        # Forward through the model
        _UpperCAmelCase						 =      self._pipeline(__lowerCAmelCase					)
        return ServeForwardResult(output=__lowerCAmelCase					)
      except Exception as e:
        raise HTTPException(500   ,				{"""error""": str(__lowerCAmelCase					)}					)
