copc_io.py

"""COPC (Cloud Optimized Point Cloud) reading + octree streaming for the web viewer.

COPC is LAZ with an embedded EPT octree, so the octree IS the LOD hierarchy —
no separate index needs building. This module exposes:

  - is_copc(path)               : detect a COPC file
  - open_copc(path)             : cached CopcReader (+ derived stats)
  - copc_meta(path)             : header + octree root geometry (JSON-able)
  - copc_hierarchy(path, ...)   : serialized octree nodes (keys, counts, bounds)
  - copc_nodes_binary(path,keys): points for a set of nodes, in viewer binary
  - ensure_copc(src)            : convert a non-COPC cloud to .copc.laz

The frontend (static/js/copc-lod.js) walks the hierarchy, picks a view-dependent
cut through the octree, and fetches only the nodes it needs.

A single fixed coordOffset (= octree center) is used for every node so all
per-node geometries share one coordinate frame (see arrays_to_binary's offset).
"""

import os
import queue
import shutil
import contextlib
import threading
import uuid

import numpy as np

from pointcloud_io import arrays_to_binary


# ── Reader cache ────────────────────────────────────────
# CopcReader holds an open file handle and the root hierarchy page; every camera
# move triggers several node fetches, so re-opening per request would be slow and
# leak descriptors. Cache a few readers keyed by realpath, invalidated on mtime.
_CACHE_MAX = 8
_readers = {}                       # realpath -> entry dict
_readers_lock = threading.Lock()


def _node_key_str(key):
    return f"{key.level}-{key.x}-{key.y}-{key.z}"


def is_copc(path):
    """True if *path* is a readable COPC file (has the COPC VLR)."""
    if not path or not os.path.isfile(path):
        return False
    ext = os.path.splitext(path)[1].lower()
    if ext not in ('.laz', '.las'):
        return False
    try:
        from laspy import CopcReader
        with open(path, 'rb') as fp:
            CopcReader.open(fp)
        return True
    except Exception:
        return False


def open_copc(path):
    """Return a cached entry {reader, has_rgb, intensity_max} for *path*."""
    from laspy import CopcReader

    realpath = os.path.realpath(path)
    mtime = os.path.getmtime(realpath)
    evicted = []
    is_owner = False
    with _readers_lock:
        entry = _readers.get(realpath)
        if entry and entry['mtime'] != mtime:
            # File replaced on disk: retire the stale entry like an eviction.
            _readers.pop(realpath)
            entry['stale'] = True
            evicted.append(entry)
            entry = None
        if entry is None:
            # Only a placeholder is registered under the lock — the actual
            # open (~0.3s) and scale estimate (seconds) run OUTSIDE it, so
            # concurrent opens of other files aren't serialized behind I/O.
            # Late arrivals for the same path wait on 'ready' instead.
            is_owner = True
            entry = {
                'realpath': realpath,
                'mtime': mtime,
                'ready': threading.Event(),  # set once reader/scales are filled in
                'error': None,               # open failure, re-raised to waiters
                'stale': False,              # evicted: close handles on return, not reuse
                'lock': threading.Lock(),   # guards one-time octree build + pool growth
                'reader_pool': queue.Queue(),  # idle CopcReader handles, borrowed per fetch
                'pool_count': 0,            # handles created so far (≤ pool_max)
                'pool_max': 16,             # cap concurrent handles (≈ server cores)
                'thread_readers': [],       # all handles spawned (borrowed ones close lazily)
            }
            # Evict oldest if over capacity. Only mark it stale here; idle
            # handles are closed outside the lock, and handles currently
            # borrowed by other threads are closed when they're returned
            # (see _borrow_reader) — never while a read is in flight.
            if len(_readers) >= _CACHE_MAX:
                old_key = next(iter(_readers))
                old = _readers.pop(old_key)
                old['stale'] = True
                evicted.append(old)
            _readers[realpath] = entry

    for old in evicted:
        _close_evicted(old)

    if not is_owner:
        entry['ready'].wait()
        if entry['error'] is not None:
            raise entry['error']
        return entry

    try:
        reader = CopcReader.open(open(realpath, 'rb'))
        dims = set(reader.header.point_format.dimension_names)
        has_rgb = {'red', 'green', 'blue'} <= dims
        intensity_max, rgb_scale = _estimate_norm_scales(reader, has_rgb)
        # HandheldSLAM stores each photo-baked point's footprint (texel spacing,
        # 0.1 mm units) in point_source_id and marks the file with its system id;
        # elsewhere point_source_id is a flight line number and means nothing here.
        splat = (str(reader.header.system_identifier).strip('\x00 ').startswith('HandheldSLAM')
                 and 'point_source_id' in dims)
    except Exception as e:
        entry['error'] = e
        with _readers_lock:
            if _readers.get(realpath) is entry:
                del _readers[realpath]
        entry['ready'].set()
        raise
    entry['reader'] = reader           # used only for header/octree build (single-threaded)
    entry['has_rgb'] = has_rgb
    entry['intensity_max'] = intensity_max
    entry['rgb_scale'] = rgb_scale
    entry['splat'] = splat
    entry['ready'].set()
    return entry


