wip

system/hardware/tici/casync.py

@@ -7,7 +7,7 @@ import sys
 import time
 from abc import ABC, abstractmethod
 from collections import defaultdict, namedtuple
-from typing import Callable, Dict, List, Optional, Tuple
+from collections.abc import Callable
 
 import requests
 from Crypto.Hash import SHA512
@@ -28,7 +28,7 @@ CHUNK_DOWNLOAD_RETRIES = 3
 CAIBX_DOWNLOAD_TIMEOUT = 120
 
 Chunk = namedtuple('Chunk', ['sha', 'offset', 'length'])
-ChunkDict = Dict[bytes, Chunk]
+ChunkDict = dict[bytes, Chunk]
 
 
 class ChunkReader(ABC):
@@ -83,7 +83,7 @@ class RemoteChunkReader(ChunkReader):
     return decompressor.decompress(contents)
 
 
-def parse_caibx(caibx_path: str) -> List[Chunk]:
+def parse_caibx(caibx_path: str) -> list[Chunk]:
   """Parses the chunks from a caibx file. Can handle both local and remote files.
   Returns a list of chunks with hash, offset and length"""
   caibx: io.BufferedIOBase
@@ -132,7 +132,7 @@ def parse_caibx(caibx_path: str) -> List[Chunk]:
   return chunks
 
 
-def build_chunk_dict(chunks: List[Chunk]) -> ChunkDict:
+def build_chunk_dict(chunks: list[Chunk]) -> ChunkDict:
   """Turn a list of chunks into a dict for faster lookups based on hash.
   Keep first chunk since it's more likely to be already downloaded."""
   r = {}
@@ -142,11 +142,11 @@ def build_chunk_dict(chunks: List[Chunk]) -> ChunkDict:
   return r
 
 
-def extract(target: List[Chunk],
-            sources: List[Tuple[str, ChunkReader, ChunkDict]],
+def extract(target: list[Chunk],
+            sources: list[tuple[str, ChunkReader, ChunkDict]],
             out_path: str,
-            progress: Optional[Callable[[int], None]] = None):
-  stats: Dict[str, int] = defaultdict(int)
+            progress: Callable[[int], None] = None):
+  stats: dict[str, int] = defaultdict(int)
 
   mode = 'rb+' if os.path.exists(out_path) else 'wb'
   with open(out_path, mode) as out:
@@ -181,7 +181,7 @@ def extract(target: List[Chunk],
   return stats
 
 
-def print_stats(stats: Dict[str, int]):
+def print_stats(stats: dict[str, int]):
   total_bytes = sum(stats.values())
   print(f"Total size: {total_bytes / 1024 / 1024:.2f} MB")
   for name, total in stats.items():