Skip to content

x_partitioner

`knime2py.nodes.x_partitioner`

X-Validation Partitioner module.

Overview

This module maps KNIME’s “X-Partitioner” node to Python, emitting code that plans cross-validation folds and opens a loop that produces train/test DataFrames for each fold.

Runtime Behavior

Inputs: - Reads a DataFrame from the context using the key corresponding to the input port.

Outputs: - Writes train and test DataFrames to the context with keys formatted as '{node_id}:1' and '{node_id}:2', respectively.

Key algorithms: - Utilizes scikit-learn's KFold, StratifiedKFold, and LeaveOneOut for partitioning. - Implements stratified sampling based on a specified class column.

Edge Cases

Handles cases with empty or constant columns, NaNs, and class imbalance.
Provides fallback paths for scenarios where valid splits cannot be created.

Generated Code Dependencies

The generated code requires the following external libraries: pandas, numpy, sklearn. These dependencies are required by the generated code, not by this module.

Usage

Typically invoked by the knime2py emitter, this module is used in workflows that require cross-validation partitioning. An example of context access is:

train_df = context['{node_id}:1']

Node Identity

KNIME factory id: - FACTORY = "org.knime.base.node.meta.xvalidation.XValidatePartitionerFactory"

Special flags: - LOOP = "start": Indicates the beginning of a loop for cross-validation.

Configuration

The settings are defined in the XPartSettings dataclass, which includes: - k: Number of folds (default: 10). - random_sampling: Whether to shuffle data (default: False). - leave_one_out: Whether to use Leave-One-Out strategy (default: False). - stratified: Whether to use stratified sampling (default: True). - class_col: Column name for stratification (default: None). - use_seed: Whether to use a random seed (default: False). - seed: Random seed value (default: None).

The parse_xpart_settings function extracts these values from the settings.xml file using XPath queries, with appropriate fallbacks.

Limitations

Currently, this module does not support certain advanced KNIME features and approximates behavior in some cases.

References

For more information, refer to the KNIME documentation and the following URL: https://hub.knime.com/knime/extensions/org.knime.features.base/latest/ org.knime.base.node.meta.xvalidation.XValidatePartitionerFactory

`first(root, xpath)`

Return the first string value for xpath, stripped, or None.

If the xpath returns an element, prefer its @value, else its .text. If it returns a scalar (string/number), cast to str and strip.

Source code in src/knime2py/nodes/node_utils.py

def first(root: ET._Element, xpath: str) -> Optional[str]:
    """Return the first string value for xpath, stripped, or None.

    If the xpath returns an element, prefer its @value, else its .text.
    If it returns a scalar (string/number), cast to str and strip.
    """
    vals = root.xpath(xpath)
    if not vals:
        return None
    v = vals[0]
    # Element -> prefer @value then .text
    if isinstance(v, ET._Element):
        if v.get("value") is not None:
            return (v.get("value") or "").strip()
        return (v.text or "").strip()
    # Scalar / attribute string / number
    return (str(v) if v is not None else "").strip()

`first_el(root, xpath)`

Return the first Element for xpath, or None (ignores non-Elements).

Source code in src/knime2py/nodes/node_utils.py

def first_el(root: ET._Element, xpath: str) -> Optional[ET._Element]:
    """Return the first Element for xpath, or None (ignores non-Elements)."""
    vals = root.xpath(xpath)
    for v in vals:
        if isinstance(v, ET._Element):
            return v
    return None

`all_values(root, xpath)`

Return all values for xpath as stripped strings.

Source code in src/knime2py/nodes/node_utils.py

def all_values(root: ET._Element, xpath: str) -> List[str]:
    """Return all values for xpath as stripped strings."""
    return [(v or "").strip() for v in root.xpath(xpath)]

`iter_entries(root)`

Yield (key, value) pairs for all KNIME nodes.

Source code in src/knime2py/nodes/node_utils.py

