Download runtime/transformers/image_processing_base.py from TeleEmbodied/umi-arena-submission: direct link, hf CLI and curl.
- Browser
- Download file 24.9 kB
-
https://huggingface.co/TeleEmbodied/umi-arena-submission/resolve/main/runtime/transformers/image_processing_base.py
- Command line
-
hf download hf://TeleEmbodied/umi-arena-submission/runtime/transformers/image_processing_base.py
-
curl -L -o image_processing_base.py https://huggingface.co/TeleEmbodied/umi-arena-submission/resolve/main/runtime/transformers/image_processing_base.py
24.9 kB
| # Copyright 2020 The HuggingFace Inc. team. | |
| # | |
| # Licensed under the Apache License, Version 2.0 (the "License"); | |
| # you may not use this file except in compliance with the License. | |
| # You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| # See the License for the specific language governing permissions and | |
| # limitations under the License. | |
| import copy | |
| import json | |
| import os | |
| import warnings | |
| from typing import Any, Optional, TypeVar, Union | |
| import numpy as np | |
| from .dynamic_module_utils import custom_object_save | |
| from .feature_extraction_utils import BatchFeature as BaseBatchFeature | |
| from .image_utils import is_valid_image, load_image | |
| from .utils import ( | |
| IMAGE_PROCESSOR_NAME, | |
| PROCESSOR_NAME, | |
| PushToHubMixin, | |
| copy_func, | |
| download_url, | |
| is_offline_mode, | |
| is_remote_url, | |
| logging, | |
| ) | |
| from .utils.hub import cached_file | |
| ImageProcessorType = TypeVar("ImageProcessorType", bound="ImageProcessingMixin") | |
| logger = logging.get_logger(__name__) | |
| # TODO: Move BatchFeature to be imported by both image_processing_utils and image_processing_utils_fast | |
| # We override the class string here, but logic is the same. | |
| class BatchFeature(BaseBatchFeature): | |
| r""" | |
| Holds the output of the image processor specific `__call__` methods. | |
| This class is derived from a python dictionary and can be used as a dictionary. | |
| Args: | |
| data (`dict`): | |
| Dictionary of lists/arrays/tensors returned by the __call__ method ('pixel_values', etc.). | |
| tensor_type (`Union[None, str, TensorType]`, *optional*): | |
| You can give a tensor_type here to convert the lists of integers in PyTorch/TensorFlow/Numpy Tensors at | |
| initialization. | |
| """ | |
| # TODO: (Amy) - factor out the common parts of this and the feature extractor | |
| class ImageProcessingMixin(PushToHubMixin): | |
| """ | |
| This is an image processor mixin used to provide saving/loading functionality for sequential and image feature | |
| extractors. | |
| """ | |
| _auto_class = None | |
| def __init__(self, **kwargs): | |
| """Set elements of `kwargs` as attributes.""" | |
| # This key was saved while we still used `XXXFeatureExtractor` for image processing. Now we use | |
| # `XXXImageProcessor`, this attribute and its value are misleading. | |
| kwargs.pop("feature_extractor_type", None) | |
| # Pop "processor_class" as it should be saved as private attribute | |
| self._processor_class = kwargs.pop("processor_class", None) | |
| # Additional attributes without default values | |
| for key, value in kwargs.items(): | |
| try: | |
| setattr(self, key, value) | |
| except AttributeError as err: | |
| logger.error(f"Can't set {key} with value {value} for {self}") | |
| raise err | |
| def _set_processor_class(self, processor_class: str): | |
| """Sets processor class as an attribute.""" | |
| self._processor_class = processor_class | |
| def from_pretrained( | |
| cls: type[ImageProcessorType], | |
| pretrained_model_name_or_path: Union[str, os.PathLike], | |
| cache_dir: Optional[Union[str, os.PathLike]] = None, | |
| force_download: bool = False, | |
| local_files_only: bool = False, | |
| token: Optional[Union[str, bool]] = None, | |
| revision: str = "main", | |
| **kwargs, | |
| ) -> ImageProcessorType: | |
| r""" | |
| Instantiate a type of [`~image_processing_utils.ImageProcessingMixin`] from an image processor. | |
| Args: | |
| pretrained_model_name_or_path (`str` or `os.PathLike`): | |
| This can be either: | |
| - a string, the *model id* of a pretrained image_processor hosted inside a model repo on | |
| huggingface.co. | |
| - a path to a *directory* containing a image processor file saved using the | |
| [`~image_processing_utils.ImageProcessingMixin.save_pretrained`] method, e.g., | |
| `./my_model_directory/`. | |
| - a path or url to a saved image processor JSON *file*, e.g., | |
| `./my_model_directory/preprocessor_config.json`. | |
| cache_dir (`str` or `os.PathLike`, *optional*): | |
| Path to a directory in which a downloaded pretrained model image processor should be cached if the | |
| standard cache should not be used. | |
| force_download (`bool`, *optional*, defaults to `False`): | |
| Whether or not to force to (re-)download the image processor files and override the cached versions if | |
| they exist. | |
| resume_download: | |
| Deprecated and ignored. All downloads are now resumed by default when possible. | |
| Will be removed in v5 of Transformers. | |
| proxies (`dict[str, str]`, *optional*): | |
| A dictionary of proxy servers to use by protocol or endpoint, e.g., `{'http': 'foo.bar:3128', | |
| 'http://hostname': 'foo.bar:4012'}.` The proxies are used on each request. | |
| token (`str` or `bool`, *optional*): | |
| The token to use as HTTP bearer authorization for remote files. If `True`, or not specified, will use | |
| the token generated when running `hf auth login` (stored in `~/.huggingface`). | |
| revision (`str`, *optional*, defaults to `"main"`): | |
| The specific model version to use. It can be a branch name, a tag name, or a commit id, since we use a | |
| git-based system for storing models and other artifacts on huggingface.co, so `revision` can be any | |
| identifier allowed by git. | |
| <Tip> | |
| To test a pull request you made on the Hub, you can pass `revision="refs/pr/<pr_number>"`. | |
| </Tip> | |
| return_unused_kwargs (`bool`, *optional*, defaults to `False`): | |
| If `False`, then this function returns just the final image processor object. If `True`, then this | |
| functions returns a `Tuple(image_processor, unused_kwargs)` where *unused_kwargs* is a dictionary | |
| consisting of the key/value pairs whose keys are not image processor attributes: i.e., the part of | |
| `kwargs` which has not been used to update `image_processor` and is otherwise ignored. | |
| subfolder (`str`, *optional*, defaults to `""`): | |
| In case the relevant files are located inside a subfolder of the model repo on huggingface.co, you can | |
| specify the folder name here. | |
| kwargs (`dict[str, Any]`, *optional*): | |
| The values in kwargs of any keys which are image processor attributes will be used to override the | |
| loaded values. Behavior concerning key/value pairs whose keys are *not* image processor attributes is | |
| controlled by the `return_unused_kwargs` keyword parameter. | |
| Returns: | |
| A image processor of type [`~image_processing_utils.ImageProcessingMixin`]. | |
| Examples: | |
| ```python | |
| # We can't instantiate directly the base class *ImageProcessingMixin* so let's show the examples on a | |
| # derived class: *CLIPImageProcessor* | |
| image_processor = CLIPImageProcessor.from_pretrained( | |
| "openai/clip-vit-base-patch32" | |
| ) # Download image_processing_config from huggingface.co and cache. | |
| image_processor = CLIPImageProcessor.from_pretrained( | |
| "./test/saved_model/" | |
| ) # E.g. image processor (or model) was saved using *save_pretrained('./test/saved_model/')* | |
| image_processor = CLIPImageProcessor.from_pretrained("./test/saved_model/preprocessor_config.json") | |
| image_processor = CLIPImageProcessor.from_pretrained( | |
| "openai/clip-vit-base-patch32", do_normalize=False, foo=False | |
| ) | |
| assert image_processor.do_normalize is False | |
| image_processor, unused_kwargs = CLIPImageProcessor.from_pretrained( | |
| "openai/clip-vit-base-patch32", do_normalize=False, foo=False, return_unused_kwargs=True | |
| ) | |
| assert image_processor.do_normalize is False | |
| assert unused_kwargs == {"foo": False} | |
| ```""" | |
| kwargs["cache_dir"] = cache_dir | |
| kwargs["force_download"] = force_download | |
| kwargs["local_files_only"] = local_files_only | |
| kwargs["revision"] = revision | |
| use_auth_token = kwargs.pop("use_auth_token", None) | |
| if use_auth_token is not None: | |
| warnings.warn( | |
| "The `use_auth_token` argument is deprecated and will be removed in v5 of Transformers. Please use `token` instead.", | |
| FutureWarning, | |
| ) | |
| if token is not None: | |
| raise ValueError( | |
| "`token` and `use_auth_token` are both specified. Please set only the argument `token`." | |
| ) | |
| token = use_auth_token | |
| if token is not None: | |
| kwargs["token"] = token | |
| image_processor_dict, kwargs = cls.get_image_processor_dict(pretrained_model_name_or_path, **kwargs) | |
| return cls.from_dict(image_processor_dict, **kwargs) | |
| def save_pretrained(self, save_directory: Union[str, os.PathLike], push_to_hub: bool = False, **kwargs): | |
| """ | |
| Save an image processor object to the directory `save_directory`, so that it can be re-loaded using the | |
| [`~image_processing_utils.ImageProcessingMixin.from_pretrained`] class method. | |
| Args: | |
| save_directory (`str` or `os.PathLike`): | |
| Directory where the image processor JSON file will be saved (will be created if it does not exist). | |
| push_to_hub (`bool`, *optional*, defaults to `False`): | |
| Whether or not to push your model to the Hugging Face model hub after saving it. You can specify the | |
| repository you want to push to with `repo_id` (will default to the name of `save_directory` in your | |
| namespace). | |
| kwargs (`dict[str, Any]`, *optional*): | |
| Additional key word arguments passed along to the [`~utils.PushToHubMixin.push_to_hub`] method. | |
| """ | |
| use_auth_token = kwargs.pop("use_auth_token", None) | |
| if use_auth_token is not None: | |
| warnings.warn( | |
| "The `use_auth_token` argument is deprecated and will be removed in v5 of Transformers. Please use `token` instead.", | |
| FutureWarning, | |
| ) | |
| if kwargs.get("token") is not None: | |
| raise ValueError( | |
| "`token` and `use_auth_token` are both specified. Please set only the argument `token`." | |
| ) | |
| kwargs["token"] = use_auth_token | |
| if os.path.isfile(save_directory): | |
| raise AssertionError(f"Provided path ({save_directory}) should be a directory, not a file") | |
| os.makedirs(save_directory, exist_ok=True) | |
| if push_to_hub: | |
| commit_message = kwargs.pop("commit_message", None) | |
| repo_id = kwargs.pop("repo_id", save_directory.split(os.path.sep)[-1]) | |
| repo_id = self._create_repo(repo_id, **kwargs) | |
| files_timestamps = self._get_files_timestamps(save_directory) | |
| # If we have a custom config, we copy the file defining it in the folder and set the attributes so it can be | |
| # loaded from the Hub. | |
| if self._auto_class is not None: | |
| custom_object_save(self, save_directory, config=self) | |
| # If we save using the predefined names, we can load using `from_pretrained` | |
| output_image_processor_file = os.path.join(save_directory, IMAGE_PROCESSOR_NAME) | |
| self.to_json_file(output_image_processor_file) | |
| logger.info(f"Image processor saved in {output_image_processor_file}") | |
| if push_to_hub: | |
| self._upload_modified_files( | |
| save_directory, | |
| repo_id, | |
| files_timestamps, | |
| commit_message=commit_message, | |
| token=kwargs.get("token"), | |
| ) | |
| return [output_image_processor_file] | |
| def get_image_processor_dict( | |
| cls, pretrained_model_name_or_path: Union[str, os.PathLike], **kwargs | |
| ) -> tuple[dict[str, Any], dict[str, Any]]: | |
| """ | |
| From a `pretrained_model_name_or_path`, resolve to a dictionary of parameters, to be used for instantiating a | |
| image processor of type [`~image_processor_utils.ImageProcessingMixin`] using `from_dict`. | |
| Parameters: | |
| pretrained_model_name_or_path (`str` or `os.PathLike`): | |
| The identifier of the pre-trained checkpoint from which we want the dictionary of parameters. | |
| subfolder (`str`, *optional*, defaults to `""`): | |
| In case the relevant files are located inside a subfolder of the model repo on huggingface.co, you can | |
| specify the folder name here. | |
| image_processor_filename (`str`, *optional*, defaults to `"config.json"`): | |
| The name of the file in the model directory to use for the image processor config. | |
| Returns: | |
| `tuple[Dict, Dict]`: The dictionary(ies) that will be used to instantiate the image processor object. | |
| """ | |
| cache_dir = kwargs.pop("cache_dir", None) | |
| force_download = kwargs.pop("force_download", False) | |
| resume_download = kwargs.pop("resume_download", None) | |
| proxies = kwargs.pop("proxies", None) | |
| token = kwargs.pop("token", None) | |
| use_auth_token = kwargs.pop("use_auth_token", None) | |
| local_files_only = kwargs.pop("local_files_only", False) | |
| revision = kwargs.pop("revision", None) | |
| subfolder = kwargs.pop("subfolder", "") | |
| image_processor_filename = kwargs.pop("image_processor_filename", IMAGE_PROCESSOR_NAME) | |
| from_pipeline = kwargs.pop("_from_pipeline", None) | |
| from_auto_class = kwargs.pop("_from_auto", False) | |
| if use_auth_token is not None: | |
| warnings.warn( | |
| "The `use_auth_token` argument is deprecated and will be removed in v5 of Transformers. Please use `token` instead.", | |
| FutureWarning, | |
| ) | |
| if token is not None: | |
| raise ValueError( | |
| "`token` and `use_auth_token` are both specified. Please set only the argument `token`." | |
| ) | |
| token = use_auth_token | |
| user_agent = {"file_type": "image processor", "from_auto_class": from_auto_class} | |
| if from_pipeline is not None: | |
| user_agent["using_pipeline"] = from_pipeline | |
| if is_offline_mode() and not local_files_only: | |
| logger.info("Offline mode: forcing local_files_only=True") | |
| local_files_only = True | |
| pretrained_model_name_or_path = str(pretrained_model_name_or_path) | |
| is_local = os.path.isdir(pretrained_model_name_or_path) | |
| if os.path.isdir(pretrained_model_name_or_path): | |
| image_processor_file = os.path.join(pretrained_model_name_or_path, image_processor_filename) | |
| if os.path.isfile(pretrained_model_name_or_path): | |
| resolved_image_processor_file = pretrained_model_name_or_path | |
| is_local = True | |
| elif is_remote_url(pretrained_model_name_or_path): | |
| image_processor_file = pretrained_model_name_or_path | |
| resolved_image_processor_file = download_url(pretrained_model_name_or_path) | |
| else: | |
| image_processor_file = image_processor_filename | |
| try: | |
| # Load from local folder or from cache or download from model Hub and cache | |
| resolved_image_processor_files = [ | |
| resolved_file | |
| for filename in [image_processor_file, PROCESSOR_NAME] | |
| if ( | |
| resolved_file := cached_file( | |
| pretrained_model_name_or_path, | |
| filename=filename, | |
| cache_dir=cache_dir, | |
| force_download=force_download, | |
| proxies=proxies, | |
| resume_download=resume_download, | |
| local_files_only=local_files_only, | |
| token=token, | |
| user_agent=user_agent, | |
| revision=revision, | |
| subfolder=subfolder, | |
| _raise_exceptions_for_missing_entries=False, | |
| ) | |
| ) | |
| is not None | |
| ] | |
| resolved_image_processor_file = resolved_image_processor_files[0] | |
| except OSError: | |
| # Raise any environment error raise by `cached_file`. It will have a helpful error message adapted to | |
| # the original exception. | |
| raise | |
| except Exception: | |
| # For any other exception, we throw a generic error. | |
| raise OSError( | |
| f"Can't load image processor for '{pretrained_model_name_or_path}'. If you were trying to load" | |
| " it from 'https://huggingface.co/models', make sure you don't have a local directory with the" | |
| f" same name. Otherwise, make sure '{pretrained_model_name_or_path}' is the correct path to a" | |
| f" directory containing a {image_processor_filename} file" | |
| ) | |
| try: | |
| # Load image_processor dict | |
| with open(resolved_image_processor_file, encoding="utf-8") as reader: | |
| text = reader.read() | |
| image_processor_dict = json.loads(text) | |
| image_processor_dict = image_processor_dict.get("image_processor", image_processor_dict) | |
| except json.JSONDecodeError: | |
| raise OSError( | |
| f"It looks like the config file at '{resolved_image_processor_file}' is not a valid JSON file." | |
| ) | |
| if is_local: | |
| logger.info(f"loading configuration file {resolved_image_processor_file}") | |
| else: | |
| logger.info( | |
| f"loading configuration file {image_processor_file} from cache at {resolved_image_processor_file}" | |
| ) | |
| return image_processor_dict, kwargs | |
| def from_dict(cls, image_processor_dict: dict[str, Any], **kwargs): | |
| """ | |
| Instantiates a type of [`~image_processing_utils.ImageProcessingMixin`] from a Python dictionary of parameters. | |
| Args: | |
| image_processor_dict (`dict[str, Any]`): | |
| Dictionary that will be used to instantiate the image processor object. Such a dictionary can be | |
| retrieved from a pretrained checkpoint by leveraging the | |
| [`~image_processing_utils.ImageProcessingMixin.to_dict`] method. | |
| kwargs (`dict[str, Any]`): | |
| Additional parameters from which to initialize the image processor object. | |
| Returns: | |
| [`~image_processing_utils.ImageProcessingMixin`]: The image processor object instantiated from those | |
| parameters. | |
| """ | |
| image_processor_dict = image_processor_dict.copy() | |
| return_unused_kwargs = kwargs.pop("return_unused_kwargs", False) | |
| # The `size` parameter is a dict and was previously an int or tuple in feature extractors. | |
| # We set `size` here directly to the `image_processor_dict` so that it is converted to the appropriate | |
| # dict within the image processor and isn't overwritten if `size` is passed in as a kwarg. | |
| if "size" in kwargs and "size" in image_processor_dict: | |
| image_processor_dict["size"] = kwargs.pop("size") | |
| if "crop_size" in kwargs and "crop_size" in image_processor_dict: | |
| image_processor_dict["crop_size"] = kwargs.pop("crop_size") | |
| image_processor = cls(**image_processor_dict) | |
| # Update image_processor with kwargs if needed | |
| to_remove = [] | |
| for key, value in kwargs.items(): | |
| if hasattr(image_processor, key): | |
| setattr(image_processor, key, value) | |
| to_remove.append(key) | |
| for key in to_remove: | |
| kwargs.pop(key, None) | |
| logger.info(f"Image processor {image_processor}") | |
| if return_unused_kwargs: | |
| return image_processor, kwargs | |
| else: | |
| return image_processor | |
| def to_dict(self) -> dict[str, Any]: | |
| """ | |
| Serializes this instance to a Python dictionary. | |
| Returns: | |
| `dict[str, Any]`: Dictionary of all the attributes that make up this image processor instance. | |
| """ | |
| output = copy.deepcopy(self.__dict__) | |
| output["image_processor_type"] = self.__class__.__name__ | |
| return output | |
| def from_json_file(cls, json_file: Union[str, os.PathLike]): | |
| """ | |
| Instantiates a image processor of type [`~image_processing_utils.ImageProcessingMixin`] from the path to a JSON | |
| file of parameters. | |
| Args: | |
| json_file (`str` or `os.PathLike`): | |
| Path to the JSON file containing the parameters. | |
| Returns: | |
| A image processor of type [`~image_processing_utils.ImageProcessingMixin`]: The image_processor object | |
| instantiated from that JSON file. | |
| """ | |
| with open(json_file, encoding="utf-8") as reader: | |
| text = reader.read() | |
| image_processor_dict = json.loads(text) | |
| return cls(**image_processor_dict) | |
| def to_json_string(self) -> str: | |
| """ | |
| Serializes this instance to a JSON string. | |
| Returns: | |
| `str`: String containing all the attributes that make up this feature_extractor instance in JSON format. | |
| """ | |
| dictionary = self.to_dict() | |
| for key, value in dictionary.items(): | |
| if isinstance(value, np.ndarray): | |
| dictionary[key] = value.tolist() | |
| # make sure private name "_processor_class" is correctly | |
| # saved as "processor_class" | |
| _processor_class = dictionary.pop("_processor_class", None) | |
| if _processor_class is not None: | |
| dictionary["processor_class"] = _processor_class | |
| return json.dumps(dictionary, indent=2, sort_keys=True) + "\n" | |
| def to_json_file(self, json_file_path: Union[str, os.PathLike]): | |
| """ | |
| Save this instance to a JSON file. | |
| Args: | |
| json_file_path (`str` or `os.PathLike`): | |
| Path to the JSON file in which this image_processor instance's parameters will be saved. | |
| """ | |
| with open(json_file_path, "w", encoding="utf-8") as writer: | |
| writer.write(self.to_json_string()) | |
| def __repr__(self): | |
| return f"{self.__class__.__name__} {self.to_json_string()}" | |
| def register_for_auto_class(cls, auto_class="AutoImageProcessor"): | |
| """ | |
| Register this class with a given auto class. This should only be used for custom image processors as the ones | |
| in the library are already mapped with `AutoImageProcessor `. | |
| Args: | |
| auto_class (`str` or `type`, *optional*, defaults to `"AutoImageProcessor "`): | |
| The auto class to register this new image processor with. | |
| """ | |
| if not isinstance(auto_class, str): | |
| auto_class = auto_class.__name__ | |
| import transformers.models.auto as auto_module | |
| if not hasattr(auto_module, auto_class): | |
| raise ValueError(f"{auto_class} is not a valid auto class.") | |
| cls._auto_class = auto_class | |
| def fetch_images(self, image_url_or_urls: Union[str, list[str], list[list[str]]]): | |
| """ | |
| Convert a single or a list of urls into the corresponding `PIL.Image` objects. | |
| If a single url is passed, the return value will be a single object. If a list is passed a list of objects is | |
| returned. | |
| """ | |
| if isinstance(image_url_or_urls, list): | |
| return [self.fetch_images(x) for x in image_url_or_urls] | |
| elif isinstance(image_url_or_urls, str): | |
| return load_image(image_url_or_urls) | |
| elif is_valid_image(image_url_or_urls): | |
| return image_url_or_urls | |
| else: | |
| raise TypeError(f"only a single or a list of entries is supported but got type={type(image_url_or_urls)}") | |
| ImageProcessingMixin.push_to_hub = copy_func(ImageProcessingMixin.push_to_hub) | |
| if ImageProcessingMixin.push_to_hub.__doc__ is not None: | |
| ImageProcessingMixin.push_to_hub.__doc__ = ImageProcessingMixin.push_to_hub.__doc__.format( | |
| object="image processor", object_class="AutoImageProcessor", object_files="image processor file" | |
| ) | |