def _close_evicted(old):
    """Close an evicted entry's IDLE handles (the ones sitting in the pool).

    Borrowed handles are mid-read on other threads — closing them here would
    corrupt those reads, so _borrow_reader closes them on return via the
    stale flag set at eviction time."""
    pool = old.get('reader_pool')
    while pool is not None:
        try:
            pool.get_nowait().close()
        except queue.Empty:
            break
        except Exception:
            pass
    reader = old.get('reader')
    if reader is not None:
        # entry lock: don't yank the header reader out from under an octree build.
        with old['lock']:
            try:
                reader.close()
            except Exception:
                pass


@contextlib.contextmanager
def _borrow_reader(entry):
    """Borrow a CopcReader from a fixed pool, returning it when done.

    laspy's CopcReader seeks+reads a single shared file handle, so two threads
    fetching through one reader race on the file position and corrupt each
    other's reads ('LazrsError: failed to fill whole buffer'). Each fetch needs
    its own handle — but opening one costs ~0.3s (it reads the root page), and
    Werkzeug's threaded server spawns a NEW thread per request, so a per-thread
    handle was re-opened on essentially every request: the open dominated while
    the actual decompress was ~3ms. A bounded pool opens at most pool_max handles
    total and recycles them, so node offsets stay absolute (every handle reads
    the same cached octree correctly) and lazrs decompress runs across them."""
    from laspy import CopcReader
    pool = entry['reader_pool']
    reader = None
    try:
        reader = pool.get_nowait()
    except queue.Empty:
        with entry['lock']:
            if entry['pool_count'] < entry['pool_max']:
                reader = CopcReader.open(open(entry['realpath'], 'rb'))
                entry['pool_count'] += 1
                entry['thread_readers'].append(reader)
        if reader is None:           # at cap → wait for a busy handle to return
            reader = pool.get()
    try:
        yield reader
    finally:
        if entry.get('stale'):
            # Entry was evicted while we were reading: close instead of
            # returning to a pool nobody will drain again.
            try:
                reader.close()
            except Exception:
                pass
        else:
            pool.put(reader)


def _estimate_norm_scales(reader, has_rgb):
    """(intensity_max, rgb_scale) for consistent per-node normalization.

    A single value must normalize every node (matching _read_las, which uses
    file-wide values) or node colors won't agree — judging 8- vs 16-bit RGB per
    node makes dark nodes of a 16-bit file render up to 257× brighter than their
    neighbors. Estimate from coarse levels — they sample the whole cloud and are
    cheap to read."""
    intensity_max, rgb_scale = 65535.0, 65535.0
    try:
        pts = reader.query(level=range(0, 4))
        arr = np.asarray(pts.intensity, dtype=np.float64)
        m = float(arr.max()) if len(arr) else 0.0
        intensity_max = m if m > 0 else 1.0
        if has_rgb:
            rmax = float(np.asarray(pts.red).max()) if len(arr) else 0.0
            rgb_scale = 65535.0 if rmax > 255 else 255.0
    except Exception:
        pass
    return intensity_max, rgb_scale