def iter_entries(root: ET._Element):
    """Yield (key, value) pairs for all KNIME <entry key="..." value="..."/> nodes."""
    for ent in root.xpath(_ENTRY_XPATH):
        k = (ent.get("key") or "").strip()
        v = ent.get("value")
        yield k, (v or "").strip() if v is not None else None

`normalize_delim(raw)`

Normalize delimiter strings to their corresponding character representation.

Source code in src/knime2py/nodes/node_utils.py

def normalize_delim(raw: Optional[str]) -> Optional[str]:
    """Normalize delimiter strings to their corresponding character representation."""
    if raw is None:
        return None
    v = raw.strip()
    if len(v) == 1:
        return v
    up = v.upper()
    if up in {"TAB", "\\T", "CTRL-I"}:
        return "\t"
    if up in {"COMMA"}:
        return ","
    if up in {"SEMICOLON", "SEMI", "SC"}:
        return ";"
    if up in {"SPACE"}:
        return " "
    if up in {"PIPE"}:
        return "|"
    if v == "\\t":
        return "\t"
    return v or None

`normalize_char(raw)`

Normalize character strings to their corresponding single character representation.

Source code in src/knime2py/nodes/node_utils.py

def normalize_char(raw: Optional[str]) -> Optional[str]:
    """Normalize character strings to their corresponding single character representation."""
    if not raw:
        return None
    v = raw.strip()
    if v.upper() in {"", "NONE", "NULL"}:
        return None
    if v == "&quot;":
        return '"'
    if v == "&apos;":
        return "'"
    return v[:1] if len(v) >= 1 else None

`looks_like_path(s)`

Check if the given string looks like a file path.

Source code in src/knime2py/nodes/node_utils.py

def looks_like_path(s: str) -> bool:
    """Check if the given string looks like a file path."""
    if not s:
        return False
    low = s.lower()
    if low.startswith(("file:", "s3:", "hdfs:", "abfss:", "http://", "https://")):
        return True
    if s.endswith(".csv"):
        return True
    if "/" in s or "\\" in s:
        return True
    return False

`bool_from_value(v)`

Convert a string value to a boolean.

Source code in src/knime2py/nodes/node_utils.py

def bool_from_value(v: Optional[str]) -> Optional[bool]:
    """Convert a string value to a boolean."""
    if v is None:
        return None
    t = v.strip().lower()
    if t in {"true", "1", "yes", "y"}:
        return True
    if t in {"false", "0", "no", "n"}:
        return False
    return None

`normalize_in_ports(in_ports)`

Accepts items like ('1393','1') or '1393:1' (or just '1393') and returns a normalized list of (src_id, port) as strings.

Source code in src/knime2py/nodes/node_utils.py

def normalize_in_ports(in_ports: List[tuple[str, str]]) -> List[Tuple[str, str]]:
    """
    Accepts items like ('1393','1') or '1393:1' (or just '1393') and
    returns a normalized list of (src_id, port) as strings.
    """
    norm: List[Tuple[str, str]] = []
    for item in in_ports or []:
        if isinstance(item, tuple) and len(item) == 2:
            src, port = str(item[0]), str(item[1] or "1")
            norm.append((src, port))
        else:
            s = str(item)
            if ":" in s:
                src, port = s.split(":", 1)
                norm.append((src, port or "1"))
            elif s:
                norm.append((s, "1"))
    if not norm:
        norm.append(("UNKNOWN", "1"))
    return norm

`context_assignment_lines(node_id, out_ports)`

For reader-like nodes that produce a dataframe named df, publish it under context keys ':'.

Source code in src/knime2py/nodes/node_utils.py

def context_assignment_lines(node_id: str, out_ports: List[str]) -> List[str]:
    """
    For reader-like nodes that produce a dataframe named `df`,
    publish it under context keys '<node_id>:<port>'.
    """
    ports = sorted({(p or "1") for p in (out_ports or [])}) or ["1"]
    return [f"context['{node_id}:{p}'] = df" for p in ports]

`extract_csv_path(root)`

Prefer keys that sound like file paths; fall back to any entry value that looks like a path. Avoid false-positives like node_file='settings.xml' via looks_like_path().

