missing_value

knime2py.nodes.missing_value

Missing Value Handler.

Overview

This module generates Python code to handle missing values in a DataFrame based on KNIME's Missing Value policies. It fits into the knime2py generator pipeline by producing code that applies specified fill strategies to input tables and writes the results to the node's context.

Runtime Behavior

Inputs: - Reads a DataFrame from the context using the key format 'src_id:in_port'.

Outputs: - Writes the processed DataFrame back to the context with the key format 'node_id:out_port', where out_port defaults to '1'.

Key algorithms or mappings: - Implements fill strategies such as mean, median, mode, forward fill, backward fill, and fixed value fills based on the configuration parsed from settings.xml.

Edge Cases

• Handles empty or constant columns by skipping them.
• Safeguards against NaN values and class imbalance by providing fallback strategies.

Generated Code Dependencies

The generated code requires the following external libraries: - pandas

These dependencies are required by the generated code, not by this module.

Usage

Typically invoked by upstream KNIME nodes that require missing value handling. Example context access:

df = context['input_table:1']

Node Identity

KNIME factory id: - FACTORY = "org.knime.base.node.preproc.pmml.missingval.compute.MissingValueHandlerNodeFactory"

Configuration

Settings are defined in the MissingValueSettings dataclass, which includes: - by_dtype: List of TypePolicy instances defining fill strategies per data type.

The parse_missing_value_settings function extracts these values from the settings.xml file using XPath queries.

Limitations

Currently, this module does not support all KNIME fill strategies and may approximate 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.preproc.pmml.missingval.compute.MissingValueHandlerNodeFactory

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

strip_rule_quotes(s)

Remove surrounding quotes from a KNIME rule token when present.

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

def strip_rule_quotes(s: str) -> str:
    """Remove surrounding quotes from a KNIME rule token when present."""
    return s[1:-1] if (len(s) >= 2 and s[0] == s[-1] and s[0] in "'\"") else s

parse_knime_rule(line)

Parse the simple KNIME Rule Engine subset shared by rule nodes.

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

def parse_knime_rule(line: str) -> Optional[Rule]:
    """Parse the simple KNIME Rule Engine subset shared by rule nodes."""
    s = html.unescape(line or "").strip()
    if not s or _RULE_COMMENT.match(s):
        return None
    m = _RULE_TRUE.match(s)
    if m:
        return Rule(kind="true", col=None, op=None, value=None, outcome=strip_rule_quotes(m.group("out")))
    m = _RULE_COMPARE.match(s)
    if m:
        return Rule(
            kind="compare",
            col=m.group("col").strip(),
            op=m.group("op"),
            value=strip_rule_quotes(m.group("val").strip()),
            outcome=strip_rule_quotes(m.group("out")),
        )
    m = _RULE_LIKE.match(s)
    if m:
        return Rule(
            kind="like",
            col=m.group("col").strip(),
            op=None,
            value=m.group("pat"),
            outcome=strip_rule_quotes(m.group("out")),
        )
    return None

parse_knime_rules_from_config(rules_cfg)

Parse numbered entries from a KNIME block.

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

def parse_knime_rules_from_config(rules_cfg: Optional[ET._Element]) -> List[Rule]:
    """Parse numbered entries from a KNIME <config key='rules'> block."""
    if rules_cfg is None:
        return []
    numbered: List[tuple[int, str]] = []
    for k, v in iter_entries(rules_cfg):
        if k.isdigit() and v is not None:
            numbered.append((int(k), v))
    rules: List[Rule] = []
    for _, raw in sorted(numbered, key=lambda item: item[0]):
        rule = parse_knime_rule(raw)
        if rule:
            rules.append(rule)
    return rules

rule_literal_py(val)

Convert a KNIME rule literal token to generated Python source.

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

def rule_literal_py(val: str) -> str:
    """Convert a KNIME rule literal token to generated Python source."""
    s = str(val).strip()
    if re.fullmatch(r"[+-]?\d+", s):
        return s
    if re.fullmatch(r"[+-]?(\d+\.\d*|\.\d+|\d+)([eE][+-]?\d+)?", s):
        return s
    return repr(s)

rule_wildcard_to_regex(pat)

Convert KNIME Rule Engine LIKE wildcard syntax to a regex.

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