# Paths whose octree warm-up thread is currently running — repeated meta
# requests during a long build would otherwise pile up threads all blocking
# on the same entry lock.
_warming = set()
_warming_lock = threading.Lock()


def warm_octree_async(path):
    """Kick off the (one-time, ~seconds) octree build in the background.

    The first hierarchy/node request otherwise pays the full load_octree_for_query
    walk (tens of seconds on big files) while the user stares at an empty view.
    Calling this when meta is served — well before the client asks for the
    hierarchy — lets the build overlap the round-trip so it's usually cached by
    the time it's needed. Safe to call repeatedly: a build already in progress
    is skipped, and _get_octree double-checks the cache."""
    key = os.path.realpath(path)
    with _warming_lock:
        if key in _warming:
            return
        _warming.add(key)

    def run():
        try:
            _get_octree(open_copc(path))
        except Exception:
            pass
        finally:
            with _warming_lock:
                _warming.discard(key)
    threading.Thread(target=run, daemon=True).start()


def copc_meta(path):
    """Header + octree root geometry, JSON-serializable."""
    entry = open_copc(path)
    # Start building the octree now so the client's hierarchy request (next
    # round-trip) finds it cached instead of blocking on a multi-second walk.
    warm_octree_async(path)
    reader = entry['reader']
    h = reader.header
    ci = reader.copc_info
    center = [float(ci.center[0]), float(ci.center[1]), float(ci.center[2])]
    try:
        import config
        point_budget = int(getattr(config, 'COPC_POINT_BUDGET', 5_000_000))
    except Exception:
        point_budget = 5_000_000
    return {
        'mode': 'copc',
        'is_copc': True,
        'point_count': int(h.point_count),
        'point_format': int(h.point_format.id),
        'has_rgb': bool(entry['has_rgb']),
        'intensityMax': float(entry['intensity_max']),
        'pointBudget': point_budget,
        # HandheldSLAM writes returns the camera never saw as RGB (0,0,0); its photo
        # pixels are never exactly black. The viewer shades those by intensity.
        'uncoloredBlack': bool(entry.get('splat')),
        'mins': [float(v) for v in h.mins],
        'maxs': [float(v) for v in h.maxs],
        'root': {
            'center': center,
            'halfsize': float(ci.halfsize),
            'spacing': float(ci.spacing),
        },
        # Fixed centering origin shared by every node payload.
        'coordOffset': center,
    }


def _cache_dir():
    """Directory for persisted octree caches (under DATA_DIR, not the maps tree)."""
    try:
        import config
        base = config.DATA_DIR
    except Exception:
        base = os.path.expanduser('~/webpointcloud')
    d = os.path.join(base, 'cache', 'octree')
    os.makedirs(d, exist_ok=True)
    return d


def _octree_cache_path(realpath, mtime):
    import hashlib
    h = hashlib.sha1(realpath.encode('utf-8')).hexdigest()[:16]
    return os.path.join(_cache_dir(), f'{h}_{int(mtime)}.pkl')


def _load_octree_cache(cpath):
    import pickle
    try:
        if os.path.exists(cpath):
            with open(cpath, 'rb') as f:
                return pickle.load(f)
    except Exception:
        pass
    return None


def _save_octree_cache(cpath, octree):
    import pickle
    import glob
    try:
        tmp = cpath + '.tmp'
        with open(tmp, 'wb') as f:
            pickle.dump(octree, f, protocol=5)
        os.replace(tmp, cpath)
        # Drop stale caches for the same file (older mtimes).
        stem = cpath.rsplit('_', 1)[0]
        for old in glob.glob(stem + '_*.pkl'):
            if old != cpath:
                try:
                    os.remove(old)
                except OSError:
                    pass
    except Exception:
        pass


