You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
118 lines
3.8 KiB
118 lines
3.8 KiB
# bsv - Backup, Synchronization, Versioning
|
|
# Copyright (C) 2023 Simon Boyé
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from datetime import UTC, datetime as DateTime, timedelta as TimeDelta
|
|
import os
|
|
from pathlib import Path
|
|
import platform
|
|
import stat
|
|
from typing import BinaryIO
|
|
|
|
|
|
EPOCH = DateTime(1970, 1, 1, tzinfo=UTC)
|
|
|
|
|
|
def time_from_timestamp_us(timestamp: int) -> DateTime:
|
|
return EPOCH + TimeDelta(microseconds=timestamp)
|
|
|
|
def timestamp_us_from_time(time: DateTime) -> int:
|
|
return (time.astimezone(UTC) - EPOCH) // TimeDelta(microseconds=1)
|
|
|
|
|
|
def read_exact(stream: BinaryIO, num_bytes: int) -> bytes:
|
|
data = stream.read(num_bytes)
|
|
if len(data) != num_bytes:
|
|
raise IOError(f"expected {num_bytes} bytes, got {len(data)}")
|
|
return data
|
|
|
|
def read_exact_or_eof(stream: BinaryIO, num_bytes: int) -> bytes | None:
|
|
data = stream.read(num_bytes)
|
|
if not data:
|
|
return None
|
|
if len(data) != num_bytes:
|
|
raise IOError(f"expected {num_bytes} bytes, got {len(data)}")
|
|
return data
|
|
|
|
|
|
def is_bsv_repository(path: Path) -> bool:
|
|
return (path / "bsv_repository.config").is_file()
|
|
|
|
|
|
def object_type_from_path(path: Path) -> bytes:
|
|
return object_type_from_mode(path.stat(follow_symlinks=False).st_mode)
|
|
|
|
def object_type_from_mode(mode: int) -> bytes:
|
|
if stat.S_ISLNK(mode):
|
|
return b"slnk"
|
|
elif stat.S_ISDIR(mode):
|
|
return b"tree"
|
|
elif stat.S_ISREG(mode):
|
|
return b"blob"
|
|
return b""
|
|
|
|
|
|
def default_bsv_config_path() -> Path:
|
|
path = Path(os.environ.get("BSV_CONFIG", ""))
|
|
if path and path.is_absolute() and path.is_file():
|
|
return path
|
|
for path in user_config_dirs():
|
|
if path.is_file():
|
|
return path
|
|
return user_config_home() / "bsv/config"
|
|
|
|
def default_local_repository_path() -> Path:
|
|
return user_data_home() / "bsv"
|
|
|
|
|
|
def user_data_home() -> Path:
|
|
if platform.system() in ("Windows", "Darwin", "Java"):
|
|
raise NotImplemented(f"{platform.system()} support not implemented yet")
|
|
else: # Assume Unix
|
|
path = Path(os.environ.get("XDG_DATA_HOME", ""))
|
|
if path and path.is_absolute():
|
|
return path
|
|
return Path.home() / ".local/share"
|
|
|
|
def user_config_home() -> Path:
|
|
if platform.system() in ("Windows", "Darwin", "Java"):
|
|
raise NotImplemented(f"{platform.system()} support not implemented yet")
|
|
else: # Assume Unix
|
|
path = Path(os.environ.get("XDG_CONFIG_HOME", ""))
|
|
if path and path.is_absolute():
|
|
return path
|
|
return Path.home() / ".config"
|
|
|
|
def user_config_dirs() -> list[Path]:
|
|
if platform.system() in ("Windows", "Darwin", "Java"):
|
|
raise NotImplemented(f"{platform.system()} support not implemented yet")
|
|
else: # Assume Unix
|
|
paths = list(filter(Path.is_absolute, map(Path, (os.environ.get("XDG_CONFIG_DIRS") or "/etc/xdg").split(":"))))
|
|
return [user_config_home()] + paths
|
|
|
|
|
|
class Hash(ABC):
|
|
name: str
|
|
digest_size: int
|
|
|
|
@abstractmethod
|
|
def update(self, *data: bytes | bytearray | memoryview):
|
|
...
|
|
|
|
@abstractmethod
|
|
def digest(self) -> bytes:
|
|
...
|
|
|