def rule_wildcard_to_regex(pat: str) -> str:
    """Convert KNIME Rule Engine LIKE wildcard syntax to a regex."""
    esc = re.escape(pat)
    return "^" + esc.replace(r"\*", ".*") + "$"

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 == """:
        return '"'
    if v == "'":
        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_missing_value_settings(node_dir)

Parse the missing value settings from the settings.xml file.

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

def parse_missing_value_settings(node_dir: Optional[Path]) -> MissingValueSettings:
    """Parse the missing value settings from the settings.xml file."""
    if not node_dir:
        return MissingValueSettings()
    sp = node_dir / "settings.xml"
    if not sp.exists():
        return MissingValueSettings()

    root = ET.parse(str(sp), parser=XML_PARSER).getroot()
    model_cfgs = root.xpath(".//*[local-name()='config' and @key='model']")
    if not model_cfgs:
        return MissingValueSettings()
    model_cfg = model_cfgs[0]

    # Column overrides
    by_column: List[ColumnPolicy] = []
    column_sections = model_cfg.xpath("./*[local-name()='config' and @key='columnSettings']")
    if column_sections:
        for cfg in column_sections[0].xpath("./*[local-name()='config']"):
            names_cfg = first_el(cfg, "./*[local-name()='config' and @key='colNames']")
            columns: List[str] = []
            if names_cfg is not None:
                for key, value in iter_entries(names_cfg):
                    if key.isdigit() and value:
                        columns.append(value)
            if not columns:
                continue

            settings_cfg = first_el(cfg, "./*[local-name()='config' and @key='settings']")
            if settings_cfg is None:
                continue
            factory_id = None
            for key, value in iter_entries(settings_cfg):
                if key == "factoryID":
                    factory_id = value
                    break
            inner_settings = first_el(settings_cfg, "./*[local-name()='config' and @key='settings']")
            fixed_val = _first_present_value(inner_settings) if inner_settings is not None else None
            strategy = _strategy_from_factory(factory_id or "")
            dtype = _dtype_from_factory(factory_id)
            for col in columns:
                by_column.append(
                    ColumnPolicy(
                        column=col,
                        strategy=strategy,
                        value=fixed_val,
                        dtype=dtype,
                    )
                )

    dts = model_cfg.xpath("./*[local-name()='config' and @key='dataTypeSettings']")
    if not dts:
        return MissingValueSettings(by_column=by_column)

    by_dtype: List[TypePolicy] = []
    for cfg in dts[0].xpath("./*[local-name()='config']"):
        cell_cls = (cfg.get("key") or "").strip()
        dtype = _CELL_TO_DTYPE.get(cell_cls)
        if not dtype:
            continue

        factory_id = None
        for k, v in iter_entries(cfg):
            if k == "factoryID":
                factory_id = v
                break
        if not factory_id:
            continue

        strategy = _strategy_from_factory(factory_id)

        fixed_val = None
        for sub in cfg.xpath("./*[local-name()='config' and @key='settings']"):
            fixed_val = _first_present_value(sub)
            if fixed_val is not None:
                break

        by_dtype.append(TypePolicy(dtype=dtype, strategy=strategy, value=fixed_val))

    return MissingValueSettings(by_dtype=by_dtype, by_column=by_column)

generate_imports()

Generate the necessary import statements for the output code.

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

def generate_imports():
    """Generate the necessary import statements for the output code."""
    return ["import pandas as pd"]

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

Generate the Python code body for the node based on its configuration and input ports.

Source code in src/knime2py/nodes/missing_value.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 based on its configuration and input ports."""
    ndir = Path(node_dir) if node_dir else None
    settings = parse_missing_value_settings(ndir)

    lines: List[str] = []
    lines.append(f"# {HUB_URL}")

    pairs = normalize_in_ports(in_ports)
    src_id, in_port = pairs[0]
    lines.append(f"df = context['{src_id}:{in_port}']  # input table")
    lines.extend(MV_HELPER_LINES)

    lines.extend(_emit_fill_code(settings))
    dtype_literal = [
        {"dtype": pol.dtype, "strategy": pol.strategy, "value": pol.value}
        for pol in settings.by_dtype
    ]
    col_literal = [
        {
            "column": pol.column,
            "dtype": pol.dtype,
            "strategy": pol.strategy,
            "value": pol.value,
        }
        for pol in settings.by_column
    ]
    lines.append(f"dtype_policies = {dtype_literal!r}")
    lines.append(f"column_policies = {col_literal!r}")
    lines.append("model_bundle = _mv_collect_bundle(df, column_policies, dtype_policies)")
    lines.append("model_bundle['strategies'] = dtype_policies")
    lines.append("model_bundle['column_strategies'] = column_policies")

    ports = out_ports or ["1"]
    port_map = {"1": "out_df", "2": "model_bundle"}
    for p in sorted({(p or '1') for p in ports}):
        target = port_map.get(p, "out_df")
        lines.append(f"context['{node_id}:{p}'] = {target}")
    return lines

get_name()

Return name of the node in KNIME workflow.

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

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

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

Handle the node processing, generating imports and body code.

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

def handle(ntype, nid, npath, incoming, outgoing):
    """Handle the node processing, generating imports and body code."""
    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"]
    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