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)