Source code in src/knime2py/nodes/node_utils.py

def extract_csv_path(root: ET._Element) -> Optional[str]:
    """
    Prefer keys that *sound* like file paths; fall back to any entry value that looks like a path.
    Avoid false-positives like node_file='settings.xml' via looks_like_path().
    """
    # Prefer specific-ish keys first
    for pat in (r"\bpath\b", r"\burl\b", r"\bfile\b", r"location"):
        v = _first_value_re(root, pat)
        if v and looks_like_path(v):
            return v
    # Fallback: any entry value that looks like a CSV/path
    for _k, v in iter_entries(root):
        if v and looks_like_path(v):
            return v
    return None

`extract_csv_sep(root)`

Extract the CSV separator from the XML configuration.

Source code in src/knime2py/nodes/node_utils.py

def extract_csv_sep(root: ET._Element) -> Optional[str]:
    """Extract the CSV separator from the XML configuration."""
    raw = _first_value_re(root, r"(delim|separator|column[_-]?delimiter)\b")
    return normalize_delim(raw)

`extract_csv_quotechar(root)`

Extract the quote character used in the CSV configuration.

Source code in src/knime2py/nodes/node_utils.py

def extract_csv_quotechar(root: ET._Element) -> Optional[str]:
    """Extract the quote character used in the CSV configuration."""
    raw = _first_value_re_excluding(root, r"\bquote(_?char)?\b", r"escape")
    if raw is None:
        # looser fallback: any 'quote' key that isn't an escape
        for k, v in iter_entries(root):
            if "quote" in k.lower() and "escape" not in k.lower():
                raw = v
                break
    return normalize_char(raw)

`extract_csv_escapechar(root)`

Extract the escape character used in the CSV configuration.

Source code in src/knime2py/nodes/node_utils.py

def extract_csv_escapechar(root: ET._Element) -> Optional[str]:
    """Extract the escape character used in the CSV configuration."""
    raw = _first_value_re(root, r"escape")
    return normalize_char(raw)

`extract_csv_encoding(root)`

Extract the character encoding from the CSV configuration.

Source code in src/knime2py/nodes/node_utils.py

def extract_csv_encoding(root: ET._Element) -> Optional[str]:
    """Extract the character encoding from the CSV configuration."""
    return (
        _first_value_re(root, r"\bcharacter_set\b")
        or _first_value_re(root, r"\bcharset\b")
        or _first_value_re(root, r"encoding")
    )

`extract_csv_header_reader(root)`

Reader: look for 'column header', 'hasheader', or plain 'header', but avoid writer-only keys like 'write_header'.

Source code in src/knime2py/nodes/node_utils.py

def extract_csv_header_reader(root: ET._Element) -> Optional[bool]:
    """
    Reader: look for 'column header', 'hasheader', or plain 'header', but avoid writer-only
    keys like 'write_header'.
    """
    for k, v in iter_entries(root):
        lk = k.lower()
        if "header" not in lk:
            continue
        if "write" in lk:
            continue
        if "column" in lk or "hasheader" in lk or lk == "header":
            return bool_from_value(v)
    return None

`extract_csv_header_writer(root)`

Writer: prefer explicit 'writeColumnHeader'/'write_header'; otherwise any key whose name contains both 'write' and 'header'.

Source code in src/knime2py/nodes/node_utils.py

def extract_csv_header_writer(root: ET._Element) -> Optional[bool]:
    """
    Writer: prefer explicit 'writeColumnHeader'/'write_header'; otherwise any key whose
    name contains both 'write' and 'header'.
    """
    v = (
        _first_value_re(root, r"\bwriteColumnHeader\b")
        or _first_value_re(root, r"\bwrite_header\b")
        or _first_value_all_tokens(root, ["write", "header"])
    )
    return bool_from_value(v) if v is not None else None

`extract_csv_na_rep(root)`

Writer NA representation

modern: key='missing_value_pattern' (may be empty string '')
older: key contains both 'missing' and 'representation'

Keep empty string '' as a real value; return None only if not set.

