Restart from scratch.

src/bsv.bak/util.py

@@ -0,0 +1,118 @@
+# 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:
+        ...