Spaces:
Build error
Build error
| # SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. | |
| # SPDX-License-Identifier: Apache-2.0 | |
| # | |
| # 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 os | |
| import tempfile | |
| from contextlib import contextmanager | |
| from pathlib import Path | |
| from typing import Generator, Union | |
| from urllib.request import urlopen | |
| from cosmos_predict1.utils.easy_io.backends.base_backend import BaseStorageBackend | |
| class HTTPBackend(BaseStorageBackend): | |
| """HTTP and HTTPS storage bachend.""" | |
| def get(self, filepath: str) -> bytes: | |
| """Read bytes from a given ``filepath``. | |
| Args: | |
| filepath (str): Path to read data. | |
| Returns: | |
| bytes: Expected bytes object. | |
| Examples: | |
| >>> backend = HTTPBackend() | |
| >>> backend.get('http://path/of/file') | |
| b'hello world' | |
| """ | |
| return urlopen(filepath).read() | |
| def get_text(self, filepath, encoding="utf-8") -> str: | |
| """Read text from a given ``filepath``. | |
| Args: | |
| filepath (str): Path to read data. | |
| encoding (str): The encoding format used to open the ``filepath``. | |
| Defaults to 'utf-8'. | |
| Returns: | |
| str: Expected text reading from ``filepath``. | |
| Examples: | |
| >>> backend = HTTPBackend() | |
| >>> backend.get_text('http://path/of/file') | |
| 'hello world' | |
| """ | |
| return urlopen(filepath).read().decode(encoding) | |
| def get_local_path(self, filepath: str) -> Generator[Union[str, Path], None, None]: | |
| """Download a file from ``filepath`` to a local temporary directory, | |
| and return the temporary path. | |
| ``get_local_path`` is decorated by :meth:`contxtlib.contextmanager`. It | |
| can be called with ``with`` statement, and when exists from the | |
| ``with`` statement, the temporary path will be released. | |
| Args: | |
| filepath (str): Download a file from ``filepath``. | |
| Yields: | |
| Iterable[str]: Only yield one temporary path. | |
| Examples: | |
| >>> backend = HTTPBackend() | |
| >>> # After existing from the ``with`` clause, | |
| >>> # the path will be removed | |
| >>> with backend.get_local_path('http://path/of/file') as path: | |
| ... # do something here | |
| """ | |
| try: | |
| f = tempfile.NamedTemporaryFile(delete=False) | |
| f.write(self.get(filepath)) | |
| f.close() | |
| yield f.name | |
| finally: | |
| os.remove(f.name) | |