From 4f7f0128959156a5dfcadd1bae981d30fe1acbb3 Mon Sep 17 00:00:00 2001 From: AxelNoun <150222552+AxelNoun@users.noreply.github.com> Date: Wed, 26 Aug 2026 06:51:35 +0200 Subject: [PATCH] fix(cache): make task fingerprints deterministic and collision-free set_task derived its cache key from json.dumps(vars(task), default=str). The default=str fallback silently stringifies anything not JSON-serialisable, which is neither injective (str() truncates numpy arrays past 1000 elements) nor stable across processes (set ordering depends on PYTHONHASHSEED, and default reprs embed memory addresses). Two different task configurations could therefore share a cache directory, and identical configurations could fail to reuse one. Replace it with an explicit canonicaliser in pyhealth/tasks/fingerprint.py that raises on anything it cannot represent deterministically, records effective __init__ arguments with defaults applied, covers class-level configuration, and adds an explicit BaseTask.version for logic changes. Also fixes the same bug in the processors cache key, and slugifies task_name before using it as a path component. Cache keys change, so existing task caches under /tasks are orphaned. This is intentional: some of them are wrong. Refs #916 Co-authored-by: Cursor --- docs/api/tasks.rst | 44 ++ docs/api/tasks/pyhealth.tasks.fingerprint.rst | 11 + examples/task_fingerprint.py | 24 + pyhealth/datasets/base_dataset.py | 88 ++- pyhealth/tasks/base_task.py | 35 ++ pyhealth/tasks/fingerprint.py | 586 ++++++++++++++++++ tests/core/test_caching.py | 62 +- tests/core/test_task_fingerprint.py | 373 +++++++++++ 8 files changed, 1134 insertions(+), 89 deletions(-) create mode 100644 docs/api/tasks/pyhealth.tasks.fingerprint.rst create mode 100644 examples/task_fingerprint.py create mode 100644 pyhealth/tasks/fingerprint.py create mode 100644 tests/core/test_task_fingerprint.py diff --git a/docs/api/tasks.rst b/docs/api/tasks.rst index bdaa9599a..01f71684d 100644 --- a/docs/api/tasks.rst +++ b/docs/api/tasks.rst @@ -184,6 +184,49 @@ PyHealth detects the existing cache and skips reprocessing. During development it is useful to set ``dev=True`` on the dataset, which limits processing to 1 000 patients so iterations are fast. +What counts as the same configuration? +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Each task configuration gets its own cache directory, named after a +fingerprint of everything that determines the generated samples: + +- the ``__init__`` arguments of your task, with defaults applied -- + ``MyTask()`` and ``MyTask(window=timedelta(days=15))`` share a cache when + 15 days is the default; +- class-level attributes, the input schema and the output schema; +- ``BaseTask.version`` (see below). + +Arguments that do not affect the output are ignored, so changing +``num_workers`` reuses the cache. Add your own via ``fingerprint_exclude``:: + + class MyTask(BaseTask): + fingerprint_exclude = frozenset({"debug_dir"}) + +The fingerprint is a digest, so the directory name alone does not tell you +which parameter changed. Each cache directory therefore contains a +``task_meta.json`` recording the full configuration in readable form, and +``set_task()`` logs its path at INFO level. + +.. warning:: + + **Editing** ``__call__`` **does not invalidate the cache.** The fingerprint + covers configuration, not code, so changing your labelling logic without + changing any argument will silently reuse the samples built by the previous + version. Bump ``version`` on the task when its logic changes:: + + class MyTask(BaseTask): + version = "2" # was "1"; excluded ICU stays under 24h + + Alternatively, set ``PYHEALTH_FINGERPRINT_SOURCE=1`` to fold a structural + hash of ``__call__`` and ``pre_filter`` into the fingerprint. This is off by + default because it invalidates the cache on cosmetic edits too. + +If a task argument cannot be fingerprinted deterministically -- an object +whose ``repr()`` embeds a memory address, for instance -- ``set_task()`` +raises ``UnfingerprintableError`` rather than risk a colliding or unstable +cache key. Give the class a ``__pyhealth_fingerprint__()`` method returning a +plain description of its configuration, or exclude the argument. + .. note:: **A note on multiprocessing.** ``set_task()`` can spawn worker processes @@ -206,6 +249,7 @@ Available Tasks :maxdepth: 3 Base Task + Task Fingerprint In-Hospital Mortality (MIMIC-IV) In-Hospital Mortality (MEDS) MIMIC-III ICD-9 Coding diff --git a/docs/api/tasks/pyhealth.tasks.fingerprint.rst b/docs/api/tasks/pyhealth.tasks.fingerprint.rst new file mode 100644 index 000000000..b4ad22e32 --- /dev/null +++ b/docs/api/tasks/pyhealth.tasks.fingerprint.rst @@ -0,0 +1,11 @@ +pyhealth.tasks.fingerprint +======================================= + +Deterministic cache keys for ``set_task``. Replaces the previous +``json.dumps(..., default=str)`` hash, which was neither stable across +processes nor injective for large arrays. + +.. automodule:: pyhealth.tasks.fingerprint + :members: UnfingerprintableError, task_spec, task_fingerprint, task_cache_name, processors_fingerprint, write_task_metadata, slugify, record_init_args + :undoc-members: + :show-inheritance: diff --git a/examples/task_fingerprint.py b/examples/task_fingerprint.py new file mode 100644 index 000000000..354be3b6b --- /dev/null +++ b/examples/task_fingerprint.py @@ -0,0 +1,24 @@ +"""Show that task cache keys include init args (issue #916). + +Two ReadmissionPredictionMIMIC3 configurations that differ only by +``window`` must produce different cache directory names. A slash in +``task_name`` (as used by BenchmarkEHRShot) is slugified to a single +path component. +""" + +from datetime import timedelta + +from pyhealth.tasks.benchmark_ehrshot import BenchmarkEHRShot +from pyhealth.tasks.fingerprint import task_cache_name, task_fingerprint +from pyhealth.tasks.readmission_prediction import ReadmissionPredictionMIMIC3 + +if __name__ == "__main__": + t15 = ReadmissionPredictionMIMIC3() + t30 = ReadmissionPredictionMIMIC3(window=timedelta(days=30)) + print("15-day window:", task_cache_name(t15)) + print("30-day window:", task_cache_name(t30)) + assert task_fingerprint(t15) != task_fingerprint(t30) + + ehrshot = task_cache_name(BenchmarkEHRShot(task="guo_los")) + print("EHRShot guo_los:", ehrshot) + assert "/" not in ehrshot diff --git a/pyhealth/datasets/base_dataset.py b/pyhealth/datasets/base_dataset.py index 3d449d579..855879730 100644 --- a/pyhealth/datasets/base_dataset.py +++ b/pyhealth/datasets/base_dataset.py @@ -43,6 +43,11 @@ from ..processors.base_processor import FeatureProcessor from .configs import load_yaml_config from .sample_dataset import SampleDataset, SampleBuilder +from ..tasks.fingerprint import ( + processors_fingerprint, + task_cache_name, + write_task_metadata, +) from ..utils import set_env # Set logging level for distributed to ERROR to reduce verbosity @@ -388,9 +393,10 @@ def _init_cache_dir(self, cache_dir: str | Path | None) -> Path: tmp/ # Temporary files during processing global_event_df.parquet/ # Cached global event dataframe tasks/ # Cached task-specific data - {task_name}_{task_uuid}/ # Cached data for specific task based on task name, schema, and args + {task_name}_{task_fingerprint}/ # One task configuration (see set_task) + task_meta.json # Readable spec behind the fingerprint task_df.ld/ # Intermediate task dataframe based on schema - samples_{proc_uuid}.ld/ # Final processed samples after applying processors + samples_{proc_fingerprint}.ld/ # Final processed samples after applying processors Returns: Path: The resolved cache directory path. @@ -1005,11 +1011,28 @@ def set_task( """Processes the base dataset to generate the task-specific sample dataset. The cache structure is as follows:: - {task_name}_{task_uuid}/ # Cached data for specific task based on task name, schema, and args - task_df.ld/ # Intermediate task dataframe based on schema - samples_{proc_uuid}.ld/ # Final processed samples after applying processors - schema.pkl # Saved SampleBuilder schema - *.bin # Processed sample files + {task_name}_{task_fingerprint}/ # One task configuration + task_meta.json # Readable spec behind the fingerprint + build.lock # Guards concurrent builds + task_df.ld/ # Intermediate task dataframe based on schema + samples_{proc_fingerprint}.ld/ # Final processed samples after applying processors + schema.pkl # Saved SampleBuilder schema + *.bin # Processed sample files + + ``task_fingerprint`` is a digest of everything that determines the + generated samples: the ``__init__`` arguments of the task (with defaults + applied, so passing a default explicitly is not a different config), + class-level attributes, the input and output schemas, and + ``BaseTask.version``. Changing any of them yields a new directory, so + two configurations can never share a cache. Arguments that do not affect + the output -- ``num_workers``, ``verbose`` -- are excluded, and a task + can exclude more of its own via ``fingerprint_exclude``. + + Because the digest is opaque, ``task_meta.json`` records the full spec + in readable form: consult it to see which parameter produced a given + directory. Note that editing ``__call__`` does *not* change the + fingerprint by itself -- bump ``version`` on the task when its logic + changes, or existing caches will be silently reused. Args: task (Optional[BaseTask]): The task to set. Uses default task if None. @@ -1040,52 +1063,25 @@ def set_task( f"Setting task {task.task_name} for {self.dataset_name} base dataset..." ) - task_params = json.dumps( - { - **vars(task), - "input_schema": task.input_schema, - "output_schema": task.output_schema, - }, - sort_keys=True, - default=str, - ) - - cache_dir = ( - self.cache_dir - / "tasks" - / f"{task.task_name}_{uuid.uuid5(uuid.NAMESPACE_DNS, task_params)}" - ) + cache_dir = self.cache_dir / "tasks" / task_cache_name(task) cache_dir.mkdir(parents=True, exist_ok=True) - proc_params = json.dumps( - { - "input_processors": ( - { - f"{k}_{v.__class__.__name__}": vars(v) - for k, v in input_processors.items() - } - if input_processors - else None - ), - "output_processors": ( - { - f"{k}_{v.__class__.__name__}": vars(v) - for k, v in output_processors.items() - } - if output_processors - else None - ), - }, - sort_keys=True, - default=str, - ) - task_df_path = Path(cache_dir) / "task_df.ld" samples_path = ( Path(cache_dir) - / f"samples_{uuid.uuid5(uuid.NAMESPACE_DNS, proc_params)}.ld" + / f"samples_{processors_fingerprint(input_processors, output_processors)[:16]}.ld" ) + # The cache key is opaque by construction; the sidecar makes it + # auditable. Written before the build so a crashed build still leaves + # a directory that explains what it was trying to produce. + meta_path = write_task_metadata(cache_dir, task) + logger.info( + "Task cache key %s -- derived from init args, schemas and task " + "version. Full spec: %s", + task_cache_name(task), + meta_path, + ) logger.info(f"Task cache paths: task_df={task_df_path}, samples={samples_path}") task_df_path.mkdir(parents=True, exist_ok=True) diff --git a/pyhealth/tasks/base_task.py b/pyhealth/tasks/base_task.py index 395686ed7..d29e5cac7 100644 --- a/pyhealth/tasks/base_task.py +++ b/pyhealth/tasks/base_task.py @@ -3,12 +3,47 @@ import polars as pl +from .fingerprint import record_init_args + class BaseTask(ABC): + """Base class for PyHealth predictive tasks. + + Init arguments, class-level configuration, and ``version`` are part of + the ``set_task`` cache key. Bump ``version`` when ``__call__`` or + ``pre_filter`` changes in a way that alters generated samples. + + Example: + >>> from pyhealth.tasks.base_task import BaseTask + >>> class ToyTask(BaseTask): + ... task_name = "toy" + ... input_schema = {"x": "sequence"} + ... output_schema = {"y": "binary"} + ... def __call__(self, patient): + ... return [] + >>> ToyTask().task_name + 'toy' + """ + task_name: str input_schema: Dict[str, Union[str, Type]] output_schema: Dict[str, Union[str, Type]] + #: Bump when ``__call__`` or ``pre_filter`` logic changes in a way that + #: alters the generated samples. Init args alone cannot detect this. + version: str = "1" + + #: Attribute names that do not affect the generated samples and must not + #: invalidate the cache (e.g. ``num_workers``, ``verbose``). Denylist, not + #: allowlist: forgetting an entry here costs a spurious rebuild, whereas + #: forgetting to allowlist a semantic arg silently reuses a stale cache. + fingerprint_exclude: frozenset[str] = frozenset() + + def __init_subclass__(cls, **kwargs) -> None: + """Record the effective ``__init__`` arguments of every task instance.""" + super().__init_subclass__(**kwargs) + record_init_args(cls) + def __init__( self, code_mapping: Optional[Dict[str, Tuple[str, str]]] = None, diff --git a/pyhealth/tasks/fingerprint.py b/pyhealth/tasks/fingerprint.py new file mode 100644 index 000000000..b7a60e8c1 --- /dev/null +++ b/pyhealth/tasks/fingerprint.py @@ -0,0 +1,586 @@ +"""Deterministic fingerprinting of tasks for cache keys. + +Proposed replacement for the inline ``json.dumps(vars(task), default=str)`` +fingerprint currently used by ``BaseDataset.set_task``. + +Design goals +------------ +1. **Deterministic** across processes, machines and Python versions: never + relies on ``PYTHONHASHSEED``-dependent iteration order or on memory + addresses leaking through ``repr()``. +2. **Lossless**: no truncating ``str()`` of large objects, so two different + configurations can never collide. +3. **Fail loud, not silent**: an argument that cannot be fingerprinted + deterministically raises with an actionable message instead of producing a + key that is either unstable (permanent cache misses) or colliding (silently + stale samples). +4. **Complete**: covers recorded ``__init__`` arguments, derived instance + state, class-level configuration, and an explicit task ``version``. +5. **Legible**: emits a human-readable spec that is written next to the cache, + so users can see *why* a cache directory exists. + +Public API +---------- +``task_spec(task)`` -> canonical JSON-safe dict describing the task +``task_fingerprint(task)`` -> stable hex digest of that spec +``task_cache_name(task)`` -> path-safe ``{slug}_{digest}`` directory name +``write_task_metadata(...)`` -> writes ``task_meta.json`` sidecar +""" + +from __future__ import annotations + +import ast +import dataclasses +import datetime as _dt +import decimal +import enum +import functools +import hashlib +import inspect +import json +import logging +import os +import re +import textwrap +import types +import uuid +from collections.abc import Mapping +from pathlib import Path, PurePath +from typing import Any + +logger = logging.getLogger(__name__) + +# Bump when the fingerprint format itself changes. Every cache key changes, +# so existing caches become unreachable (they are not deleted). +FINGERPRINT_VERSION = 2 + +# Attribute set on task instances by ``record_init_args``. +_INIT_ARGS_ATTR = "_pyhealth_init_args" + +# Opt-out for environments with exotic task arguments. Strict mode is the +# default because a non-strict fallback reintroduces silent collisions. +_STRICT = os.environ.get("PYHEALTH_FINGERPRINT_STRICT", "1") not in ("0", "false", "False") + +# Include a structural hash of ``__call__``/``pre_filter`` source. Off by +# default: editing a comment should not invalidate a 40-minute cache build. +_HASH_SOURCE = os.environ.get("PYHEALTH_FINGERPRINT_SOURCE", "0") in ("1", "true", "True") + +_ADDRESS_RE = re.compile(r"0x[0-9a-fA-F]{6,}") +_MAX_DEPTH = 64 + + +class UnfingerprintableError(TypeError): + """Raised when a task attribute has no deterministic representation. + + Example: + >>> from pyhealth.tasks.fingerprint import UnfingerprintableError + >>> isinstance(UnfingerprintableError("opaque"), TypeError) + True + """ + + +# -------------------------------------------------------------------------- +# canonicalisation +# -------------------------------------------------------------------------- + + +def _digest(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _hint(obj: Any, name: str) -> str: + return ( + f"Cannot deterministically fingerprint attribute {name!r} of type " + f"{type(obj).__module__}.{type(obj).__qualname__}. Its repr() is not " + f"stable across processes, so it would either break caching or, worse, " + f"silently reuse another configuration's cache.\n" + f"Fix it in one of three ways:\n" + f" 1. give the class a __pyhealth_fingerprint__() method returning a " + f"JSON-safe description of its configuration;\n" + f" 2. list {name!r} in the task's `fingerprint_exclude` if it does not " + f"affect the generated samples;\n" + f" 3. store the plain configuration (str/int/tuple) on the task instead " + f"of the constructed object." + ) + + +def _canon(obj: Any, name: str = "", depth: int = 0, seen: set[int] | None = None) -> Any: + """Return a JSON-safe, type-tagged, order-stable representation of ``obj``.""" + if depth > _MAX_DEPTH: + raise UnfingerprintableError(f"{name}: nesting deeper than {_MAX_DEPTH}") + + seen = seen if seen is not None else set() + + # -- scalars ---------------------------------------------------------- + if obj is None: + return None + if isinstance(obj, bool): # must precede int + return ["bool", obj] + if isinstance(obj, int): + return ["int", str(obj)] # str() keeps arbitrary-precision ints exact + if isinstance(obj, float): + return ["float", obj.hex()] # exact and round-trippable, incl. -0.0/nan + if isinstance(obj, str): + return ["str", obj] + if isinstance(obj, (bytes, bytearray)): + return ["bytes", _digest(bytes(obj))] + if isinstance(obj, complex): + return ["complex", obj.real.hex(), obj.imag.hex()] + if isinstance(obj, decimal.Decimal): + return ["decimal", str(obj)] + + # -- stdlib value types ------------------------------------------------ + if isinstance(obj, enum.Enum): + return ["enum", _qualname(type(obj)), _canon(obj.value, f"{name}.value", depth + 1, seen)] + if isinstance(obj, _dt.datetime): + return ["datetime", obj.isoformat(), str(obj.tzinfo)] + if isinstance(obj, _dt.date): + return ["date", obj.isoformat()] + if isinstance(obj, _dt.time): + return ["time", obj.isoformat()] + if isinstance(obj, _dt.timedelta): + return ["timedelta", obj.days, obj.seconds, obj.microseconds] + if isinstance(obj, PurePath): + return ["path", str(obj)] + if isinstance(obj, uuid.UUID): + return ["uuid", str(obj)] + if isinstance(obj, range): + return ["range", obj.start, obj.stop, obj.step] + + # -- cycles ------------------------------------------------------------ + if id(obj) in seen: + raise UnfingerprintableError(f"{name}: circular reference") + seen = seen | {id(obj)} + + # -- third-party numerics (duck-typed, no hard dependency) ------------- + special = _canon_scientific(obj, name, depth, seen) + if special is not None: + return special + + # -- containers -------------------------------------------------------- + if isinstance(obj, Mapping): + # Fast path for the common all-string-keys case (e.g. a 100k-token + # vocabulary): sort the raw keys instead of serialising each one. + if all(isinstance(k, str) for k in obj): + keys = sorted(obj.keys()) + items = [ + (["str", k], _canon(obj[k], f"{name}[{k!r}]", depth + 1, seen)) for k in keys + ] + return ["dict", items] + items = [ + (_canon(k, f"{name}.", depth + 1, seen), _canon(v, f"{name}[{k!r}]", depth + 1, seen)) + for k, v in obj.items() + ] + # Sort on the serialised key, never on the raw key: sorting raw keys + # crashes on mixed int/str keys (the current json sort_keys=True bug). + items.sort(key=lambda kv: json.dumps(kv[0], sort_keys=True)) + return ["dict", items] + if isinstance(obj, (set, frozenset)): + tag = "frozenset" if isinstance(obj, frozenset) else "set" + if all(isinstance(v, str) for v in obj): + return [tag, [["str", v] for v in sorted(obj)]] + elems = [_canon(v, f"{name}.", depth + 1, seen) for v in obj] + elems.sort(key=lambda e: json.dumps(e, sort_keys=True)) # kills PYTHONHASHSEED dependence + return [tag, elems] + if isinstance(obj, tuple): + return ["tuple", [_canon(v, f"{name}[{i}]", depth + 1, seen) for i, v in enumerate(obj)]] + if isinstance(obj, list): + return ["list", [_canon(v, f"{name}[{i}]", depth + 1, seen) for i, v in enumerate(obj)]] + + # -- callables and types ---------------------------------------------- + if isinstance(obj, functools.partial): + return [ + "partial", + _canon(obj.func, f"{name}.func", depth + 1, seen), + _canon(obj.args, f"{name}.args", depth + 1, seen), + _canon(obj.keywords, f"{name}.keywords", depth + 1, seen), + ] + if isinstance(obj, type): + return ["class", _qualname(obj)] + if isinstance(obj, (types.FunctionType, types.MethodType, types.BuiltinFunctionType)): + return _canon_callable(obj, name) + + # -- user hook --------------------------------------------------------- + hook = getattr(type(obj), "__pyhealth_fingerprint__", None) + if callable(hook): + return ["hook", _qualname(type(obj)), _canon(hook(obj), f"{name}.", depth + 1, seen)] + + if dataclasses.is_dataclass(obj): + fields = {f.name: getattr(obj, f.name) for f in dataclasses.fields(obj)} + return ["dataclass", _qualname(type(obj)), _canon(fields, name, depth + 1, seen)] + + # -- plain objects: recurse into their state --------------------------- + state = _object_state(obj) + if state is not None: + return ["object", _qualname(type(obj)), _canon(state, name, depth + 1, seen)] + + # -- last resort ------------------------------------------------------- + text = repr(obj) + if type(obj).__repr__ is object.__repr__ or _ADDRESS_RE.search(text): + if _STRICT: + raise UnfingerprintableError(_hint(obj, name)) + logger.warning("%s -- falling back to an unstable repr()", _hint(obj, name)) + return ["repr", _qualname(type(obj)), text] + + +def _canon_scientific(obj: Any, name: str, depth: int, seen: set[int]) -> Any: + """Handle numpy / torch / pandas / polars without importing them.""" + mod = type(obj).__module__.split(".")[0] + + if mod == "numpy": + import numpy as np + + if isinstance(obj, np.ndarray): + if obj.dtype == object: # tobytes() would hash pointers + return ["ndarray_object", list(obj.shape), _canon(obj.tolist(), name, depth + 1, seen)] + return ["ndarray", obj.dtype.str, list(obj.shape), _digest(np.ascontiguousarray(obj).tobytes())] + if isinstance(obj, np.dtype): + return ["dtype", str(obj)] + if isinstance(obj, np.generic): + return _canon(obj.item(), name, depth + 1, seen) + + if mod == "torch": + import torch + + if isinstance(obj, torch.Tensor): + arr = obj.detach().cpu().contiguous().numpy() + return ["tensor", str(obj.dtype), list(obj.shape), _digest(arr.tobytes())] + if isinstance(obj, torch.dtype): + return ["torch.dtype", str(obj)] + if isinstance(obj, torch.device): + return ["torch.device", str(obj)] + + if mod == "pandas": + import pandas as pd + + if isinstance(obj, (pd.DataFrame, pd.Series)): + values = pd.util.hash_pandas_object(obj, index=True).values + return ["pandas", list(getattr(obj, "shape", ())), _digest(values.tobytes())] + + if mod == "polars": + import polars as pl + + if isinstance(obj, pl.DataFrame): + return ["polars", list(obj.shape), _digest(obj.hash_rows().to_numpy().tobytes())] + + return None + + +def _canon_callable(fn: Any, name: str) -> Any: + qn = _qualname(fn) + anonymous = "" in qn or "" in qn + src = _structural_source(fn) + if src is not None: + return ["function", qn, src] + if anonymous: + if _STRICT: + raise UnfingerprintableError( + f"{name}: anonymous callable {qn!r} has no retrievable source, so " + f"two different lambdas would share a cache key. Use a module-level " + f"function, or exclude it via `fingerprint_exclude`." + ) + logger.warning("%s: anonymous callable with no source; cache key may collide", name) + return ["function", qn, None] + + +def _structural_source(fn: Any) -> str | None: + """Hash the AST of ``fn``, ignoring comments, whitespace and docstrings.""" + try: + src = textwrap.dedent(inspect.getsource(fn)) + tree = ast.parse(src) + except (OSError, TypeError, SyntaxError, IndentationError): + return None + for node in ast.walk(tree): + body = getattr(node, "body", None) + if isinstance(body, list) and body: + first = body[0] + if isinstance(first, ast.Expr) and isinstance(first.value, ast.Constant) and isinstance( + first.value.value, str + ): + body.pop(0) + return _digest(ast.dump(tree).encode())[:16] + + +def _object_state(obj: Any) -> dict[str, Any] | None: + """Instance state of a plain object, covering ``__dict__`` and ``__slots__``.""" + state: dict[str, Any] = {} + found = False + if hasattr(obj, "__dict__") and isinstance(getattr(obj, "__dict__", None), dict): + state.update(obj.__dict__) + found = True + for klass in type(obj).__mro__: + for slot in getattr(klass, "__slots__", ()) or (): + if isinstance(slot, str) and hasattr(obj, slot): + state[slot] = getattr(obj, slot) + found = True + return state if found else None + + +def _qualname(obj: Any) -> str: + return f"{getattr(obj, '__module__', '?')}.{getattr(obj, '__qualname__', type(obj).__qualname__)}" + + +# -------------------------------------------------------------------------- +# init-arg recording +# -------------------------------------------------------------------------- + + +def record_init_args(cls: type) -> type: + """Record the *effective* ``__init__`` arguments of every instance. + + Wraps ``cls.__init__`` so that bound arguments -- with defaults applied -- + are stored on the instance. Applying defaults means ``Task()`` and + ``Task(window=timedelta(days=15))`` produce the same key when 15 days is + the default, and adding a new keyword argument changes the key for + everyone (which is correct: the behaviour changed). + + Intended to be called from ``BaseTask.__init_subclass__``. + + Example: + >>> from pyhealth.tasks.fingerprint import record_init_args + >>> class T: + ... def __init__(self, n=0): + ... self.n = n + >>> record_init_args(T) is T + True + >>> T()._pyhealth_init_args["n"] + 0 + """ + init = cls.__dict__.get("__init__") + if init is None or getattr(init, "_pyhealth_recorded", False): + return cls + try: + sig = inspect.signature(init) + except (TypeError, ValueError): + return cls + + @functools.wraps(init) + def wrapper(self, *args, **kwargs): + init(self, *args, **kwargs) + try: + bound = sig.bind(self, *args, **kwargs) + bound.apply_defaults() + recorded = dict(bound.arguments) + recorded.pop(next(iter(sig.parameters)), None) # drop 'self' + except TypeError: # pragma: no cover - init would have raised already + return + # Most-derived __init__ returns last, so it wins. + object.__setattr__(self, _INIT_ARGS_ATTR, recorded) + + wrapper._pyhealth_recorded = True # type: ignore[attr-defined] + cls.__init__ = wrapper # type: ignore[assignment] + return cls + + +# -------------------------------------------------------------------------- +# spec / fingerprint +# -------------------------------------------------------------------------- + +# Never part of the identity of a task's *output*. +_ALWAYS_EXCLUDED = frozenset( + { + _INIT_ARGS_ATTR, + "num_workers", + "n_jobs", + "verbose", + "cache_dir", + "refresh_cache", + "progress_bar", + } +) + + +def _class_config(task: Any, excluded: frozenset) -> dict[str, Any]: + """Class-level configuration, which ``vars(task)`` cannot see.""" + config: dict[str, Any] = {} + for klass in reversed(type(task).__mro__): + if klass in (object,): + continue + for key, value in vars(klass).items(): + # Underscore-prefixed class attributes are machinery, not + # configuration (e.g. ABCMeta's ``_abc_impl``, which is a C object + # with an address-bearing repr and would otherwise raise). + if key.startswith("_") or key in excluded: + continue + if callable(value) or isinstance( + value, (property, staticmethod, classmethod, types.MemberDescriptorType) + ): + continue + config[key] = value + return config + + +def task_spec(task: Any, *, include_source: bool | None = None) -> dict[str, Any]: + """Build the canonical, JSON-safe description that identifies ``task``. + + Example: + >>> class T: + ... task_name = "toy" + ... input_schema = {} + ... output_schema = {} + >>> task_spec(T())["task_name"] + 'toy' + """ + include_source = _HASH_SOURCE if include_source is None else include_source + excluded = _ALWAYS_EXCLUDED | frozenset(getattr(task, "fingerprint_exclude", ()) or ()) + + init_args = dict(getattr(task, _INIT_ARGS_ATTR, {}) or {}) + instance_state = { + k: v for k, v in (_object_state(task) or {}).items() if k not in excluded + } + class_config = {k: v for k, v in _class_config(task, excluded).items()} + + spec: dict[str, Any] = { + "fingerprint_version": FINGERPRINT_VERSION, + "task_class": _qualname(type(task)), + "task_name": getattr(task, "task_name", None), + "task_version": getattr(task, "version", "1"), + "init_args": _canon({k: v for k, v in init_args.items() if k not in excluded}, "init_args"), + "class_config": _canon(class_config, "class_config"), + "instance_state": _canon(instance_state, "instance_state"), + "input_schema": _canon(getattr(task, "input_schema", None), "input_schema"), + "output_schema": _canon(getattr(task, "output_schema", None), "output_schema"), + } + if include_source: + spec["source"] = { + hook: _structural_source(getattr(type(task), hook, None)) + for hook in ("__call__", "pre_filter") + } + return spec + + +def task_fingerprint(task: Any, *, include_source: bool | None = None) -> str: + """Stable hex digest identifying the task configuration. + + Example: + >>> class T: + ... task_name = "toy" + ... input_schema = {} + ... output_schema = {} + >>> len(task_fingerprint(T())) + 64 + >>> task_fingerprint(T()) == task_fingerprint(T()) + True + """ + spec = task_spec(task, include_source=include_source) + return _digest(json.dumps(spec, sort_keys=True, separators=(",", ":")).encode()) + + +_SLUG_RE = re.compile(r"[^A-Za-z0-9._-]+") + + +def slugify(name: str, max_len: int = 48) -> str: + """Make a task name safe for a single path component. + + Example: + >>> slugify("BenchmarkEHRShot/guo_los") + 'BenchmarkEHRShot-guo_los' + """ + slug = _SLUG_RE.sub("-", str(name)).strip("-.") or "task" + return slug[:max_len] + + +def task_cache_name(task: Any, *, include_source: bool | None = None) -> str: + """Path-safe cache directory name: ``{slug}_{digest[:16]}``. + + Example: + >>> class T: + ... task_name = "Bench/mark" + ... input_schema = {} + ... output_schema = {} + >>> "/" not in task_cache_name(T()) + True + """ + return f"{slugify(getattr(task, 'task_name', type(task).__name__))}_" \ + f"{task_fingerprint(task, include_source=include_source)[:16]}" + + +def processors_fingerprint( + input_processors: Mapping[str, Any] | None, + output_processors: Mapping[str, Any] | None, +) -> str: + """Fingerprint pre-fitted processors for the ``samples_*`` cache key. + + Same failure modes as tasks: ``SequenceProcessor`` holds a ``CrossMap`` + (address-bearing repr) and a ``code_vocab`` typed ``Dict[Any, int]`` + (mixed keys crash ``sort_keys=True``). + + Example: + >>> processors_fingerprint(None, None) == processors_fingerprint({}, {}) + True + >>> len(processors_fingerprint(None, None)) + 64 + """ + payload = { + "fingerprint_version": FINGERPRINT_VERSION, + "input": _canon( + {f"{k}:{_qualname(type(v))}": v for k, v in (input_processors or {}).items()}, "input" + ), + "output": _canon( + {f"{k}:{_qualname(type(v))}": v for k, v in (output_processors or {}).items()}, "output" + ), + } + return _digest(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()) + + +def write_task_metadata( + cache_dir: Path, + task: Any, + extra: dict[str, Any] | None = None, + overwrite: bool = False, +) -> Path: + """Write a legible sidecar so a cache directory explains itself. + + Answers the second half of issue #916: an opaque digest tells nobody which + parameter changed. ``task_meta.json`` does. + + The sidecar is purely diagnostic, so this function never raises: failing to + write a comment must not abort a multi-hour build. It is also skipped when + the file already exists -- the directory name *is* the fingerprint, so an + existing sidecar necessarily describes the same configuration, and + rewriting it on every cache hit would turn ``created_at`` into a + last-accessed timestamp. + + Example: + >>> import tempfile + >>> class T: + ... task_name = "toy" + ... input_schema = {} + ... output_schema = {} + >>> path = write_task_metadata(Path(tempfile.mkdtemp()), T()) + >>> path.name + 'task_meta.json' + """ + path = Path(cache_dir) / "task_meta.json" + if path.exists() and not overwrite: + return path + + try: + from pyhealth import __version__ as pyhealth_version + except ImportError: # pragma: no cover + pyhealth_version = "unknown" + + payload = { + "fingerprint": task_fingerprint(task), + "created_at": _dt.datetime.now(_dt.UTC).isoformat(), + "pyhealth_version": pyhealth_version, + "spec": task_spec(task), + **(extra or {}), + } + + # Unique temp name per writer. This function runs outside the build lock, + # so parallel hyper-parameter jobs reach it concurrently; a shared temp + # name means they clobber each other's partial writes on POSIX and fail + # outright on Windows, where a second open() of the same path raises + # ERROR_SHARING_VIOLATION (WinError 32). + tmp = path.with_name(f"task_meta.{os.getpid()}.{uuid.uuid4().hex[:8]}.tmp") + try: + tmp.write_text(json.dumps(payload, indent=2, sort_keys=True)) + tmp.replace(path) # atomic: concurrent readers never see a partial file + except OSError as exc: + logger.debug("Could not write task metadata to %s: %s", path, exc) + try: + tmp.unlink(missing_ok=True) + except OSError: # pragma: no cover - best effort cleanup + pass + return path diff --git a/tests/core/test_caching.py b/tests/core/test_caching.py index a346b98e1..594eaed03 100644 --- a/tests/core/test_caching.py +++ b/tests/core/test_caching.py @@ -6,12 +6,12 @@ import dask.dataframe as dd import torch import json -import uuid from tests.base import BaseTestCase from pyhealth.datasets.base_dataset import BaseDataset from pyhealth.tasks.base_task import BaseTask from pyhealth.datasets.sample_dataset import SampleDataset +from pyhealth.tasks.fingerprint import task_cache_name, task_fingerprint class MockTask(BaseTask): @@ -140,15 +140,19 @@ def test_set_task_writes_cache_and_metadata(self): self.assertEqual(sample_dataset.task_name, self.task.task_name) self.assertEqual(len(sample_dataset), 4) - # Ensure intermediate cache files are created in default location - task_params = json.dumps( - {"input_schema": {"test_attribute": "raw"}, "output_schema": {"test_label": "binary"}, "param": 0}, - sort_keys=True, - default=str - ) - task_cache_dir = self.dataset.cache_dir / "tasks" / f"{self.task.task_name}_{uuid.uuid5(uuid.NAMESPACE_DNS, task_params)}" + # Ensure intermediate cache files are created in default location. + # Ask the public helper for the directory name instead of + # re-implementing the hashing expression here: an inlined copy only + # asserts that the implementation equals itself. + task_cache_dir = self.dataset.cache_dir / "tasks" / task_cache_name(self.task) self.assertTrue((task_cache_dir / "task_df.ld" / "index.json").exists()) + # The sidecar must explain the cache directory in plain text. + meta = json.loads((task_cache_dir / "task_meta.json").read_text()) + self.assertEqual(meta["spec"]["task_name"], self.task.task_name) + self.assertEqual(meta["fingerprint"], task_fingerprint(self.task)) + self.assertIn("param", json.dumps(meta["spec"]["init_args"])) + # Cache artifacts should be present for StreamingDataset assert sample_dataset.input_dir.path is not None sample_dir = Path(sample_dataset.input_dir.path) @@ -177,13 +181,7 @@ def test_set_task_writes_cache_and_metadata(self): def test_default_cache_dir_is_used(self): """When cache_dir is omitted, default cache dir should be used.""" - task_params = json.dumps( - {"input_schema": {"test_attribute": "raw"}, "output_schema": {"test_label": "binary"}, "param": 0}, - sort_keys=True, - default=str - ) - - task_cache = self.dataset.cache_dir / "tasks" / f"{self.task.task_name}_{uuid.uuid5(uuid.NAMESPACE_DNS, task_params)}" + task_cache = self.dataset.cache_dir / "tasks" / task_cache_name(self.task) sample_dataset = self.dataset.set_task(self.task) self.assertTrue(task_cache.exists()) @@ -213,20 +211,9 @@ def test_tasks_with_diff_param_values_get_diff_caches(self): self.assertNotEqual(sample_dataset1.path, sample_dataset2.path) - task_params1 = json.dumps( - {"input_schema": {"test_attribute": "raw"}, "output_schema": {"test_label": "binary"}, "param": 1}, - sort_keys=True, - default=str - ) - - task_params2 = json.dumps( - {"input_schema": {"test_attribute": "raw"}, "output_schema": {"test_label": "binary"}, "param": 2}, - sort_keys=True, - default=str - ) - - task_cache1 = self.dataset.cache_dir / "tasks" / f"{self.task.task_name}_{uuid.uuid5(uuid.NAMESPACE_DNS, task_params1)}" - task_cache2 = self.dataset.cache_dir / "tasks" / f"{self.task.task_name}_{uuid.uuid5(uuid.NAMESPACE_DNS, task_params2)}" + task_cache1 = self.dataset.cache_dir / "tasks" / task_cache_name(MockTask(param=1)) + task_cache2 = self.dataset.cache_dir / "tasks" / task_cache_name(MockTask(param=2)) + self.assertNotEqual(task_cache1, task_cache2) self.assertTrue(task_cache1.exists()) self.assertTrue(task_cache2.exists()) @@ -245,20 +232,9 @@ def test_tasks_with_diff_output_schemas_get_diff_caches(self): self.assertNotEqual(sample_dataset1.path, sample_dataset2.path) - task_params1 = json.dumps( - {"input_schema": {"test_attribute": "raw"}, "output_schema": {"test_label": "binary"}, "param": 0}, - sort_keys=True, - default=str - ) - - task_params2 = json.dumps( - {"input_schema": {"test_attribute": "raw"}, "output_schema": {"test_label": "multiclass"}, "param": 0}, - sort_keys=True, - default=str - ) - - task_cache1 = self.dataset.cache_dir / "tasks" / f"{self.task.task_name}_{uuid.uuid5(uuid.NAMESPACE_DNS, task_params1)}" - task_cache2 = self.dataset.cache_dir / "tasks" / f"{self.task.task_name}_{uuid.uuid5(uuid.NAMESPACE_DNS, task_params2)}" + task_cache1 = self.dataset.cache_dir / "tasks" / task_cache_name(MockTask()) + task_cache2 = self.dataset.cache_dir / "tasks" / task_cache_name(MockTask2()) + self.assertNotEqual(task_cache1, task_cache2) self.assertTrue(task_cache1.exists()) self.assertTrue(task_cache2.exists()) diff --git a/tests/core/test_task_fingerprint.py b/tests/core/test_task_fingerprint.py new file mode 100644 index 000000000..726aa8ab3 --- /dev/null +++ b/tests/core/test_task_fingerprint.py @@ -0,0 +1,373 @@ +"""Property tests for deterministic task fingerprinting (issue #916). + +Each test maps to a numbered finding in the audit. They assert *properties* -- +equal configs produce equal keys, different configs produce different keys -- +rather than recomputing the hashing expression, so the implementation can be +changed without rewriting the tests. +""" + +import importlib.util +import json +import os +import subprocess +import sys +import tempfile +import unittest +from datetime import timedelta +from pathlib import Path + +import numpy as np + +from pyhealth.tasks.base_task import BaseTask +from pyhealth.tasks.fingerprint import ( + FINGERPRINT_VERSION, + UnfingerprintableError, + processors_fingerprint, + slugify, + task_cache_name, + task_fingerprint, + task_spec, + write_task_metadata, +) + +_FINGERPRINT_PATH = Path(importlib.util.find_spec("pyhealth.tasks.fingerprint").origin) + + +class _Task(BaseTask): + """Minimal concrete task; subclasses override what each test needs.""" + + task_name = "unit_test_task" + input_schema = {"x": "sequence"} + output_schema = {"y": "binary"} + + def __call__(self, patient): + return [] + + +class Readmission(_Task): + task_name = "ReadmissionPredictionMIMIC3" + + def __init__(self, window=timedelta(days=15), exclude_minors=True, num_workers=4): + super().__init__() + self.window = window + self.exclude_minors = exclude_minors + self.num_workers = num_workers + + +class TestDeterminismAcrossProcesses(unittest.TestCase): + """Findings 1 & 2: keys must not depend on PYTHONHASHSEED. + + The module is loaded standalone by file path so each subprocess stays fast + (importing the pyhealth package would pull in torch). + """ + + def _fingerprint_with_seed(self, seed, body): + snippet = ( + "import importlib.util\n" + f"s = importlib.util.spec_from_file_location('fp', r'{_FINGERPRINT_PATH}')\n" + "fp = importlib.util.module_from_spec(s); s.loader.exec_module(fp)\n" + f"{body}\n" + "print(fp.task_fingerprint(T()))" + ) + result = subprocess.run( + [sys.executable, "-c", snippet], + capture_output=True, + text=True, + env={**os.environ, "PYTHONHASHSEED": str(seed)}, + ) + self.assertEqual(result.returncode, 0, result.stderr) + return result.stdout.strip() + + def test_set_valued_argument_is_seed_independent(self): + body = ( + "class T:\n" + " task_name='T'; input_schema={}; output_schema={}\n" + " def __init__(self): self.codes={'A','B','C','D','E','F'}" + ) + keys = {self._fingerprint_with_seed(s, body) for s in (0, 1, 7)} + self.assertEqual(len(keys), 1, f"set iteration order leaked into the key: {keys}") + + def test_object_argument_is_seed_independent(self): + body = ( + "class Cfg:\n" + " def __init__(self): self.a=1; self.b={'x','y'}\n" + "class T:\n" + " task_name='T'; input_schema={}; output_schema={}\n" + " def __init__(self): self.cfg=Cfg()" + ) + keys = {self._fingerprint_with_seed(s, body) for s in (0, 1, 7)} + self.assertEqual(len(keys), 1, f"object identity leaked into the key: {keys}") + + +class TestNoCollisions(unittest.TestCase): + def test_large_arrays_do_not_collide(self): + """Finding 3: str() truncates arrays past 1000 elements; digests do not.""" + a, b = np.arange(2000), np.arange(2000) + b[500] = -999 + + class T(_Task): + def __init__(self, bins): + super().__init__() + self.bins = bins + + self.assertNotEqual(task_fingerprint(T(a)), task_fingerprint(T(b))) + self.assertEqual(task_fingerprint(T(a)), task_fingerprint(T(np.arange(2000)))) + + def test_class_level_config_is_captured(self): + """Finding 5: vars() cannot see class attributes.""" + + class T30(_Task): + task_name = "Shared" + horizon_days = 30 + + class T365(_Task): + task_name = "Shared" + horizon_days = 365 + + self.assertNotEqual(task_fingerprint(T30()), task_fingerprint(T365())) + + def test_types_are_not_conflated(self): + class T(_Task): + def __init__(self, v): + super().__init__() + self.v = v + + self.assertNotEqual(task_fingerprint(T(1)), task_fingerprint(T("1"))) + self.assertNotEqual(task_fingerprint(T([1, 2])), task_fingerprint(T((1, 2)))) + self.assertNotEqual(task_fingerprint(T(0.0)), task_fingerprint(T(-0.0))) + self.assertNotEqual(task_fingerprint(T(1)), task_fingerprint(T(True))) + + def test_version_bump_changes_key(self): + """Finding 7: changing __call__ logic must be expressible.""" + + class T1(_Task): + version = "1" + + class T2(_Task): + version = "2" + + self.assertNotEqual(task_fingerprint(T1()), task_fingerprint(T2())) + + def test_code_mapping_changes_key(self): + """BaseTask.__init__ rewrites input_schema; that must be reflected.""" + plain = task_fingerprint(_Task()) + mapped = task_fingerprint(_Task(code_mapping={"x": ("ICD9CM", "CCSCM")})) + other = task_fingerprint(_Task(code_mapping={"x": ("NDC", "ATC")})) + self.assertEqual(len({plain, mapped, other}), 3) + + +class TestNoCrashes(unittest.TestCase): + def test_mixed_key_dict_does_not_crash(self): + """Finding 4: json.dumps(sort_keys=True) raises on mixed key types.""" + + class T(_Task): + def __init__(self, mapping): + super().__init__() + self.mapping = mapping + + self.assertEqual(len(task_fingerprint(T({1: "a", "b": 2, (3, 4): None}))), 64) + + def test_slots_task_does_not_crash(self): + """Finding 6: vars() requires __dict__.""" + + class T(BaseTask): + __slots__ = ("w",) + task_name = "slotted" + input_schema = {} + output_schema = {} + + def __init__(self): + self.w = 7 + + def __call__(self, patient): + return [] + + self.assertEqual(len(task_fingerprint(T())), 64) + + +class TestFailLoud(unittest.TestCase): + def test_unfingerprintable_argument_raises_with_guidance(self): + class Opaque: + __slots__ = () + + class T(_Task): + def __init__(self): + super().__init__() + self.thing = Opaque() + + with self.assertRaises(UnfingerprintableError) as ctx: + task_fingerprint(T()) + message = str(ctx.exception) + self.assertIn("__pyhealth_fingerprint__", message) + self.assertIn("fingerprint_exclude", message) + + def test_user_hook_is_honoured(self): + class Mapper: + __slots__ = ("src", "tgt") + + def __init__(self, src, tgt): + self.src, self.tgt = src, tgt + + def __pyhealth_fingerprint__(self): + return {"src": self.src, "tgt": self.tgt} + + class T(_Task): + def __init__(self, mapper): + super().__init__() + self.mapper = mapper + + self.assertEqual( + task_fingerprint(T(Mapper("ICD9CM", "CCSCM"))), + task_fingerprint(T(Mapper("ICD9CM", "CCSCM"))), + ) + self.assertNotEqual( + task_fingerprint(T(Mapper("ICD9CM", "CCSCM"))), + task_fingerprint(T(Mapper("NDC", "ATC"))), + ) + + +class TestInitArgs(unittest.TestCase): + def test_explicit_default_equals_omitted_default(self): + self.assertEqual( + task_fingerprint(Readmission()), + task_fingerprint(Readmission(window=timedelta(days=15))), + ) + + def test_changed_argument_changes_key(self): + self.assertNotEqual( + task_fingerprint(Readmission()), + task_fingerprint(Readmission(window=timedelta(days=30))), + ) + + def test_non_semantic_argument_does_not_change_key(self): + self.assertEqual( + task_fingerprint(Readmission(num_workers=1)), + task_fingerprint(Readmission(num_workers=16)), + ) + + def test_init_args_are_recorded_with_defaults_applied(self): + recorded = getattr(Readmission(exclude_minors=False), "_pyhealth_init_args") + self.assertIs(recorded["exclude_minors"], False) + self.assertEqual(recorded["window"], timedelta(days=15)) + + def test_fingerprint_exclude_is_honoured(self): + class T(_Task): + fingerprint_exclude = frozenset({"scratch"}) + + def __init__(self, scratch, real): + super().__init__() + self.scratch = scratch + self.real = real + + self.assertEqual(task_fingerprint(T("a", 1)), task_fingerprint(T("b", 1))) + self.assertNotEqual(task_fingerprint(T("a", 1)), task_fingerprint(T("a", 2))) + + +class TestProcessors(unittest.TestCase): + def test_processor_with_address_bearing_attribute(self): + """Finding 8: SequenceProcessor._mapper is a CrossMap with a raw repr.""" + + class CrossMapLike: + def __init__(self, src, tgt): + self.src, self.tgt = src, tgt + + class Proc: + def __init__(self, mapping): + self.code_vocab = {"": 0, "": 1} + self._mapper = CrossMapLike(*mapping) + + a = processors_fingerprint({"conditions": Proc(("ICD9CM", "CCSCM"))}, None) + b = processors_fingerprint({"conditions": Proc(("ICD9CM", "CCSCM"))}, None) + c = processors_fingerprint({"conditions": Proc(("NDC", "ATC"))}, None) + self.assertEqual(a, b, "identical processors must share a cache key") + self.assertNotEqual(a, c) + + def test_mixed_key_vocabulary_does_not_crash(self): + """code_vocab is typed Dict[Any, int]; mixed keys crash sort_keys=True.""" + + class Proc: + def __init__(self): + self.code_vocab = {"": 0, 1: 1, ("a", "b"): 2} + + self.assertEqual(len(processors_fingerprint({"x": Proc()}, None)), 64) + + def test_none_processors_are_stable(self): + self.assertEqual(processors_fingerprint(None, None), processors_fingerprint({}, {})) + + +class TestPathSafety(unittest.TestCase): + def test_slash_in_task_name_is_neutralised(self): + """Finding 9: BenchmarkEHRShot sets task_name = 'BenchmarkEHRShot/{task}'.""" + self.assertEqual(slugify("BenchmarkEHRShot/guo_los"), "BenchmarkEHRShot-guo_los") + self.assertNotIn("/", slugify("a/b/../c")) + + def test_cache_name_is_a_single_path_component(self): + class T(_Task): + task_name = "Bench/mark: v2" + + self.assertEqual(len(Path(task_cache_name(T())).parts), 1) + + +class TestMetadataSidecar(unittest.TestCase): + def test_sidecar_is_written_atomically_and_is_readable(self): + with tempfile.TemporaryDirectory() as tmp: + path = write_task_metadata(Path(tmp), Readmission(window=timedelta(days=30))) + payload = json.loads(Path(path).read_text()) + self.assertEqual(payload["spec"]["fingerprint_version"], FINGERPRINT_VERSION) + self.assertEqual(payload["spec"]["task_name"], "ReadmissionPredictionMIMIC3") + self.assertIn("30", json.dumps(payload["spec"]["init_args"])) + self.assertFalse(list(Path(tmp).glob("*.tmp")), "temp file left behind") + + def test_spec_is_json_serialisable(self): + json.dumps(task_spec(Readmission())) + + def test_second_call_does_not_refresh_created_at(self): + """created_at must mean created, not last accessed.""" + with tempfile.TemporaryDirectory() as tmp: + first = json.loads(write_task_metadata(Path(tmp), Readmission()).read_text()) + second = json.loads(write_task_metadata(Path(tmp), Readmission()).read_text()) + self.assertEqual(first["created_at"], second["created_at"]) + + def test_concurrent_writers_leave_no_temp_files(self): + """Runs outside the build lock, so parallel jobs reach it at once.""" + from concurrent.futures import ThreadPoolExecutor + + with tempfile.TemporaryDirectory() as tmp: + with ThreadPoolExecutor(max_workers=8) as pool: + list(pool.map( + lambda _: write_task_metadata(Path(tmp), Readmission(), overwrite=True), + range(24), + )) + self.assertFalse(list(Path(tmp).glob("*.tmp")), "temp files left behind") + payload = json.loads((Path(tmp) / "task_meta.json").read_text()) + self.assertEqual(payload["fingerprint"], task_fingerprint(Readmission())) + + def test_write_failure_is_not_fatal(self): + """A diagnostic file must never abort a multi-hour build.""" + with tempfile.TemporaryDirectory() as tmp: + unwritable = Path(tmp) / "does" / "not" / "exist" + self.assertIsNotNone(write_task_metadata(unwritable, Readmission())) + + +class TestRealTasks(unittest.TestCase): + """Smoke test against tasks actually shipped in the package.""" + + def test_shipped_task_arguments_change_the_key(self): + from pyhealth.tasks.readmission_prediction import ReadmissionPredictionMIMIC3 + + t15 = ReadmissionPredictionMIMIC3() + t30 = ReadmissionPredictionMIMIC3(window=timedelta(days=30)) + self.assertNotEqual(task_fingerprint(t15), task_fingerprint(t30)) + self.assertEqual(task_fingerprint(t15), task_fingerprint(t15)) + self.assertTrue(task_cache_name(t15).startswith("ReadmissionPredictionMIMIC3_")) + + def test_benchmark_ehrshot_name_is_slugified(self): + from pyhealth.tasks.benchmark_ehrshot import BenchmarkEHRShot + + name = task_cache_name(BenchmarkEHRShot(task="guo_los")) + self.assertNotIn("/", name) + self.assertEqual(len(Path(name).parts), 1) + + +if __name__ == "__main__": + unittest.main(verbosity=2)