def _get_octree(entry):
    """Build (once, cached on the reader entry) the full key→OctreeNode map.

    load_octree_for_query walks the whole hierarchy and lazily pulls child pages
    from disk — costly (~15s on a big file). Three tiers of caching avoid paying
    it twice:
      1. in-memory on the reader entry (every fetch this process serves),
      2. on-disk pickle keyed by realpath+mtime (survives restarts; ~4s to load
         vs ~15s to rebuild — child links are stripped first since they're only
         used during the walk, not for fetching, which keeps the pickle small),
      3. otherwise a fresh walk, then persisted for next time."""
    octree = entry.get('octree')
    if octree is not None:
        return octree
    # Build once under the lock (double-checked). The build reads heavily from
    # reader.source; serializing it keeps it from racing concurrent node fetches.
    with entry['lock']:
        octree = entry.get('octree')
        if octree is not None:
            return octree
        cpath = _octree_cache_path(entry['realpath'], entry['mtime'])
        octree = _load_octree_cache(cpath)
        if octree is None:
            from laspy.copc import load_octree_for_query
            reader = entry['reader']
            nodes = load_octree_for_query(
                reader.source, reader.copc_info, reader.root_page,
                query_bounds=None, level_range=range(0, 32),
            )
            # Child links are only needed to walk the hierarchy; dropping them
            # makes the pickle ~8× smaller and far faster to load.
            for nd in nodes:
                nd.childs = []
            octree = {_node_key_str(nd.key): nd for nd in nodes}
            _save_octree_cache(cpath, octree)
        entry['octree'] = octree
        return octree


def copc_hierarchy(path, max_depth=32):
    """Serialize the octree node list reachable up to *max_depth* levels.

    Returns {"nodes": [{key, level, pointCount, mins, maxs}, ...]} with bounds in
    real-world (UTM) coordinates."""
    octree = _get_octree(open_copc(path))
    out = []
    for nd in octree.values():
        if nd.point_count <= 0 or nd.key.level >= int(max_depth):
            continue
        b = nd.bounds
        out.append({
            'key': _node_key_str(nd.key),
            'level': int(nd.key.level),
            'pointCount': int(nd.point_count),
            'mins': [float(b.mins[0]), float(b.mins[1]), float(b.mins[2])],
            'maxs': [float(b.maxs[0]), float(b.maxs[1]), float(b.maxs[2])],
        })
    return {'nodes': out}


def _pack_node_points(pts, entry, center):
    """Normalize a node's points (matching pointcloud_io._read_las) and pack into
    the viewer binary format, centered by the fixed octree center."""
    n = len(pts)
    x = np.asarray(pts.x, dtype=np.float64)
    y = np.asarray(pts.y, dtype=np.float64)
    z = np.asarray(pts.z, dtype=np.float64)

    imax = entry['intensity_max']
    intensity = np.asarray(pts.intensity, dtype=np.float32)
    if imax > 0:
        intensity = intensity / imax
    np.clip(intensity, 0.0, 1.0, out=intensity)

    if entry['has_rgb']:
        r = np.asarray(pts.red, dtype=np.float32)
        g = np.asarray(pts.green, dtype=np.float32)
        b = np.asarray(pts.blue, dtype=np.float32)
        scale = entry['rgb_scale']   # file-global 8/16-bit judgment, not per-node
        r /= scale; g /= scale; b /= scale
        np.clip(r, 0.0, 1.0, out=r); np.clip(g, 0.0, 1.0, out=g); np.clip(b, 0.0, 1.0, out=b)
    else:
        r = np.full(n, 0.5, dtype=np.float32)
        g = np.full(n, 0.5, dtype=np.float32)
        b = np.full(n, 0.5, dtype=np.float32)

    classification = np.asarray(pts.classification, dtype=np.float32)
    splat = (np.asarray(pts.point_source_id, dtype=np.float32) * 1e-4
             if entry.get('splat') else None)

    return arrays_to_binary(
        x, y, z, intensity, r, g, b, n,
        offset=[float(center[0]), float(center[1]), float(center[2])],
        classification=classification,
        splat=splat,
    )