Source code in src/knime2py/nodes/node_utils.py

def extract_csv_na_rep(root: ET._Element) -> Optional[str]:
    """
    Writer NA representation:
      - modern: key='missing_value_pattern' (may be empty string '')
      - older: key contains both 'missing' and 'representation'
    Keep empty string '' as a real value; return None only if not set.
    """
    v = _first_value_re(root, r"^missing_value_pattern$")
    if v is None:
        v = _first_value_all_tokens(root, ["missing", "representation"])
    return v

`extract_csv_include_index(root)`

Extract whether to include the index in the CSV output.

Source code in src/knime2py/nodes/node_utils.py

def extract_csv_include_index(root: ET._Element) -> Optional[bool]:
    """Extract whether to include the index in the CSV output."""
    raw = _first_value_re(root, r"include[_-]?index")
    return bool_from_value(raw)

`extract_table_spec_types(root)`

Return {column_name: java_class} from table_spec_config_Internals. Looks under .../individual_specs/*/ blocks.

Source code in src/knime2py/nodes/node_utils.py

def extract_table_spec_types(root: ET._Element) -> dict:
    """
    Return {column_name: java_class} from table_spec_config_Internals.
    Looks under .../individual_specs/*/<config key='0'..> blocks.
    """
    out = {}
    for cfg in root.xpath(
        ".//*[local-name()='config' and @key='table_spec_config_Internals']"
        "/*[local-name()='config' and @key='individual_specs']"
        "/*[local-name()='config']"  # per file block
        "/*[local-name()='config' and re:test(@key, '^[0-9]+$')]",
        namespaces={'re': "http://exslt.org/regular-expressions"}
    ):
        name = first(cfg, ".//*[local-name()='entry' and @key='name']/@value")
        jcls = first(cfg, ".//*[local-name()='config' and @key='type']"
                          "/*[local-name()='entry' and @key='class']/@value")
        if name:
            out[name] = jcls or ""
    return out

`java_to_pandas_dtype(java_class)`

Map KNIME java types to pandas nullable dtypes.

Source code in src/knime2py/nodes/node_utils.py

def java_to_pandas_dtype(java_class: str) -> Optional[str]:
    """
    Map KNIME java types to pandas nullable dtypes.
    """
    j = (java_class or "").lower()
    if "integer" in j or "long" in j or "intcell" in j:
        return "Int64"
    if "double" in j or "float" in j:
        return "Float64"
    if "boolean" in j:
        return "boolean"
    if "string" in j:
        return "string"
    # leave unknowns to inference
    return None

`collect_module_imports(mod_or_func)`

Return a sorted list of unique import lines from either

a module object that defines generate_imports()
a callable (e.g. the generate_imports function itself)

Source code in src/knime2py/nodes/node_utils.py

def collect_module_imports(mod_or_func: Optional[Union[object, Callable[[], Iterable[str]]]]) -> List[str]:
    """
    Return a sorted list of unique import lines from either:
      - a module object that defines generate_imports()
      - a callable (e.g. the generate_imports function itself)
    """
    imports = set()
    try:
        if mod_or_func is None:
            return []
        # If they passed the function directly
        if callable(mod_or_func):
            result = mod_or_func()
            items = _coerce_iterable(result)
        else:
            gi = getattr(mod_or_func, "generate_imports", None)
            if callable(gi):
                result = gi()
                items = _coerce_iterable(result)
            else:
                items = []
        for line in items:
            s = (line or "").strip()
            if s:
                imports.add(s)
    except Exception:
        # don’t let import gathering crash codegen
        return []
    return sorted(imports)

`split_out_imports(lines)`

Return (found_imports, body_without_imports). Any line that begins with 'import ' or 'from ' (ignoring leading spaces) is treated as an import.

Source code in src/knime2py/nodes/node_utils.py

def split_out_imports(lines: List[str]) -> tuple[List[str], List[str]]:
    """
    Return (found_imports, body_without_imports).
    Any line that begins with 'import ' or 'from ' (ignoring leading spaces) is treated as an import.
    """
    found: List[str] = []
    body: List[str] = []
    for ln in lines or []:
        s = ln.lstrip()
        if s.startswith("import ") or s.startswith("from "):
            found.append(s.strip())
        else:
            body.append(ln)
    return found, body

`resolve_reader_path(root, node_dir)`

Resolve the path from settings.xml. Supports: - LOCAL: absolute path is used as-is - RELATIVE + knime.workflow: path is relative to the workflow directory

Source code in src/knime2py/nodes/node_utils.py

def resolve_reader_path(root: ET._Element, node_dir: Path) -> Optional[str]:
    """
    Resolve the path from settings.xml. Supports:
      - LOCAL: absolute path is used as-is
      - RELATIVE + knime.workflow: path is relative to the workflow directory
    """
    path_cfg = first_el(root, ".//*[local-name()='config' and @key='path']")
    if path_cfg is None:
        return None

    raw_path = first(path_cfg, ".//*[local-name()='entry' and @key='path']/@value")
    fs_type  = first(path_cfg, ".//*[local-name()='entry' and @key='file_system_type']/@value")
    spec     = first(path_cfg, ".//*[local-name()='entry' and @key='file_system_specifier']/@value")

    if not raw_path:
        return None

    node_has_settings = (node_dir / "settings.xml").exists()
    workflow_dir = node_dir.parent if node_has_settings else node_dir

    try:
        p = Path(raw_path)
        if (fs_type or "").upper() == "LOCAL" or p.is_absolute():
            return str(p.expanduser().resolve())

        if (fs_type or "").upper() == "RELATIVE" and (spec or "").lower() == "knime.workflow":
            return str((workflow_dir / raw_path).expanduser().resolve())

        # Fallback: treat as relative to workflow_dir
        return str((workflow_dir / raw_path).expanduser().resolve())
    except Exception:
        # Last-ditch: just return the raw string
        return raw_path

`parse_xpart_settings(node_dir)`

Parse the settings from the settings.xml file.

Parameters:

Name	Type	Description	Default
`node_dir`	`Optional[Path]`	The directory containing the settings.xml file.	required

Returns:

Name	Type	Description
`XPartSettings`	`XPartSettings`	An instance of XPartSettings populated with values from the settings.xml.

Source code in src/knime2py/nodes/x_partitioner.py

def parse_xpart_settings(node_dir: Optional[Path]) -> XPartSettings:
    """
    Parse the settings from the settings.xml file.

    Args:
        node_dir (Optional[Path]): The directory containing the settings.xml file.

    Returns:
        XPartSettings: An instance of XPartSettings populated with values from the settings.xml.
    """
    if not node_dir:
        return XPartSettings()

    sp = node_dir / "settings.xml"
    if not sp.exists():
        return XPartSettings()

    root = ET.parse(str(sp), parser=XML_PARSER).getroot()
    model = first_el(root, ".//*[local-name()='config' and @key='model']")

    k = _to_int(first(model, ".//*[local-name()='entry' and @key='validations']/@value"), 10) if model is not None else 10
    rnd = _bool(first(model, ".//*[local-name()='entry' and @key='randomSampling']/@value"), False) if model is not None else False
    loo = _bool(first(model, ".//*[local-name()='entry' and @key='leaveOneOut']/@value"), False) if model is not None else False
    strat = _bool(first(model, ".//*[local-name()='entry' and @key='stratifiedSampling']/@value"), True) if model is not None else True
    cls = first(model, ".//*[local-name()='entry' and @key='classColumn']/@value") if model is not None else None
    use_seed = _bool(first(model, ".//*[local-name()='entry' and @key='useRandomSeed']/@value"), False) if model is not None else False
    seed = _to_long(first(model, ".//*[local-name()='entry' and @key='randomSeed']/@value"), 0) if (model is not None and use_seed) else None

    if not loo and k < 2:
        k = 2

    return XPartSettings(
        k=k,
        random_sampling=rnd,
        leave_one_out=loo,
        stratified=strat,
        class_col=(cls or None),
        use_seed=use_seed,
        seed=seed if rnd else None,  # seed used only when shuffle/randomSampling is True
    )

`generate_imports()`

Generate the necessary import statements for the code.

Returns:

Type	Description
	list[str]: A list of import statements.

Source code in src/knime2py/nodes/x_partitioner.py

def generate_imports():
    """
    Generate the necessary import statements for the code.

    Returns:
        list[str]: A list of import statements.
    """
    return [
        "import pandas as pd",
        "import numpy as np",
        "from sklearn.model_selection import KFold, StratifiedKFold, LeaveOneOut",
    ]

`generate_py_body(node_id, node_dir, in_ports, out_ports=None)`

Generate the Python code body for the node.

Parameters:

Name	Type	Description	Default
`node_id`	`str`	The unique identifier for the node.	required
`node_dir`	`Optional[str]`	The directory of the node.	required
`in_ports`	`List[object]`	The incoming ports for the node.	required
`out_ports`	`Optional[List[str]]`	The outgoing ports for the node.	`None`

Returns:

Type	Description
`List[str]`	List[str]: A list of code lines for the node's body.

Source code in src/knime2py/nodes/x_partitioner.py

def generate_py_body(
    node_id: str,
    node_dir: Optional[str],
    in_ports: List[tuple[str, str]],
    out_ports: Optional[List[str]] = None,
) -> List[str]:
    """
    Generate the Python code body for the node.

    Args:
        node_id (str): The unique identifier for the node.
        node_dir (Optional[str]): The directory of the node.
        in_ports (List[object]): The incoming ports for the node.
        out_ports (Optional[List[str]]): The outgoing ports for the node.

    Returns:
        List[str]: A list of code lines for the node's body.
    """
    ndir = Path(node_dir) if node_dir else None
    cfg = parse_xpart_settings(ndir)

    pairs = normalize_in_ports(in_ports)
    src_id, in_port = pairs[0]

    lines: List[str] = []
    lines.append(f"# {HUB_URL}")
    # Freeze the loop source DF in a unique variable that won't be clobbered by inner nodes
    src_var = f"__xpart_src_{node_id}"
    lines.append(f"{src_var} = context['{src_id}:{in_port}']  # input table (frozen for the whole CV loop)")

    # Emit loop body using the frozen variable
    lines.extend(_emit_xpart_code(cfg, node_id, src_var))
    return lines

`get_name()`

Return name of the node in KNIME workflow.

Source code in src/knime2py/nodes/x_partitioner.py

def get_name() -> str:
    """Return name of the node in KNIME workflow."""
    return "X_Partitioner"

`handle(ntype, nid, npath, incoming, outgoing)`

Handle the node and return the necessary imports and body lines.

Parameters:

Name	Type	Description	Default
`ntype`		The type of the node.	required
`nid`		The unique identifier for the node.	required
`npath`		The path to the node.	required
`incoming`		The incoming connections to the node.	required
`outgoing`		The outgoing connections from the node.	required

Returns:

Name	Type	Description
`tuple`		A tuple containing the imports and body lines if this module can handle the node; else None.

Source code in src/knime2py/nodes/x_partitioner.py

def handle(ntype, nid, npath, incoming, outgoing):
    """
    Handle the node and return the necessary imports and body lines.

    Args:
        ntype: The type of the node.
        nid: The unique identifier for the node.
        npath: The path to the node.
        incoming: The incoming connections to the node.
        outgoing: The outgoing connections from the node.

    Returns:
        tuple: A tuple containing the imports and body lines if this module can handle the node; else None.
    """
    explicit_imports = collect_module_imports(generate_imports)

    in_ports  = [(src_id, str(getattr(e, "source_port", "") or "1")) for src_id, e in (incoming or [])]
    out_ports = [str(getattr(e, "source_port", "") or "1") for _, e in (outgoing or [])] or ["1", "2"]

    node_lines = generate_py_body(nid, npath, in_ports, out_ports)
    found_imports, body = split_out_imports(node_lines)

    imports = sorted(set(explicit_imports) | set(found_imports))
    return imports, body