Repository: mmcdermott/MEDS_transforms

Issue:
Add `extract_numeric_values` transformation
This may leverage the match & revise syntax of #101

Discussion:

Patch:
diff --git a/pyproject.toml b/pyproject.toml
index bbc397d..cc193b6 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -54,6 +54,7 @@ MEDS_transform-filter_patients = "MEDS_transforms.filters.filter_patients:main"
 ## Transforms
 MEDS_transform-reorder_measurements = "MEDS_transforms.transforms.reorder_measurements:main"
 MEDS_transform-add_time_derived_measurements = "MEDS_transforms.transforms.add_time_derived_measurements:main"
+MEDS_transform-extract_values = "MEDS_transforms.transforms.extract_values:main"
 MEDS_transform-normalization = "MEDS_transforms.transforms.normalization:main"
 MEDS_transform-occlude_outliers = "MEDS_transforms.transforms.occlude_outliers:main"
 MEDS_transform-tensorization = "MEDS_transforms.transforms.tensorization:main"
diff --git a/src/MEDS_transforms/__init__.py b/src/MEDS_transforms/__init__.py
index e0aaaf3..38f2ac9 100644
--- a/src/MEDS_transforms/__init__.py
+++ b/src/MEDS_transforms/__init__.py
@@ -31,3 +31,11 @@ DEPRECATED_NAMES = {
     "timestamp": "time",
     "subject_id": "patient_id",
 }
+
+INFERRED_STAGE_KEYS = {
+    "is_metadata",
+    "data_input_dir",
+    "metadata_input_dir",
+    "output_dir",
+    "reducer_output_dir",
+}
diff --git a/src/MEDS_transforms/mapreduce/mapper.py b/src/MEDS_transforms/mapreduce/mapper.py
index 7649082..cdc60e0 100644
--- a/src/MEDS_transforms/mapreduce/mapper.py
+++ b/src/MEDS_transforms/mapreduce/mapper.py
@@ -3,7 +3,7 @@
 import inspect
 from collections.abc import Callable, Generator
 from datetime import datetime
-from enum import Enum, auto
+from enum import Enum, StrEnum, auto
 from functools import partial, wraps
 from pathlib import Path
 from typing import Any, NotRequired, TypedDict, TypeVar
@@ -223,6 +223,30 @@ def read_and_filter_fntr(filter_expr: pl.Expr, read_fn: Callable[[Path], DF_T])
 
 MATCH_REVISE_KEY = "_match_revise"
 MATCHER_KEY = "_matcher"
+MATCH_REVISE_MODE_KEY = "_match_revise_mode"
+
+
+class MatchReviseMode(StrEnum):
+    """The different modes for match and revise operations.
+
+    Future modes to be considered, match and add, multi-match and add, filter and revise, multi-filter and
+    revise.
+
+    Attributes:
+        MATCH_AND_REVISE: The match and revise mode, which iterates through the list of matcher/function pairs
+            and filters the input DataFrame for rows that match the matcher and applies a local compute
+            function to the filtered DataFrame. The DataFrame to be matched in future iterations is restricted
+            to only those rows that have not yet been matched. The unmatched dataframe at the end of the
+            operation is concatenated with the outputs of all intermediate dataframes.
+        MULTI_MATCH_AND_REVISE: The match and revise mode, which iterates through the list of matcher/function
+            pairs and filters the input DataFrame for rows that match the matcher and applies a local compute
+            function to the filtered DataFrame. The DataFrame to be matched in future iterations is the entire
+            raw dataframe, including rows that have already been matched. The portion of the dataframe that
+            didn't match anything on input is concatenated with the outputs of all intermediate dataframes.
+    """
+
+    MATCH_AND_REVISE = auto()
+    MULTI_MATCH_AND_REVISE = auto()
 
 
 def is_match_revise(stage_cfg: DictConfig) -> bool:
@@ -266,7 +290,7 @@ def validate_match_revise(stage_cfg: DictConfig):
         >>> validate_match_revise(DictConfig({"_match_revise": [{"_matcher": {32: "bar"}}]}))
         Traceback (most recent call last):
             ...