def copc_preview_binary(path, max_points=3_000_000):
    """Coarse whole-extent sample as ONE legacy viewer binary.

    Walks octree levels top-down, keeping whole levels while the running total
    fits max_points (level 0 is always included). The compare overlay renders
    this as a static cloud — it needs full spatial coverage at bounded size,
    not the streamed LOD."""
    entry = open_copc(path)
    octree = _get_octree(entry)
    per_level = {}
    for nd in octree.values():
        if nd.point_count > 0:
            per_level[nd.key.level] = per_level.get(nd.key.level, 0) + nd.point_count
    total, max_level = 0, -1
    for lvl in sorted(per_level):
        if max_level >= 0 and total + per_level[lvl] > int(max_points):
            break
        total += per_level[lvl]
        max_level = lvl
    sel = [nd for nd in octree.values()
           if nd.point_count > 0 and nd.key.level <= max_level]
    with _borrow_reader(entry) as reader:  # pooled handle: no shared-seek race
        pts = reader._fetch_and_decompress_points_of_nodes(sel)
        center = reader.copc_info.center
    return _pack_node_points(pts, entry, center)


def copc_nodes_binary(path, keys):
    """Points for the requested node *keys* as one combined viewer binary."""
    entry = open_copc(path)
    octree = _get_octree(entry)
    want = list(dict.fromkeys(keys))
    sel = [octree[k] for k in want if k in octree]
    if not sel:
        return arrays_to_binary([], [], [], [], [], [], [], 0,
                                offset=[0.0, 0.0, 0.0])
    with _borrow_reader(entry) as reader:  # pooled handle: no shared-seek race
        pts = reader._fetch_and_decompress_points_of_nodes(sel)
        center = reader.copc_info.center
    return _pack_node_points(pts, entry, center)


def copc_nodes_multiblob(path, keys):
    """Points for many node *keys* in ONE response, split per node so the client
    builds a separate geometry per node while paying a single HTTP round-trip.

    Format: [uint32 numBlobs] then per blob:
            [uint32 keyLen][key utf8][uint32 payloadLen][payload]
    where payload is the same binary as copc_nodes_binary for that one node."""
    import struct
    entry = open_copc(path)
    octree = _get_octree(entry)
    want = list(dict.fromkeys(keys))

    # Fetch each node separately: the batched read reorders points by disk
    # offset, losing the per-node boundary. With the octree cached, each fetch is
    # just a chunk decompress (the expensive hierarchy walk is gone).
    parts = []
    count = 0
    with _borrow_reader(entry) as reader:  # pooled handle: no shared-seek race
        center = reader.copc_info.center
        for k in want:
            nd = octree.get(k)
            if nd is None:
                continue
            pts = reader._fetch_and_decompress_points_of_nodes([nd])
            payload = _pack_node_points(pts, entry, center)
            kb = k.encode('utf-8')
            parts.append(struct.pack('<I', len(kb)))
            parts.append(kb)
            parts.append(struct.pack('<I', len(payload)))
            parts.append(payload)
            count += 1
    return struct.pack('<I', count) + b''.join(parts)


