3. Build a Reusable Entity Query Helper Module

Why a Query Helper Module

Cleanup scripts fail in predictable ways when query logic is copy-pasted. One script filters components by cardimage, another forgets to exclude /^HM_/ collector names, a third counts elements with a different mask than the one it deletes with. The result is inconsistent counts, silent mismatches between "found" and "deleted", and audit logs that cannot be trusted. A reusable helper module fixes this by centralizing three concerns: filtering (which entities match a predicate), counting (how many, grouped by a key), and reporting (a stable, structured record of what was queried).

This case builds hm_query.py, a small module that wraps the HyperMesh Python API entity collections behind a uniform interface. It does not perform cleanup itself; it is the substrate that later cleanup scripts (empty-component detection, deletion with safety checks, load audits) will import. The design goal is that every query returns the same shape of result, so downstream code never has to branch on entity type.

Design Contract

The module exposes one primary entry point, query(model, entity_type, predicate=None, name_pattern=None, limit=None), plus three utilities: count_by(model, entity_type, key), summarize(results), and report(results, path). The contract is deliberately narrow:

• Entity type is a string key resolved through a registry, so callers pass "components" or "elements" rather than reaching into API internals.
• Predicate is a callable receiving a single entity and returning True/False. It is applied after the type is resolved, so predicates stay type-agnostic where possible.
• Name pattern is a compiled regular expression matched against the entity's name attribute; it is applied before the predicate to keep predicate cost low on large models.
• Result is a QueryResult dataclass carrying entity_type, entities (list), total_scanned, matched, and elapsed_ms.

Keeping total_scanned alongside matched is the single most useful diagnostic: a query that matches zero entities out of zero scanned means the type resolution failed, while zero out of 40,000 means the filter is too strict. Those are different bugs and the result object distinguishes them.

Implementation

The module resolves entity collections through a registry that maps logical names to accessor functions. This isolates the one place where API specifics live, so an API change touches a single dictionary rather than every script.

# hm_query.py
import re
import time
from dataclasses import dataclass, field
from typing import Any, Callable, Iterable, Optional

@dataclass
class QueryResult:
    entity_type: str
    entities: list = field(default_factory=list)
    total_scanned: int = 0
    elapsed_ms: float = 0.0

    @property
    def matched(self) -> int:
        return len(self.entities)

    def __bool__(self):
        return self.matched > 0

# Registry: logical name -> accessor taking the HyperMesh model object.
# Accessors are lambdas so the model is bound at call time, not import time.
_REGISTRY = {
    "components": lambda m: m.components,
    "properties": lambda m: m.properties,
    "materials":  lambda m: m.materials,
    "elements":   lambda m: m.elements,
    "nodes":      lambda m: m.nodes,
    "loads":      lambda m: m.loads,
    "loadsteps":  lambda m: m.loadsteps,
}

def _resolve(model, entity_type: str) -> Iterable:
    try:
        accessor = _REGISTRY[entity_type]
    except KeyError:
        raise ValueError(
            f"Unknown entity type {entity_type!r}; "
            f"known: {sorted(_REGISTRY)}"
        ) from None
    return accessor(model)

def query(model, entity_type, predicate=None, name_pattern=None, limit=None):
    """Return a QueryResult for entities of entity_type matching the filters."""
    start = time.perf_counter()
    collection = _resolve(model, entity_type)
    pattern = re.compile(name_pattern) if name_pattern else None

    result = QueryResult(entity_type=entity_type)
    for entity in collection:
        result.total_scanned += 1
        if pattern is not None and not pattern.search(getattr(entity, "name", "") or ""):
            continue
        if predicate is not None and not predicate(entity):
            continue
        result.entities.append(entity)
        if limit is not None and result.matched >= limit:
            break

    result.elapsed_ms = (time.perf_counter() - start) * 1000.0
    return result

Two details matter for correctness. First, getattr(entity, "name", "") or "" guards against entities whose name is None; without it, re.search raises TypeError on unnamed collectors. Second, limit is checked after appending, so limit=1 returns exactly one match rather than zero — an off-by-one that is easy to introduce and hard to notice in a count-only test.

Counting and Reporting

Counting is where duplicated logic usually diverges. count_by takes a key function and returns an ordered mapping, so callers get deterministic output for logs and diffs.

from collections import Counter

def count_by(model, entity_type, key, predicate=None, name_pattern=None):
    """Group matched entities by key(entity) and return a sorted dict."""
    res = query(model, entity_type, predicate=predicate, name_pattern=name_pattern)
    counts = Counter(key(e) for e in res.entities)
    return dict(sorted(counts.items(), key=lambda kv: (-kv[1], str(kv[0]))))

Sorting by descending count then by key string makes reports stable across runs and across machines, which matters when a cleanup log is diffed against a previous baseline. summarize and report build on the same result object:

import json

def summarize(results):
    """results: iterable of QueryResult -> list of plain dicts."""
    return [
        {
            "entity_type": r.entity_type,
            "scanned": r.total_scanned,
            "matched": r.matched,
            "elapsed_ms": round(r.elapsed_ms, 2),
        }
        for r in results
    ]

def report(results, path):
    payload = summarize(results)
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(payload, fh, indent=2)
    return payload

Query Pipeline and Result Shape

The diagram below shows how a single call flows from logical type name through the registry, the name-pattern gate, the predicate gate, and into the result object. The two gates are ordered deliberately: the cheap regex runs first, so an expensive predicate never sees entities that cannot match by name.

Using the Module Against a Live Model

The following driver exercises the module on a session and prints a report. It assumes hm is the HyperMesh Python module and model is the current model object, consistent with the session baseline established earlier in the course.

import hm
import hm_query as q

model = hm.Model()

# 1. All components whose name starts with "BODY_"
body = q.query(model, "components", name_pattern=r"^BODY_")

# 2. Components with no elements attached (predicate only)
empty = q.query(
    model, "components",
    predicate=lambda c: len(c.elements) == 0,
)

# 3. Element count grouped by property cardimage
by_card = q.count_by(
    model, "elements",
    key=lambda e: getattr(e.property, "cardimage", "UNKNOWN"),
)

# 4. Persist a combined report
payload = q.report([body, empty], "query_report.json")
print(payload)

Expected observable output for a model with 1,240 components, 38,000 elements, and 12 empty components would resemble the following — these numbers are illustrative, not measured:

[
  {"entity_type": "components", "scanned": 1240, "matched": 87,  "elapsed_ms": 3.1},
  {"entity_type": "components", "scanned": 1240, "matched": 12,  "elapsed_ms": 2.4}
]

Verification and Failure Symptoms

Verify the module before trusting it in cleanup scripts. A minimal check suite catches the common defects:

Deliverable: Module Interface Card

The following card documents the module's public surface for reuse across the cleanup suite. It is the artifact later scripts reference instead of re-reading the source.

With this module in place, subsequent cleanup routines reduce to a predicate and a key function. The query mechanics, the scan accounting, and the report format stay fixed — which is exactly what makes the later deletion and audit scripts comparable to one another.