Engine of OpenMMLab projects for training deep learning models based on PyTorch with large-scale training frameworks, configuration management, and monitoring capabilities
—
Unified file operations supporting multiple storage backends including local filesystem, HTTP, Petrel, LMDB, and Memcached with transparent backend switching and format-specific handlers. This system enables seamless file operations across different storage environments.
Unified client for file operations across different storage backends with transparent backend switching.
class FileClient:
def __init__(self, backend: str = 'disk', **kwargs):
"""
Initialize FileClient with specified backend.
Parameters:
- backend: Backend type ('disk', 'petrel', 'memcached', 'lmdb', 'http')
- **kwargs: Backend-specific configuration options
"""
def get(self, filepath: str) -> bytes:
"""
Read file content as bytes.
Parameters:
- filepath: Path to file
Returns:
File content as bytes
"""
def get_text(self, filepath: str, encoding: str = 'utf-8') -> str:
"""
Read file content as text.
Parameters:
- filepath: Path to file
- encoding: Text encoding
Returns:
File content as string
"""
def put(self, obj: bytes, filepath: str):
"""
Write bytes to file.
Parameters:
- obj: Bytes to write
- filepath: Destination file path
"""
def put_text(self, obj: str, filepath: str, encoding: str = 'utf-8'):
"""
Write text to file.
Parameters:
- obj: Text to write
- filepath: Destination file path
- encoding: Text encoding
"""
def exists(self, filepath: str) -> bool:
"""
Check if file exists.
Parameters:
- filepath: File path to check
Returns:
True if file exists, False otherwise
"""
def isdir(self, filepath: str) -> bool:
"""
Check if path is directory.
Parameters:
- filepath: Path to check
Returns:
True if path is directory, False otherwise
"""
def isfile(self, filepath: str) -> bool:
"""
Check if path is file.
Parameters:
- filepath: Path to check
Returns:
True if path is file, False otherwise
"""
def list_dir_or_file(self, dir_path: str, list_dir: bool = True, list_file: bool = True, suffix: str = None, recursive: bool = False) -> list:
"""
List directory contents.
Parameters:
- dir_path: Directory path
- list_dir: Whether to list directories
- list_file: Whether to list files
- suffix: File suffix filter
- recursive: Whether to search recursively
Returns:
List of paths
"""Various storage backend implementations for different storage systems.
class BaseStorageBackend:
def get(self, filepath: str) -> bytes: ...
def get_text(self, filepath: str, encoding: str = 'utf-8') -> str: ...
def put(self, obj: bytes, filepath: str): ...
def put_text(self, obj: str, filepath: str, encoding: str = 'utf-8'): ...
def exists(self, filepath: str) -> bool: ...
def isdir(self, filepath: str) -> bool: ...
def isfile(self, filepath: str) -> bool: ...
class LocalBackend(BaseStorageBackend):
def __init__(self): ...
class HardDiskBackend(BaseStorageBackend):
def __init__(self): ...
class HTTPBackend(BaseStorageBackend):
def __init__(self): ...
class PetrelBackend(BaseStorageBackend):
def __init__(self, path_mapping: dict = None, enable_mc: bool = True, conf_path: str = None): ...
class MemcachedBackend(BaseStorageBackend):
def __init__(self, server_list_cfg: str, client_cfg: str, sys_path: str = None): ...
class LmdbBackend(BaseStorageBackend):
def __init__(self, db_path: str, readonly: bool = True, lock: bool = False, readahead: bool = False, **kwargs): ...Convenient high-level functions for common file operations with automatic backend selection.
def load(file: str, file_format: str = None, backend: str = 'disk', **kwargs):
"""
Load data from file with automatic format detection.
Parameters:
- file: File path or file-like object
- file_format: File format ('json', 'yaml', 'pkl')
- backend: Storage backend
- **kwargs: Additional arguments
Returns:
Loaded data
"""
def dump(obj, file: str = None, file_format: str = None, backend: str = 'disk', **kwargs):
"""
Dump data to file with automatic format detection.
Parameters:
- obj: Object to dump
- file: File path or file-like object
- file_format: File format ('json', 'yaml', 'pkl')
- backend: Storage backend
- **kwargs: Additional arguments
Returns:
Dumped string if file is None
"""
def exists(filepath: str, backend: str = 'disk') -> bool:
"""
Check if file exists.
Parameters:
- filepath: File path
- backend: Storage backend
Returns:
True if file exists
"""
def isdir(filepath: str, backend: str = 'disk') -> bool:
"""
Check if path is directory.
Parameters:
- filepath: Path to check
- backend: Storage backend
Returns:
True if path is directory
"""
def isfile(filepath: str, backend: str = 'disk') -> bool:
"""
Check if path is file.
Parameters:
- filepath: Path to check
- backend: Storage backend
Returns:
True if path is file
"""
def get(filepath: str, backend: str = 'disk') -> bytes:
"""
Get file content as bytes.
Parameters:
- filepath: File path
- backend: Storage backend
Returns:
File content as bytes
"""
def get_text(filepath: str, encoding: str = 'utf-8', backend: str = 'disk') -> str:
"""
Get file content as text.
Parameters:
- filepath: File path
- encoding: Text encoding
- backend: Storage backend
Returns:
File content as string
"""
def put(obj: bytes, filepath: str, backend: str = 'disk'):
"""
Put bytes to file.
Parameters:
- obj: Bytes to write
- filepath: Destination path
- backend: Storage backend
"""
def put_text(obj: str, filepath: str, encoding: str = 'utf-8', backend: str = 'disk'):
"""
Put text to file.
Parameters:
- obj: Text to write
- filepath: Destination path
- encoding: Text encoding
- backend: Storage backend
"""Functions for copying files and directories across different backends.
def copyfile(src: str, dst: str, backend: str = 'disk'):
"""
Copy file from source to destination.
Parameters:
- src: Source file path
- dst: Destination file path
- backend: Storage backend
"""
def copyfile_from_local(src: str, dst: str, backend: str = 'disk'):
"""
Copy file from local to remote backend.
Parameters:
- src: Local source file path
- dst: Remote destination path
- backend: Remote storage backend
"""
def copyfile_to_local(src: str, dst: str, backend: str = 'disk'):
"""
Copy file from remote backend to local.
Parameters:
- src: Remote source file path
- dst: Local destination path
- backend: Remote storage backend
"""
def copytree(src: str, dst: str, backend: str = 'disk'):
"""
Copy directory tree.
Parameters:
- src: Source directory path
- dst: Destination directory path
- backend: Storage backend
"""Extensible system for handling different file formats with registration support.
class BaseFileHandler:
def load_from_fileobj(self, file, **kwargs): ...
def dump_to_fileobj(self, obj, file, **kwargs): ...
def load_from_path(self, filepath: str, **kwargs): ...
def dump_to_path(self, obj, filepath: str, **kwargs): ...
class JsonHandler(BaseFileHandler):
def load_from_fileobj(self, file, **kwargs): ...
def dump_to_fileobj(self, obj, file, **kwargs): ...
class PickleHandler(BaseFileHandler):
def load_from_fileobj(self, file, **kwargs): ...
def dump_to_fileobj(self, obj, file, **kwargs): ...
class YamlHandler(BaseFileHandler):
def load_from_fileobj(self, file, **kwargs): ...
def dump_to_fileobj(self, obj, file, **kwargs): ...
def register_handler(handler: BaseFileHandler, file_formats: list):
"""
Register file format handler.
Parameters:
- handler: Handler instance
- file_formats: List of supported file formats
"""
def register_backend(name: str, backend: BaseStorageBackend = None, force: bool = False, prefixes: str = None):
"""
Register storage backend.
Parameters:
- name: Backend name
- backend: Backend class or instance
- force: Whether to override existing backend
- prefixes: URL prefixes handled by backend
"""Utilities for loading structured data from files.
def list_from_file(filename: str, prefix: str = '', offset: int = 0, max_num: int = 0, encoding: str = 'utf-8', backend: str = 'disk') -> list:
"""
Load list from file with each line as an element.
Parameters:
- filename: File path
- prefix: Prefix to add to each line
- offset: Line offset to start reading
- max_num: Maximum number of lines to read
- encoding: Text encoding
- backend: Storage backend
Returns:
List of lines
"""
def dict_from_file(filename: str, key_type: type = str, encoding: str = 'utf-8', backend: str = 'disk') -> dict:
"""
Load dictionary from file.
Parameters:
- filename: File path
- key_type: Type to convert keys
- encoding: Text encoding
- backend: Storage backend
Returns:
Dictionary loaded from file
"""from mmengine import fileio
# Load JSON data
data = fileio.load('config.json')
# Save data as JSON
fileio.dump(data, 'output.json')
# Check if file exists
if fileio.exists('data.pkl'):
data = fileio.load('data.pkl')
# Read text file
content = fileio.get_text('readme.txt')from mmengine.fileio import FileClient
# Local filesystem
client = FileClient('disk')
data = client.get('local_file.txt')
# HTTP backend
client = FileClient('http')
content = client.get('https://example.com/data.json')
# Petrel backend (for cloud storage)
client = FileClient('petrel', path_mapping={'s3://bucket': '/path/to/local'})
data = client.get('s3://bucket/data.pkl')from mmengine import fileio
# Copy from local to remote
fileio.copyfile_from_local('local_data.json', 's3://bucket/remote_data.json', backend='petrel')
# Copy from remote to local
fileio.copyfile_to_local('s3://bucket/remote_data.json', 'local_copy.json', backend='petrel')
# Copy entire directory
fileio.copytree('local_dir/', 's3://bucket/remote_dir/', backend='petrel')Install with Tessl CLI
npx tessl i tessl/pypi-mmengine