def ensure_copc(src_path, dst_path=None, progress=None):
    """Convert a non-COPC cloud to .copc.laz, returning the output path.

    Tool ladder: untwine → PDAL>=2.4 (writers.copc) → copclib builder → raise.
    Callers should fall back to the legacy whole-cloud path if this raises.
    ``progress(done, total[, phase])`` is called during the untwine and copclib
    builds (the PDAL path is a single CLI call with no progress)."""
    if is_copc(src_path):
        return src_path
    if dst_path is None:
        base = os.path.splitext(src_path)[0]
        dst_path = base + '.copc.laz'

    # Reuse a previous conversion if it's present and at least as new as the source.
    if (os.path.exists(dst_path)
            and os.path.getmtime(dst_path) >= os.path.getmtime(src_path)
            and is_copc(dst_path)):
        if progress is not None:
            progress(1, 1)
        return dst_path

    # Build into a temp file, then atomically rename. Writing dst directly means
    # a crash mid-convert leaves a partial file that can pass the mtime reuse
    # check above, and two concurrent conversions would corrupt each other.
    # (.copc.laz suffix kept so PDAL's extension-based writer inference still
    # picks writers.copc for the temp output.)
    # pid alone collides when two threads of the same process convert the same
    # source concurrently — add a random tail so each build gets its own temp.
    tmp_path = f'{dst_path}.{os.getpid()}.{uuid.uuid4().hex[:8]}.tmp.copc.laz'
    threads = int(os.environ.get('WPC_COPC_THREADS', os.cpu_count() or 1))
    try:
        converted = False
        # 1) untwine: out-of-core (spills to a temp dir on disk), so RAM stays
        #    bounded — the only tool here that survives clouds larger than RAM.
        if shutil.which('untwine'):
            _run_untwine(src_path, tmp_path, threads, progress)
            converted = True
        # 2) PDAL CLI with writers.copc (PDAL >= 2.4) — holds the whole cloud in RAM
        if not converted and _pdal_has_copc_writer():
            import subprocess
            # forward=all keeps the source header's scale/offset: writers.copc
            # otherwise re-quantizes to its 0.01 default and merges mm-spaced points.
            base_cmd = ['pdal', 'translate', src_path, tmp_path, '-w', 'writers.copc',
                        '--writers.copc.forward=all']
            # writers.copc runs near single-threaded by default; threads=ncpu
            # is ~2.2x faster (7.8M pts: 13.8s -> 5.9s) for ~10% more RAM.
            attempts = [base_cmd]
            if threads > 1:
                attempts.insert(0, base_cmd + [f'--writers.copc.threads={threads}'])
            for cmd in attempts:
                try:
                    subprocess.run(cmd, check=True, capture_output=True)
                    converted = True
                    break
                except subprocess.CalledProcessError as e:
                    if e.returncode < 0:
                        # Killed by a signal — almost always the OOM killer.
                        # copclib needs even more RAM, and running it in-process
                        # would drag the whole server into swap: stop here.
                        raise RuntimeError(
                            f'PDAL COPC conversion was killed (signal {-e.returncode}, '
                            'likely out of memory); install untwine for large clouds'
                        ) from e
                    # pdal accepted the drivers query but rejected the translate
                    # (arg parsing, version quirks) — retry without threads,
                    # then fall through to copclib.
        if not converted:
            # 3) copclib octree builder (self-contained, bulk Unpack)
            _check_fits_in_memory(src_path)
            try:
                from tools.las_to_copc import las_to_copc
            except Exception:
                try:
                    import sys
                    sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'tools'))
                    from las_to_copc import las_to_copc
                except Exception as e:
                    raise RuntimeError(
                        'COPC conversion unavailable: install untwine, PDAL>=2.4 '
                        '(writers.copc) or copclib'
                    ) from e
            las_to_copc(src_path, tmp_path, progress=progress)
        os.replace(tmp_path, dst_path)
    finally:
        if os.path.exists(tmp_path):
            try:
                os.remove(tmp_path)
            except OSError:
                pass
    return dst_path


# untwine --progress_fd messages (native-endian, see untwine/generic/progress.hpp):
#   1000 progress: int32 id, uint32 percent, uint32 len, bytes message
#   1001 error:    int32 id, uint32 len, bytes message
# Its 0-40% span is the distribute pass over the source; it reports that span
# only on the fd (--progress_debug stays silent until 40%).
_UNTWINE_MSG_PROGRESS = 1000
_UNTWINE_MSG_ERROR = 1001
_UNTWINE_DISTRIBUTE_PCT = 40