-        ValueError: Match revise config 0 must contain a valid matcher in _matcher
+        ValueError: Match revise config 0 must contain a valid matcher in _matcher: ...
         >>> validate_match_revise(DictConfig({"_match_revise": [{"_matcher": {"code": "CODE//TEMP"}}]}))
     """
 
@@ -284,8 +308,11 @@ def validate_match_revise(stage_cfg: DictConfig):
         if MATCHER_KEY not in match_revise_cfg:
             raise ValueError(f"Match revise config {i} must contain a {MATCHER_KEY} key")
 
-        if not is_matcher(match_revise_cfg[MATCHER_KEY]):
-            raise ValueError(f"Match revise config {i} must contain a valid matcher in {MATCHER_KEY}")
+        matcher_valid, matcher_errs = is_matcher(match_revise_cfg[MATCHER_KEY])
+        if not matcher_valid:
+            raise ValueError(
+                f"Match revise config {i} must contain a valid matcher in {MATCHER_KEY}: {matcher_errs}"
+            )
 
 
 def match_revise_fntr(cfg: DictConfig, stage_cfg: DictConfig, compute_fn: ANY_COMPUTE_FN_T) -> COMPUTE_FN_T:
@@ -412,6 +439,10 @@ def match_revise_fntr(cfg: DictConfig, stage_cfg: DictConfig, compute_fn: ANY_CO
 
     stage_cfg = hydra.utils.instantiate(stage_cfg)
 
+    match_revise_mode = stage_cfg.pop(MATCH_REVISE_MODE_KEY, "match_and_revise")
+    if match_revise_mode not in {x.value for x in MatchReviseMode}:
+        raise ValueError(f"Invalid match and revise mode: {match_revise_mode}")
+
     matchers_and_fns = []
     for match_revise_cfg in stage_cfg.pop(MATCH_REVISE_KEY):
         matcher, cols = matcher_to_expr(match_revise_cfg.pop(MATCHER_KEY))
@@ -422,10 +453,11 @@ def match_revise_fntr(cfg: DictConfig, stage_cfg: DictConfig, compute_fn: ANY_CO
 
     @wraps(compute_fn)
     def match_revise_fn(df: DF_T) -> DF_T:
-        unmatched_df = df
+        matchable_df = df
         cols = set(df.collect_schema().names())
 
         revision_parts = []
+        final_part_filters = []
         for i, (matcher_expr, need_cols, local_compute_fn) in enumerate(matchers_and_fns):
             if not need_cols.issubset(cols):
                 cols_str = "', '".join(x for x in sorted(cols))
@@ -433,12 +465,20 @@ def match_revise_fntr(cfg: DictConfig, stage_cfg: DictConfig, compute_fn: ANY_CO
                     f"Missing needed columns {need_cols - cols} for local matcher {i}: "
                     f"{matcher_expr}\nColumns available: '{cols_str}'"
                 )
-            matched_df = unmatched_df.filter(matcher_expr)
-            unmatched_df = unmatched_df.filter(~matcher_expr)
+            matched_df = matchable_df.filter(matcher_expr)
+
+            match match_revise_mode:
+                case MatchReviseMode.MATCH_AND_REVISE:
+                    matchable_df = matchable_df.filter(~matcher_expr)
+                case MatchReviseMode.MULTI_MATCH_AND_REVISE:
+                    final_part_filters.append(~matcher_expr)
 
             revision_parts.append(local_compute_fn(matched_df))
 
-        revision_parts.append(unmatched_df)
+        if final_part_filters:
+            revision_parts.append(matchable_df.filter(pl.all_horizontal(final_part_filters)))
+        else:
+            revision_parts.append(matchable_df)
         return pl.concat(revision_parts, how="vertical").sort(["patient_id", "time"], maintain_order=True)
 
     return match_revise_fn
diff --git a/src/MEDS_transforms/parser.py b/src/MEDS_transforms/parser.py
index 5ac4830..948ca00 100644
--- a/src/MEDS_transforms/parser.py
+++ b/src/MEDS_transforms/parser.py
@@ -32,13 +32,13 @@ from __future__ import annotations
 
 import re
 from enum import StrEnum
-from typing import Any
+from typing import Annotated, Any
 
 import polars as pl
 from omegaconf import DictConfig, ListConfig, OmegaConf
 
 
-def is_matcher(matcher_cfg: dict[str, Any]) -> bool:
+def is_matcher(matcher_cfg: dict[str, Any]) -> tuple[bool, str | None]:
     """Checks if a dictionary is a valid matcher configuration.
 
     Args:
@@ -49,17 +49,34 @@ def is_matcher(matcher_cfg: dict[str, Any]) -> bool:
 
     Examples:
         >>> is_matcher({"foo": "bar"})
-        True
+        (True, None)
         >>> is_matcher(DictConfig({"foo": "bar"}))
-        True
-        >>> is_matcher({"foo": "bar", 32: "baz"})
-        False
+        (True, None)
         >>> is_matcher(["foo", "bar"])
