Spaces:
Running
Running
File size: 2,217 Bytes
5708b1b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 | """Cache en memoria del DataFrame del preprocessed.
``pd.read_csv`` de un archivo de 2.47 GB tarda ~60 s en CPU básico. Si lo
guardamos en RAM, la primera corrida sigue tardando lo mismo, pero las que
vienen después son instantáneas (parte CPU del pipeline).
La key del cache incluye ``mtime`` para que, si el archivo se actualiza en
disco (sync semanal del preprocessed), la entrada se invalide sola.
Single-container HF Space, sin multiproceso → un dict global con lock alcanza.
"""
from __future__ import annotations
import threading
from collections import OrderedDict
from pathlib import Path
import pandas as pd
_CACHE: "OrderedDict[tuple[str, float, bool], pd.DataFrame]" = OrderedDict()
_LOCK = threading.Lock()
_MAX_ENTRIES = 2 # preprocessed base + (eventual) labeled distinto
def get_cached(path: Path, *, low_memory: bool = False) -> pd.DataFrame:
"""Devuelve un DataFrame del CSV, leyendo del cache si está disponible.
Importante: el DataFrame es **compartido**. Los callers que vayan a mutar
columnas deben hacer ``.copy()`` antes. Mutaciones tipo ``df[col] = ...``
sobre el devuelto afectan al cache.
"""
p = Path(path).resolve()
key = (str(p), p.stat().st_mtime, low_memory)
with _LOCK:
hit = _CACHE.get(key)
if hit is not None:
_CACHE.move_to_end(key)
return hit
# Leer fuera del lock para no bloquear otros lookups durante 60 s.
df = pd.read_csv(p, low_memory=low_memory)
with _LOCK:
# Si otro thread cargó la misma key en el medio, preferimos su copia.
existing = _CACHE.get(key)
if existing is not None:
return existing
_CACHE[key] = df
_CACHE.move_to_end(key)
while len(_CACHE) > _MAX_ENTRIES:
_CACHE.popitem(last=False)
return df
def is_cached(path: Path, *, low_memory: bool = False) -> bool:
"""Si el archivo ya está cargado en RAM (sin tocar disco más allá del stat)."""
p = Path(path).resolve()
if not p.exists():
return False
key = (str(p), p.stat().st_mtime, low_memory)
with _LOCK:
return key in _CACHE
def clear() -> None:
with _LOCK:
_CACHE.clear()
|