def _read_exact(fp, n):
    buf = fp.read(n)
    return buf if len(buf) == n else None


def _read_untwine_messages(fp, on_progress, on_error):
    """Parse untwine's progress-fd stream until EOF (or a garbled header)."""
    import struct
    while True:
        head = _read_exact(fp, 4)
        if head is None:
            return
        (msg_id,) = struct.unpack('=i', head)
        if msg_id == _UNTWINE_MSG_PROGRESS:
            rest = _read_exact(fp, 8)
            if rest is None:
                return
            percent, size = struct.unpack('=II', rest)
        elif msg_id == _UNTWINE_MSG_ERROR:
            rest = _read_exact(fp, 4)
            if rest is None:
                return
            percent, (size,) = None, struct.unpack('=I', rest)
        else:
            return
        text = _read_exact(fp, size)
        if text is None:
            return
        text = text.decode('utf-8', 'replace')
        if percent is None:
            on_error(text)
        else:
            on_progress(percent, text)


def _run_untwine(src_path, dst_path, threads, progress=None):
    """Run the untwine CLI, forwarding its percent to *progress*. Its temp dir
    sits beside the output (same disk) and needs about the source's size free."""
    import subprocess
    tmp_dir = dst_path + '.d'
    read_fd, write_fd = os.pipe()
    cmd = ['untwine', '-i', src_path, '-o', dst_path, '--temp_dir', tmp_dir,
           '--threads', str(max(threads, 1)), '--progress_fd', str(write_fd)]
    errors = []

    def on_progress(percent, _text):
        if progress is not None:
            phase = 'reading' if percent < _UNTWINE_DISTRIBUTE_PCT else 'building'
            progress(min(percent, 100), 100, phase)

    try:
        try:
            proc = subprocess.Popen(cmd, pass_fds=(write_fd,), stdout=subprocess.DEVNULL,
                                    stderr=subprocess.PIPE)
        finally:
            # Parent's copy closed so the reader sees EOF once untwine exits.
            os.close(write_fd)
        # stderr drained on a side thread so neither pipe can fill and stall untwine.
        stderr_tail = []

        def drain_stderr():
            for line in proc.stderr:
                stderr_tail[:] = (stderr_tail + [line.decode('utf-8', 'replace').rstrip()])[-20:]
        t = threading.Thread(target=drain_stderr, daemon=True)
        t.start()
        with os.fdopen(read_fd, 'rb') as fp:
            read_fd = None
            _read_untwine_messages(fp, on_progress, errors.append)
        proc.wait()
        t.join()
        if proc.returncode != 0:
            detail = ' | '.join(l for l in errors + stderr_tail if l)
            raise RuntimeError(f'untwine failed (exit {proc.returncode}): {detail}')
    finally:
        if read_fd is not None:
            os.close(read_fd)
        shutil.rmtree(tmp_dir, ignore_errors=True)


def _check_fits_in_memory(src_path):
    """Refuse the in-process copclib build when it would not fit in RAM: it
    peaks at ~3x the source size, and swapping it would stall the server."""
    try:
        ram = os.sysconf('SC_PAGE_SIZE') * os.sysconf('SC_PHYS_PAGES')
    except (ValueError, OSError, AttributeError):
        return
    need = os.path.getsize(src_path) * 3
    if need > ram:
        raise RuntimeError(
            f'COPC conversion needs ~{need / 2**30:.0f} GB RAM with copclib '
            f'({ram / 2**30:.0f} GB available); install untwine for large clouds')


def _pdal_has_copc_writer():
    try:
        import subprocess
        out = subprocess.run(['pdal', '--drivers'], capture_output=True, text=True)
        return 'writers.copc' in out.stdout
    except Exception:
        return False