-        False
+        (False, "Matcher configuration must be a dictionary. Got ['foo', 'bar']")
+        >>> is_matcher({"foo": "bar", 32: "baz"})
+        (False, "Matcher configuration must be a dictionary with string keys. Got {'foo': 'bar', 32: 'baz'}")
+        >>> is_matcher({"foo": {"regex": "bar"}})
+        (True, None)
+        >>> is_matcher({"foo": {"regex": "bar", "group_index": 0}}) # doctest: +NORMALIZE_WHITESPACE
+        (False, "If the matcher spec is a dictionary, it must have only a 'regex' key.
+                 Got {'regex': 'bar', 'group_index': 0} for foo")
         >>> is_matcher({})
-        True
+        (True, None)
     """
-    return isinstance(matcher_cfg, (dict, DictConfig)) and all(isinstance(k, str) for k in matcher_cfg.keys())
+    if not isinstance(matcher_cfg, (dict, DictConfig)):
+        return False, f"Matcher configuration must be a dictionary. Got {matcher_cfg}"
+    if not all(isinstance(k, str) for k in matcher_cfg.keys()):
+        return False, f"Matcher configuration must be a dictionary with string keys. Got {matcher_cfg}"
+    for k, cfg in matcher_cfg.items():
+        if isinstance(cfg, (dict, DictConfig)) and set(cfg.keys()) != {
+            "regex",
+        }:
+            return False, (
+                f"If the matcher spec is a dictionary, it must have only a 'regex' key. Got {cfg} for {k}"
+            )
+
+    return True, None
 
 
 def matcher_to_expr(matcher_cfg: DictConfig | dict) -> tuple[pl.Expr, set[str]]:
@@ -89,15 +106,29 @@ def matcher_to_expr(matcher_cfg: DictConfig | dict) -> tuple[pl.Expr, set[str]]:
         [(col("foo")) == (String(bar))].all_horizontal([[(col("buzz")) == (String(baz))]])
         >>> sorted(cols)
         ['buzz', 'foo']
+        >>> expr, cols = matcher_to_expr(DictConfig({"foo": "bar", "buzz": {"regex": "baz"}}))
+        >>> print(expr)
+        [(col("foo")) == (String(bar))].all_horizontal([col("buzz").str.contains([String(baz)])])
+        >>> sorted(cols)
+        ['buzz', 'foo']
         >>> matcher_to_expr(["foo", "bar"])
         Traceback (most recent call last):
             ...
-        ValueError: Matcher configuration must be a dictionary with string keys. Got: ['foo', 'bar']
+        ValueError: Matcher configuration must be a dictionary. Got ['foo', 'bar']
     """
-    if not is_matcher(matcher_cfg):
-        raise ValueError(f"Matcher configuration must be a dictionary with string keys. Got: {matcher_cfg}")
+    matcher_valid, matcher_errs = is_matcher(matcher_cfg)
+    if not matcher_valid:
+        raise ValueError(matcher_errs)
 
-    return pl.all_horizontal((pl.col(k) == v) for k, v in matcher_cfg.items()), set(matcher_cfg.keys())
+    all_exprs = []
+    for k, v in matcher_cfg.items():
+        if isinstance(v, (dict, DictConfig)):
+            expr = pl.col(k).str.contains(v["regex"])
+        else:
+            expr = pl.col(k) == v
+        all_exprs.append(expr)
+
+    return pl.all_horizontal(all_exprs), set(matcher_cfg.keys())
 
 
 STR_INTERPOLATION_REGEX = r"\{([^}]+)\}"
@@ -112,11 +143,75 @@ class ColExprType(StrEnum):
             via python's f-string syntax.
         LITERAL: A column expression that is a literal value regardless of type. No interpolation is allowed
             here.
+        EXTRACT: A column expression that extracts a substring from a column using a regex pattern.
     """
 
     COL = "col"
     STR = "str"
     LITERAL = "literal"
+    EXTRACT = "extract"
+
+    @classmethod
+    def _is_valid_extract_cfg(cls, cfg: dict[Any, Any]) -> tuple[bool, str | None]:
+        """Checks if a dictionary is a valid extract configuration.
+
+        Args:
+            cfg: A dictionary of extract configuration. This is the "inner part" of an overall column
+                expression."
+
+        Returns:
+            bool: True if the input is a valid extract configuration, False otherwise.
+            str | None: The reason the input is invalid, if it is invalid.
+
+        Examples:
+            >>> ColExprType._is_valid_extract_cfg({"from": "foo", "regex": "bar"})
+            (True, None)
+            >>> ColExprType._is_valid_extract_cfg({"from": "foo", "regex": "bar", "group_index": 0})
+            (True, None)
+            >>> ColExprType._is_valid_extract_cfg(32)
+            (False, 'Extract expressions must be a dictionary. Got 32')
+            >>> ColExprType._is_valid_extract_cfg({"from": "foo"})
+            (False, "Extract expressions must have a 'from' and 'regex' key. Got ['from']")
+            >>> ColExprType._is_valid_extract_cfg({"from": ["buzz"], "regex": "bar"})
+            (False, "Extract expressions must have a <class 'str'> value for 'from'. Got ['buzz']")
+            >>> ColExprType._is_valid_extract_cfg({"from": "foo", "regex": 32})
+            (False, "Extract expressions must have a <class 'str'> value for 'regex'. Got 32")
+            >>> ColExprType._is_valid_extract_cfg({"from": "foo", "regex": "bar", "group_index": "baz"})
+            (False, "Extract expressions must have a non-negative integer value for 'group_index'. Got baz")
+            >>> ColExprType._is_valid_extract_cfg({"from": "foo", "regex": "bar", "group_index": -1})
+            (False, "Extract expressions must have a non-negative integer value for 'group_index'. Got -1")
+            >>> ColExprType._is_valid_extract_cfg({"from": "foo", "regex": "bar", "ex": "baz"})
+            (False, "Extract expressions must have only 'from', 'regex', and 'group_index' keys. Got ['ex']")
+        """
+
+        MANDATORY_KEYS = {"from": str, "regex": str}
+        ALLOWED_KEYS = {**MANDATORY_KEYS, "group_index": Annotated[int, "non-negative integer"]}
+
+        if not isinstance(cfg, (dict, DictConfig)):
+            return False, f"Extract expressions must be a dictionary. Got {cfg}"
+        if not set(MANDATORY_KEYS).issubset(set(cfg.keys())):
+            return False, f"Extract expressions must have a 'from' and 'regex' key. Got {list(cfg.keys())}"
+        if not set(ALLOWED_KEYS).issuperset(set(cfg.keys())):
+            return False, (
+                "Extract expressions must have only 'from', 'regex', and 'group_index' keys. "
+                f"Got {sorted(list(set(cfg.keys()) - set(ALLOWED_KEYS)))}"
+            )
+
+        for key, allowed_T in ALLOWED_KEYS.items():
+            if key not in cfg:
+                continue
+            val = cfg[key]
+
+            if key == "group_index":
+                if not isinstance(val, int) or val < 0:
+                    return False, (
+                        f"Extract expressions must have a non-negative integer value for '{key}'. Got {val}"
+                    )
+            else:
+                if not isinstance(val, allowed_T):
+                    return False, f"Extract expressions must have a {allowed_T} value for '{key}'. Got {val}"
+
+        return True, None
 
     @classmethod
     def is_valid(cls, expr_dict: dict[ColExprType, Any]) -> tuple[bool, str | None]:
@@ -144,12 +239,12 @@ class ColExprType(StrEnum):
             (False, "Column expressions can only contain a single key-value pair.
                     Got {'col': 'foo', 'str': 'bar'}")
             >>> ColExprType.is_valid({"foo": "bar"})
-            (False, "Column expressions must have a key in ColExprType: ['col', 'str', 'literal']. Got foo")
+            (False, "Column expressions must have a key in ColExprType: ... Got foo")
             >>> ColExprType.is_valid([("col", "foo")])
             (False, "Column expressions must be a dictionary. Got [('col', 'foo')]")
         """
 
-        if not isinstance(expr_dict, dict):
+        if not isinstance(expr_dict, (dict, DictConfig)):
             return False, f"Column expressions must be a dictionary. Got {expr_dict}"
         if len(expr_dict) != 1:
             return False, f"Column expressions can only contain a single key-value pair. Got {expr_dict}"
@@ -166,6 +261,8 @@ class ColExprType(StrEnum):
                 return False, f"String interpolation expressions must have a string value. Got {expr_val}"
             case cls.LITERAL:
                 return True, None
+            case cls.EXTRACT:
+                return cls._is_valid_extract_cfg(expr_val)
             case _:
                 return (
                     False,
@@ -203,6 +300,11 @@ class ColExprType(StrEnum):
             ['foo', 'bar']
             >>> cols
             set()
+            >>> expr, cols = ColExprType.to_pl_expr(ColExprType.EXTRACT, {"from": "foo", "regex": "bar"})
+            >>> print(expr)
+            col("foo").str.extract([String(bar)])
+            >>> sorted(cols)
+            ['foo']
             >>> ColExprType.to_pl_expr(ColExprType.COL, 32)
             Traceback (most recent call last):
                 ...
@@ -223,6 +325,12 @@ class ColExprType(StrEnum):
                 if isinstance(expr_val, ListConfig):
                     expr_val = OmegaConf.to_object(expr_val)
                 return pl.lit(expr_val), set()
+            case cls.EXTRACT:
+                expr = pl.col(expr_val["from"]).str.extract(
+                    expr_val["regex"], group_index=expr_val.get("group_index", 1)
+                )
+                cols = {expr_val["from"]}
+                return expr, cols
 
 
 def parse_col_expr(cfg: str | list | dict[str, str] | ListConfig | DictConfig) -> dict:
@@ -249,17 +357,17 @@ def parse_col_expr(cfg: str | list | dict[str, str] | ListConfig | DictConfig) -
         >>> parse_col_expr({"output": "foo", "matcher": {32: "baz"}}) # doctest: +NORMALIZE_WHITESPACE
         Traceback (most recent call last):
             ...
-        ValueError: A pre-specified output/matcher configuration must have a valid matcher dictionary,
-                    which is a dictionary with string-type keys. Got cfg['matcher']={32: 'baz'}
+        ValueError: A pre-specified output/matcher configuration must have a valid matcher dictionary. Got
+                    cfg['matcher']={32: 'baz'} which has errors: ...
         >>> parse_col_expr({"foo": {"bar": "baz"}})
         {'output': {'col': 'foo'}, 'matcher': {'bar': 'baz'}}
         >>> parse_col_expr({"foo": {32: "baz"}}) # doctest: +NORMALIZE_WHITESPACE
         Traceback (most recent call last):
             ...
         ValueError: A simple-form conditional expression is expressed with a single key-value pair dict,
-                    where the key is not a column expression type and the value is a valid matcher dict,
-                    which is a dictionary with string-type keys. This config has a single key-value pair
-                    with key foo but an invalid matcher: {32: 'baz'}
+                    where the key is not a column expression type and the value is a valid matcher dict. This
+                    config has a single key-value pair with key foo but an invalid matcher: {32: 'baz'} which
+                    has errors: ...
         >>> parse_col_expr(["bar//{foo}", {"str": "bar//UNK"}])
         [{'str': 'bar//{foo}'}, {'str': 'bar//UNK'}]
         >>> parse_col_expr({"foo": "bar", "buzz": "baz", "fuzz": "fizz"}) # doctest: +NORMALIZE_WHITESPACE
@@ -303,25 +411,27 @@ def parse_col_expr(cfg: str | list | dict[str, str] | ListConfig | DictConfig) -
                 "If a list (which coalesces columns), all elements must be strings or dictionaries. "
                 f"Got: {cfg}"
             )
-        case dict() | DictConfig() if set(cfg.keys()) == {"output", "matcher"} and is_matcher(cfg["matcher"]):
-            return {"output": parse_col_expr(cfg["output"]), "matcher": cfg["matcher"]}
         case dict() | DictConfig() if set(cfg.keys()) == {"output", "matcher"}:
-            raise ValueError(
-                "A pre-specified output/matcher configuration must have a valid matcher dictionary, which is "
-                f"a dictionary with string-type keys. Got cfg['matcher']={cfg['matcher']}"
-            )
+            matcher_valid, matcher_errs = is_matcher(cfg["matcher"])
+            if not matcher_valid:
+                raise ValueError(
+                    "A pre-specified output/matcher configuration must have a valid matcher dictionary. "
+                    f"Got cfg['matcher']={cfg['matcher']} which has errors: {matcher_errs}"
+                )
+            return {"output": parse_col_expr(cfg["output"]), "matcher": cfg["matcher"]}
         case dict() | DictConfig() if len(cfg) == 1 and ColExprType.is_valid(cfg)[0]:
             return cfg
         case dict() | DictConfig() if len(cfg) == 1:
             out_cfg, matcher_cfg = next(iter(cfg.items()))
-            if is_matcher(matcher_cfg):
+            matcher_valid, matcher_errs = is_matcher(matcher_cfg)
+            if matcher_valid:
                 return {"output": parse_col_expr(out_cfg), "matcher": matcher_cfg}
             else:
                 raise ValueError(
                     "A simple-form conditional expression is expressed with a single key-value pair dict, "
-                    "where the key is not a column expression type and the value is a valid matcher dict, "
-                    "which is a dictionary with string-type keys. This config has a single key-value pair "
-                    f"with key {out_cfg} but an invalid matcher: {matcher_cfg}"
+                    "where the key is not a column expression type and the value is a valid matcher dict. "
+                    f"This config has a single key-value pair with key {out_cfg} but an invalid matcher: "
+                    f"{matcher_cfg} which has errors: {matcher_errs}"
                 )
         case dict() | DictConfig():
             raise ValueError(
@@ -362,11 +472,16 @@ def structured_expr_to_pl(cfg: dict | list[dict] | ListConfig | DictConfig) -> t
         .when([(col("bar")) == (String(baz))].all_horizontal()).then(String(foo)).otherwise(null)
         >>> sorted(cols)
         ['bar']
-        >>> expr, cols = structured_expr_to_pl({"col": "bar"})
+        >>> expr, cols = structured_expr_to_pl({"extract": {"from": "foo", "regex": r"b/(.*)"}})
         >>> print(expr)
-        col("bar")
+        col("foo").str.extract([String(b/(.*))])
         >>> sorted(cols)
-        ['bar']
+        ['foo']
+        >>> expr, cols = structured_expr_to_pl({"extract": {"from": "foo", "regex": "bar", "group_index": 0}})
+        >>> print(expr)
+        col("foo").str.extract([String(bar)])
+        >>> sorted(cols)
+        ['foo']
         >>> structured_expr_to_pl(["foo", 32])
         Traceback (most recent call last):
             ...
@@ -379,8 +494,8 @@ def structured_expr_to_pl(cfg: dict | list[dict] | ListConfig | DictConfig) -> t
         >>> structured_expr_to_pl({"output": "foo", "matcher": {32: "baz"}}) # doctest: +NORMALIZE_WHITESPACE
         Traceback (most recent call last):
             ...
-        ValueError: A pre-specified output/matcher configuration must have a valid matcher dictionary, which
-                    is a dictionary with string-type keys. Got cfg['matcher']={32: 'baz'}
+        ValueError: A pre-specified output/matcher configuration must have a valid matcher dictionary. Got
+                    cfg['matcher']={32: 'baz'} which has errors: ...
         >>> structured_expr_to_pl({"col": 32})
         Traceback (most recent call last):
             ...
@@ -403,19 +518,21 @@ def structured_expr_to_pl(cfg: dict | list[dict] | ListConfig | DictConfig) -> t
                 component_exprs.append(expr)
                 needed_cols.update(cols)
             return pl.coalesce(*component_exprs), needed_cols
-        case dict() | DictConfig() if set(cfg.keys()) == {"output", "matcher"} and is_matcher(cfg["matcher"]):
+        case dict() | DictConfig() if set(cfg.keys()) == {"output", "matcher"}:
+            matcher_valid, matcher_errs = is_matcher(cfg["matcher"])
+            if not matcher_valid:
+                raise ValueError(
+                    "A pre-specified output/matcher configuration must have a valid matcher dictionary. "
+                    f"Got cfg['matcher']={cfg['matcher']} which has errors: {matcher_errs}"
+                )
+
             matcher_expr, matcher_cols = matcher_to_expr(cfg["matcher"])
             try:
                 out_expr, out_cols = cfg_to_expr(cfg["output"])
             except ValueError as e:
                 raise ValueError(f"Error processing output/matcher config output expression for {cfg}") from e
             return pl.when(matcher_expr).then(out_expr), out_cols | matcher_cols
-        case dict() | DictConfig() if set(cfg.keys()) == {"output", "matcher"}:
-            # TODO(mmd): DRY out this and other error messages.
-            raise ValueError(
-                "A pre-specified output/matcher configuration must have a valid matcher dictionary, which is "
-                f"a dictionary with string-type keys. Got cfg['matcher']={cfg['matcher']}"
-            )
+
         case dict() | DictConfig() if ColExprType.is_valid(cfg)[0]:
             expr_type, expr_val = next(iter(cfg.items()))
             return ColExprType.to_pl_expr(expr_type, expr_val)
@@ -465,6 +582,11 @@ def cfg_to_expr(cfg: str | ListConfig | DictConfig) -> tuple[pl.Expr, set[str]]:
         ['1//d', None, None]
         >>> sorted(cols)
         ['bar', 'baz', 'foo']
+        >>> expr, cols = cfg_to_expr({"extract": {"regex": "([ac]).*", "from": "foo"}})
+        >>> data.select(expr.alias("out"))["out"].to_list()
+        ['a', None, 'c']
+        >>> sorted(cols)
+        ['foo']
         >>> cfg = [
         ...    {"matcher": {"baz": 2}, "output": {"str": "bar//{baz}"}},
         ...    {"literal": "34.2"},
diff --git a/src/MEDS_transforms/transforms/add_time_derived_measurements.py b/src/MEDS_transforms/transforms/add_time_derived_measurements.py
index 19a7abf..01ec7f9 100644
--- a/src/MEDS_transforms/transforms/add_time_derived_measurements.py
+++ b/src/MEDS_transforms/transforms/add_time_derived_measurements.py
@@ -7,7 +7,7 @@ import polars as pl
 from loguru import logger
 from omegaconf import DictConfig, OmegaConf
 
-from MEDS_transforms import PREPROCESS_CONFIG_YAML
+from MEDS_transforms import INFERRED_STAGE_KEYS, PREPROCESS_CONFIG_YAML
 from MEDS_transforms.mapreduce.mapper import map_over
 
 
@@ -365,14 +365,6 @@ def time_of_day_fntr(cfg: DictConfig) -> Callable[[pl.DataFrame], pl.DataFrame]:
 
 
 def add_time_derived_measurements_fntr(stage_cfg: DictConfig) -> Callable[[pl.LazyFrame], pl.LazyFrame]:
-    INFERRED_STAGE_KEYS = {
-        "is_metadata",
-        "data_input_dir",
-        "metadata_input_dir",
-        "output_dir",
-        "reducer_output_dir",
-    }
-
     compute_fns = []
     # We use the raw stages object as the induced `stage_cfg` has extra properties like the input and output
     # directories.
diff --git a/src/MEDS_transforms/transforms/extract_values.py b/src/MEDS_transforms/transforms/extract_values.py
new file mode 100644
index 0000000..a1d42b6
--- /dev/null
+++ b/src/MEDS_transforms/transforms/extract_values.py
@@ -0,0 +1,134 @@
+#!/usr/bin/env python
+"""Transformations for extracting numeric and/or categorical values from the MEDS dataset."""
+from collections.abc import Callable
+
+import hydra
+import polars as pl
+from loguru import logger
+from omegaconf import DictConfig
+
+from MEDS_transforms import DEPRECATED_NAMES, INFERRED_STAGE_KEYS, MANDATORY_TYPES, PREPROCESS_CONFIG_YAML
+from MEDS_transforms.mapreduce.mapper import map_over
+from MEDS_transforms.parser import cfg_to_expr
+
+
+def extract_values_fntr(stage_cfg: DictConfig) -> Callable[[pl.LazyFrame], pl.LazyFrame]:
+    """Create a function that extracts values from a MEDS cohort.
+
+    This functor does not filter the applied dataframe prior to applying the extraction process. It is likely
+    best used with match & revise to filter the data before applying the extraction process.
+
+    Args:
+        stage_cfg: The configuration for the extraction stage. This should be a mapping from output column
+            name to the parser configuration for the value you want to extract from the MEDS measurement for
+            that column.
+
+    Returns:
+        A function that takes a LazyFrame and returns a LazyFrame with the original data and the extracted
+        values.
+
+    Examples:
+        >>> stage_cfg = {"numeric_value": "foo", "categorical_value": "bar"}
+        >>> fn = extract_values_fntr(stage_cfg)
+        >>> df = pl.DataFrame({
+        ...     "patient_id": [1, 1, 1], "time": [1, 2, 3],
+        ...     "foo": ["1", "2", "3"], "bar": [1.0, 2.0, 4.0],
+        ... })
+        >>> fn(df)
+        shape: (3, 6)
+        ┌────────────┬──────┬─────┬─────┬───────────────┬───────────────────┐
+        │ patient_id ┆ time ┆ foo ┆ bar ┆ numeric_value ┆ categorical_value │
+        │ ---        ┆ ---  ┆ --- ┆ --- ┆ ---           ┆ ---               │
+        │ i64        ┆ i64  ┆ str ┆ f64 ┆ f32           ┆ str               │
+        ╞════════════╪══════╪═════╪═════╪═══════════════╪═══════════════════╡
+        │ 1          ┆ 1    ┆ 1   ┆ 1.0 ┆ 1.0           ┆ 1.0               │
+        │ 1          ┆ 2    ┆ 2   ┆ 2.0 ┆ 2.0           ┆ 2.0               │
+        │ 1          ┆ 3    ┆ 3   ┆ 4.0 ┆ 3.0           ┆ 4.0               │
+        └────────────┴──────┴─────┴─────┴───────────────┴───────────────────┘
+        >>> stage_cfg = {32: "foo"}
+        >>> fn = extract_values_fntr(stage_cfg)
+        Traceback (most recent call last):
+            ...
+        ValueError: Invalid column name: 32
+        >>> stage_cfg = {"numeric_value": {"lit": 1}}
+        >>> fn = extract_values_fntr(stage_cfg)
+        Traceback (most recent call last):
+            ...
+        ValueError: Error building expression for numeric_value...
+        >>> stage_cfg = {"numeric_value": "foo", "categorical_value": "bar"}
+        >>> fn = extract_values_fntr(stage_cfg)
+        >>> df = pl.DataFrame({"patient_id": [1, 1, 1], "time": [1, 2, 3]})
+        >>> fn(df)
+        Traceback (most recent call last):
+            ...
+        ValueError: Missing columns: ['bar', 'foo']
+
+    Note that deprecated column names like "numerical_value" or "timestamp" won't be re-typed.
+        >>> stage_cfg = {"numerical_value": "foo"}
+        >>> fn = extract_values_fntr(stage_cfg)
+        >>> df = pl.DataFrame({"patient_id": [1, 1, 1], "time": [1, 2, 3], "foo": ["1", "2", "3"]})
+        >>> fn(df)
+        shape: (3, 4)
+        ┌────────────┬──────┬─────┬─────────────────┐
+        │ patient_id ┆ time ┆ foo ┆ numerical_value │
+        │ ---        ┆ ---  ┆ --- ┆ ---             │
+        │ i64        ┆ i64  ┆ str ┆ str             │
+        ╞════════════╪══════╪═════╪═════════════════╡
+        │ 1          ┆ 1    ┆ 1   ┆ 1               │
+        │ 1          ┆ 2    ┆ 2   ┆ 2               │
+        │ 1          ┆ 3    ┆ 3   ┆ 3               │
+        └────────────┴──────┴─────┴─────────────────┘
+    """
+
+    new_cols = []
+    need_cols = set()
+    for out_col_n, value_cfg in stage_cfg.items():
+        if out_col_n in INFERRED_STAGE_KEYS:
+            continue
+
+        try:
+            expr, cols = cfg_to_expr(value_cfg)
+        except ValueError as e:
+            raise ValueError(f"Error building expression for {out_col_n}") from e
+
+        match out_col_n:
+            case str() if out_col_n in MANDATORY_TYPES:
+                expr = expr.cast(MANDATORY_TYPES[out_col_n])
+                if out_col_n == "patient_id":
+                    logger.warning("You should almost CERTAINLY not be extracting patient_id as a value.")
+                if out_col_n == "time":
+                    logger.warning("Warning: `time` is being extracted post-hoc!")
+            case str() if out_col_n in DEPRECATED_NAMES:
+                logger.warning(
+                    f"Deprecated column name: {out_col_n} -> {DEPRECATED_NAMES[out_col_n]}. "
+                    "This column name will not be re-typed."
+                )
+            case str():
+                pass
+            case _:
+                raise ValueError(f"Invalid column name: {out_col_n}")
+
+        new_cols.append(expr.alias(out_col_n))
+        need_cols.update(cols)
+
+    def compute_fn(df: pl.LazyFrame) -> pl.LazyFrame:
+        in_cols = set(df.collect_schema().names())
+        if not need_cols.issubset(in_cols):
+            raise ValueError(f"Missing columns: {sorted(list(need_cols - in_cols))}")
+
+        return df.with_columns(new_cols).sort("patient_id", "time", maintain_order=True)
+
+    return compute_fn
+
+
+@hydra.main(
+    version_base=None, config_path=str(PREPROCESS_CONFIG_YAML.parent), config_name=PREPROCESS_CONFIG_YAML.stem
+)
+def main(cfg: DictConfig):
+    """TODO."""
+
+    map_over(cfg, compute_fn=extract_values_fntr)
+
+
+if __name__ == "__main__":
+    main()


Test patch:
diff --git a/tests/test_extract_values.py b/tests/test_extract_values.py
new file mode 100644
index 0000000..8a3a65b
--- /dev/null
+++ b/tests/test_extract_values.py
@@ -0,0 +1,116 @@
+"""Tests the extract values script.
+
+Set the bash env variable `DO_USE_LOCAL_SCRIPTS=1` to use the local py files, rather than the installed
+scripts.
+"""
+
+from .transform_tester_base import EXTRACT_VALUES_SCRIPT, parse_shards_yaml, single_stage_transform_tester
+
+INPUT_SHARDS = parse_shards_yaml(
+    """
+    train/0: |-2
+      patient_id,time,code,numeric_value,text_value
+      239684,,EYE_COLOR//BROWN,,
+      239684,"12/28/1980, 00:00:00",DOB,,
+      239684,"05/11/2010, 17:41:51",BP,,"120/80"
+      1195293,,EYE_COLOR//BLUE,,
+      1195293,"06/20/1978, 00:00:00",DOB,,
+      1195293,"06/20/2010, 19:23:52",BP,,"144/96"
+      1195293,"06/20/2010, 19:23:52",HR,80,
+      1195293,"06/20/2010, 19:23:52",TEMP,,"100F"
+    train/1: |-2
+      patient_id,time,code,numeric_value,text_value
+      68729,,EYE_COLOR//HAZEL,,
+      68729,"03/09/1978, 00:00:00",DOB,,
+      814703,"02/05/2010, 05:55:39",HR,170.2,
+    tuning/0: |-2
+      patient_id,time,code,numeric_value,text_value
+      754281,,EYE_COLOR//BROWN,,
+      754281,"12/19/1988, 00:00:00",DOB,,
+      754281,"01/03/2010, 06:27:59",HR,142.0,
+      754281,"06/20/2010, 20:23:50",BP,,"134/76"
+      754281,"06/20/2010, 21:00:02",TEMP,,"36.2C"
+    held_out/0: |-2
+      patient_id,time,code,numeric_value,text_value
+      1500733,,EYE_COLOR//BROWN,,
+      1500733,"07/20/1986, 00:00:00",DOB,,
+      1500733,"06/03/2010, 14:54:38",HR,91.4
+      1500733,"06/03/2010, 14:54:38",BP,,"123/82"
+    """
+)
+
+WANT_SHARDS = parse_shards_yaml(
+    """
+    train/0: |-2
+      patient_id,time,code,numeric_value,text_value
+      239684,,EYE_COLOR//BROWN,,
+      239684,"12/28/1980, 00:00:00",DOB,,
+      239684,"05/11/2010, 17:41:51",BP//SYSTOLIC,120,
+      239684,"05/11/2010, 17:41:51",BP//DIASTOLIC,80,
+      1195293,,EYE_COLOR//BLUE,,
+      1195293,"06/20/1978, 00:00:00",DOB,,
+      1195293,"06/20/2010, 19:23:52",BP//SYSTOLIC,144,
+      1195293,"06/20/2010, 19:23:52",BP//DIASTOLIC,96,
+      1195293,"06/20/2010, 19:23:52",TEMP//F,100,
+      1195293,"06/20/2010, 19:23:52",HR,80,
+    train/1: |-2
+      patient_id,time,code,numeric_value,text_value
+      68729,,EYE_COLOR//HAZEL,,
+      68729,"03/09/1978, 00:00:00",DOB,,
+      814703,"02/05/2010, 05:55:39",HR,170.2,
+    tuning/0: |-2
+      patient_id,time,code,numeric_value,text_value
+      754281,,EYE_COLOR//BROWN,,
+      754281,"12/19/1988, 00:00:00",DOB,,
+      754281,"01/03/2010, 06:27:59",HR,142.0,
+      754281,"06/20/2010, 20:23:50",BP//SYSTOLIC,134,
+      754281,"06/20/2010, 20:23:50",BP//DIASTOLIC,76,
+      754281,"06/20/2010, 21:00:02",TEMP//C,36.2,
+    held_out/0: |-2
+      patient_id,time,code,numeric_value,text_value
+      1500733,,EYE_COLOR//BROWN,,
+      1500733,"07/20/1986, 00:00:00",DOB,,
+      1500733,"06/03/2010, 14:54:38",BP//SYSTOLIC,123,
+      1500733,"06/03/2010, 14:54:38",BP//DIASTOLIC,82,
+      1500733,"06/03/2010, 14:54:38",HR,91.4,
+    """
+)
+
+
+def test_extract_values():
+    single_stage_transform_tester(
+        transform_script=EXTRACT_VALUES_SCRIPT,
+        stage_name="extract_values",
+        transform_stage_kwargs={
+            "_match_revise_mode": "multi_match_and_revise",
+            "_match_revise": [
+                {
+                    "_matcher": {"code": "BP"},
+                    "numeric_value": {"extract": {"from": "text_value", "regex": r"(\d+)/.*"}},
+                    "code": "{code}//SYSTOLIC",
+                    "text_value": {"literal": None},
+                },
+                {
+                    "_matcher": {"code": "BP"},
+                    "numeric_value": {"extract": {"from": "text_value", "regex": r".*/(\d+)"}},
+                    "code": "{code}//DIASTOLIC",
+                    "text_value": {"literal": None},
+                },
+                {
+                    "_matcher": {"code": "TEMP", "text_value": {"regex": r"^[\d\.]+C$"}},
+                    "numeric_value": {"extract": {"from": "text_value", "regex": r"^([\d\.]+)C$"}},
+                    "code": "{code}//C",
+                    "text_value": {"literal": None},
+                },
+                {
+                    "_matcher": {"code": "TEMP", "text_value": {"regex": r"^[\d\.]+F$"}},
+                    "numeric_value": {"extract": {"from": "text_value", "regex": r"^([\d\.]+)F$"}},
+                    "code": "{code}//F",
+                    "text_value": {"literal": None},
+                },
+            ],
+        },
+        input_shards=INPUT_SHARDS,
+        want_data=WANT_SHARDS,
+        do_use_config_yaml=True,
+    )
diff --git a/tests/test_filter_measurements.py b/tests/test_filter_measurements.py
index cb919d1..3bc807f 100644
--- a/tests/test_filter_measurements.py
+++ b/tests/test_filter_measurements.py
@@ -214,9 +214,7 @@ def test_match_revise_filter_measurements():
         stage_name="filter_measurements",
         transform_stage_kwargs={
             "_match_revise": [
-                {"_matcher": {"code": "ADMISSION//CARDIAC"}, "min_patients_per_code": 2},
-                {"_matcher": {"code": "ADMISSION//ORTHOPEDIC"}, "min_patients_per_code": 2},
-                {"_matcher": {"code": "ADMISSION//PULMONARY"}, "min_patients_per_code": 2},
+                {"_matcher": {"code": {"regex": "ADMISSION//.*"}}, "min_patients_per_code": 2},
                 {"_matcher": {"code": "HR"}, "min_patients_per_code": 15},
                 {"_matcher": {"code": "EYE_COLOR//BLUE"}, "min_patients_per_code": 4},
                 {"_matcher": {"code": "EYE_COLOR//BROWN"}, "min_patients_per_code": 4},
diff --git a/tests/transform_tester_base.py b/tests/transform_tester_base.py
index 6845e0c..bca36ad 100644
--- a/tests/transform_tester_base.py
+++ b/tests/transform_tester_base.py
@@ -45,6 +45,7 @@ if os.environ.get("DO_USE_LOCAL_SCRIPTS", "0") == "1":
     # Transforms
     ADD_TIME_DERIVED_MEASUREMENTS_SCRIPT = transforms_root / "add_time_derived_measurements.py"
     REORDER_MEASUREMENTS_SCRIPT = transforms_root / "reorder_measurements.py"
+    EXTRACT_VALUES_SCRIPT = transforms_root / "extract_values.py"
     NORMALIZATION_SCRIPT = transforms_root / "normalization.py"
     OCCLUDE_OUTLIERS_SCRIPT = transforms_root / "occlude_outliers.py"
     TENSORIZATION_SCRIPT = transforms_root / "tensorization.py"
@@ -62,6 +63,7 @@ else:
     # Transforms
     ADD_TIME_DERIVED_MEASUREMENTS_SCRIPT = "MEDS_transform-add_time_derived_measurements"
     REORDER_MEASUREMENTS_SCRIPT = "MEDS_transform-reorder_measurements"
+    EXTRACT_VALUES_SCRIPT = "MEDS_transform-extract_values"
     NORMALIZATION_SCRIPT = "MEDS_transform-normalization"
     OCCLUDE_OUTLIERS_SCRIPT = "MEDS_transform-occlude_outliers"
     TENSORIZATION_SCRIPT = "MEDS_transform-tensorization"
diff --git a/tests/utils.py b/tests/utils.py
index e6c9d3f..e7220c9 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -13,9 +13,10 @@ DEFAULT_CSV_TS_FORMAT = "%m/%d/%Y, %H:%M:%S"
 MEDS_PL_SCHEMA = {
     "patient_id": pl.UInt32,
     "time": pl.Datetime("us"),
-    "code": pl.Utf8,
+    "code": pl.String,
     "numeric_value": pl.Float32,
     "numeric_value/is_inlier": pl.Boolean,
+    "text_value": pl.String,
 }