Python Coding and Data Structures Questions

import threading
from collections import deque
from typing import Any

class BoundedQueue:
    def __init__(self, maxsize: int):
        if maxsize <= 0:
            raise ValueError("maxsize must be > 0")
        self.maxsize = maxsize
        self.buffer = deque()
        self.lock = threading.Lock()
        self.not_empty = threading.Condition(self.lock)
        self.not_full = threading.Condition(self.lock)

    def put(self, item: Any):
        """Block if full, then append item."""
        with self.not_full:
            while len(self.buffer) >= self.maxsize:    # predicate loop for spurious wakeups
                self.not_full.wait()
            self.buffer.append(item)
            # Notify one waiter on not_empty that an item is available
            self.not_empty.notify()

    def get(self) -> Any:
        """Block if empty, then pop left and return item."""
        with self.not_empty:
            while len(self.buffer) == 0:
                self.not_empty.wait()
            item = self.buffer.popleft()
            # Notify one waiter on not_full that space is available
            self.not_full.notify()
            return item

    def qsize(self) -> int:
        with self.lock:
            return len(self.buffer)

    def empty(self) -> bool:
        with self.lock:
            return len(self.buffer) == 0

    def full(self) -> bool:
        with self.lock:
            return len(self.buffer) >= self.maxsize

import chardet

def robust_read_text(path, sample_size=8192):
    with open(path, 'rb') as f:
        raw = f.read(sample_size)
        if raw.startswith(b'\xef\xbb\xbf'):
            enc = 'utf-8-sig'
        else:
            guess = chardet.detect(raw)
            enc = guess['encoding'] or 'latin-1'
    try:
        with open(path, 'r', encoding=enc, errors='strict', newline='') as f:
            return f.read()
    except UnicodeDecodeError:
        # fallback to safe decode preserving problematic bytes
        with open(path, 'r', encoding='utf-8', errors='surrogateescape') as f:
            return f.read()

import bisect
import hashlib
from typing import Dict, List, Tuple

def _hash(value: str) -> int:
    # 32-bit unsigned integer hash
    return int(hashlib.md5(value.encode()).hexdigest()[:8], 16)

class ConsistentHashRing:
    def __init__(self, replicas: int = 100):
        self.replicas = replicas
        self._ring: List[int] = []               # sorted hashes (tokens)
        self._ring_map: Dict[int, str] = {}      # token_hash -> node_id

    def add_node(self, node_id: str):
        """Add physical node with 'replicas' virtual nodes."""
        for i in range(self.replicas):
            token = _hash(f"{node_id}:{i}")
            if token in self._ring_map:
                # rare collision; skip or continue probing
                continue
            bisect.insort(self._ring, token)
            self._ring_map[token] = node_id

    def remove_node(self, node_id: str):
        """Remove a node and all its virtual nodes."""
        to_remove = []
        for i in range(self.replicas):
            token = _hash(f"{node_id}:{i}")
            if token in self._ring_map and self._ring_map[token] == node_id:
                to_remove.append(token)
        for token in to_remove:
            idx = bisect.bisect_left(self._ring, token)
            if idx < len(self._ring) and self._ring[idx] == token:
                self._ring.pop(idx)
            self._ring_map.pop(token, None)

    def get_node(self, key: str) -> Tuple[str, int]:
        """Return (node_id, token_hash) responsible for this key."""
        if not self._ring:
            raise KeyError("No nodes in ring")
        token = _hash(key)
        idx = bisect.bisect(self._ring, token)
        if idx == len(self._ring):
            idx = 0  # wrap around
        chosen = self._ring[idx]
        return self._ring_map[chosen], chosen

import numpy as np
import json
# metadata
meta = {"shape": (1000000, 128), "dtype": "float32", "created_by": "me"}
with open("array.meta.json","w") as f:
    json.dump(meta, f)

shape = meta["shape"]; dtype = np.dtype(meta["dtype"])
# create memmap file (reserves disk without loading)
arr = np.memmap("data.dat", dtype=dtype, mode="w+", shape=shape)

# stream-write in chunks (generator yields chunk arrays)
def stream_generator():
    for i in range(0, shape[0], 1000):
        yield np.random.rand(min(1000, shape[0]-i), shape[1]).astype(dtype)

offset = 0
for chunk in stream_generator():
    arr[offset:offset+len(chunk)] = chunk
    arr.flush()  # ensure data persisted
    offset += len(chunk)
# later, read with memmap (no full RAM)
arr_read = np.memmap("data.dat", dtype=dtype, mode="r", shape=shape)
print(arr_read[0:5])

import h5py
with h5py.File("data.h5","w") as f:
    dset = f.create_dataset("X", shape=shape, dtype=dtype,
                            chunks=(1000, shape[1]), compression="gzip")
    f.attrs.update(meta)  # attach metadata
    offset = 0
    for chunk in stream_generator():
        dset[offset:offset+len(chunk)] = chunk
        offset += len(chunk)