module ptwm.stores

def decode_member_tensor(registry_json: str, tensor_json: str, members: dict[str, bytes]) -> torch.Tensor

def detect_format(path: str | Path) -> str | None

def explode_model(blob: bytes) -> ExplodedModel

class ExplodedModel:

def __init__(self, registry_json: str, tensors: list[tuple[str, str, str]], members: dict[str, bytes]) -> None

def key(self, name: str) -> str

def member_size(self, name: str) -> int

members = members

def names(self) -> list[str]

registry_json = registry_json

def tensor_json(self, name: str) -> str

tensors = tensors

def jcs_canonicalize(obj: Any) -> str

class LmdbWeightStore:

def __enter__(self) -> Self

def __exit__(self, *exc: object = ()) -> None

def __init__(self, path: str | Path, map_size: int = _DEFAULT_MAP_SIZE, readonly: bool = True) -> None

def close(self) -> None

def get_bytes(self, name: str) -> bytes

def get_tensor(self, name: str) -> torch.Tensor

layout: str

def names(self) -> list[str]

def open(cls, path: str | Path, readonly: bool = True) -> LmdbWeightStore

def put_tensor(self, name: str, raw: bytes, dtype: Any, shape: tuple[int, ...], delta_reference: str | None = None, method_hint: int = 3) -> None

def stream_tensors(self, names: Iterable[str] | None = None) -> Iterator[tuple[str, torch.Tensor]]

def open_store(path: str | Path) -> WeightStore

def raw_from_members(registry_json: str, tensor_json: str, members: dict[str, bytes]) -> bytes

SHARDMETA_MEMBER = '__shardmeta__.json'

WDS_FORMAT = 'ptwm-wds'

class WebDatasetWeightStore:

def __enter__(self) -> Self

def __exit__(self, *exc: object = ()) -> None

def __init__(self, path: str | Path) -> None

def close(self) -> None

def get_bytes(self, name: str) -> bytes

def get_tensor(self, name: str) -> torch.Tensor

def names(self) -> list[str]

def open(cls, path: str | Path) -> WebDatasetWeightStore

def stream_tensors(self, names: Iterable[str] | None = None) -> Iterator[tuple[str, torch.Tensor]]

def webdataset_pipeline(self) -> Any

class WeightStore(Protocol):

def get_bytes(self, name: str) -> bytes

def get_tensor(self, name: str) -> torch.Tensor

def names(self) -> list[str]

def stream_tensors(self, names: Iterable[str] | None = None) -> Iterator[tuple[str, torch.Tensor]]

def write_lmdb(path: str | Path, tensors: Sequence[tuple[str, bytes, Any, tuple[int, ...]]], layout: str = 'blob', map_size: int = _DEFAULT_MAP_SIZE, method_hint: int = 3, chains_per_tensor: list[list[bytes]] | None = None) -> Path

def write_webdataset(out_dir: str | Path, tensors: Sequence[tuple[str, bytes, Any, tuple[int, ...]]], max_shard_size: int = _DEFAULT_MAX_SHARD_SIZE, method_hint: int = 3, chains_per_tensor: list[list[bytes]] | None = None) -> Path