Skip to content

Weight Computation Utilities

Metadata-driven weight computation for aggregation operations in workflows.

This module provides functions for computing weights from metadata using a weight specification. Weights are computed as the product of matching field-value weights, with a default of 1.0 for unspecified values.

Core Functions

compute_weight 

compute_weight(
    metadata: Dict[str, Any], weight_spec: Dict[str, Dict[str, float]]
) -> float

Compute a weight from metadata using a weight specification.

The weight is computed as the product of all matching field-value weights. For fields in the weight specification that are not present in the metadata, or for values not in the weight specification, a default weight of 1.0 is used.

Parameters:

  • metadata

    (Dict[str, Any]) –

    Dictionary of metadata field names to values.

  • weight_spec

    (Dict[str, Dict[str, float]]) –

    Mapping from metadata field to value-weight pairs. Each field maps to a dictionary of value-to-weight mappings.

Returns:

  • float –

    The computed weight (product of all matching weights).

Example

weight_spec = { ... "action": {"generate_graph": 1.0, "migrate_trace": 0.5}, ... "algorithm": {"pc": 1.0, "fci": 0.8}, ... } compute_weight({"action": "migrate_trace", "algorithm": "fci"}, ... weight_spec) 0.4 compute_weight({"action": "generate_graph"}, weight_spec) 1.0 compute_weight({"other": "value"}, weight_spec) 1.0

validate_weight_spec 

validate_weight_spec(weight_spec: Dict[str, Dict[str, float]]) -> None

Validate a weight specification.

Checks that the weight specification is well-formed: - Must be a dictionary - Keys must be strings (metadata field names) - Values must be dictionaries mapping field values to weights - Weights must be non-negative numbers

Parameters:

  • weight_spec

    (Dict[str, Dict[str, float]]) –

    Mapping from metadata field to value-weight pairs.

Raises:

Example

validate_weight_spec({"action": {"pc": 1.0, "fci": 0.5}}) validate_weight_spec({"action": {"pc": -1.0}}) Traceback (most recent call last): ... WeightSpecError: Weight for 'action.pc' must be non-negative, got -1.0

get_weight_fields 

get_weight_fields(weight_spec: Dict[str, Dict[str, float]]) -> set

Get the set of metadata field names used in a weight specification.

Useful for validating that required metadata fields are present.

Parameters:

  • weight_spec

    (Dict[str, Dict[str, float]]) –

    Mapping from metadata field to value-weight pairs.

Returns:

  • set –

    Set of field names referenced in the weight specification.

Example

weight_spec = { ... "action": {"generate_graph": 1.0}, ... "algorithm": {"pc": 1.0}, ... } get_weight_fields(weight_spec)

Exceptions

WeightSpecError

Exception raised for invalid weight specifications.

Weight Specification Format

Weight specifications are dictionaries mapping metadata field names to value-weight pairs:

weight_spec = {
    "field_name": {
        "value1": 1.0,
        "value2": 0.5,
    },
    "another_field": {
        "valueA": 0.8,
        "valueB": 1.2,
    },
}

Weight Calculation

The final weight is the product of all matching field-value weights:

  • If a metadata field is in the weight spec and its value matches, that weight is multiplied into the result
  • If a field is not in metadata, or value is not in the spec, weight 1.0 is used (no effect on product)

Usage Examples

Basic Weight Computation

from causaliq_core.utils import compute_weight

weight_spec = {
    "action": {"generate_graph": 1.0, "migrate_trace": 0.5},
    "algorithm": {"pc": 1.0, "fci": 0.8},
}

# Both fields match
metadata = {"action": "migrate_trace", "algorithm": "fci"}
compute_weight(metadata, weight_spec)  # 0.5 * 0.8 = 0.4

# Only action matches
metadata = {"action": "migrate_trace", "algorithm": "ges"}
compute_weight(metadata, weight_spec)  # 0.5 * 1.0 = 0.5

# No matches - default weight
metadata = {"network": "asia"}
compute_weight(metadata, weight_spec)  # 1.0

Validating Weight Specifications

from causaliq_core.utils import validate_weight_spec, WeightSpecError

# Valid specification
validate_weight_spec({"action": {"pc": 1.0, "fci": 0.5}})  # OK

# Invalid - negative weight
try:
    validate_weight_spec({"action": {"pc": -1.0}})
except WeightSpecError as e:
    print(f"Invalid: {e}")

Getting Weight Fields

from causaliq_core.utils import get_weight_fields

weight_spec = {
    "action": {"generate_graph": 1.0},
    "algorithm": {"pc": 1.0},
}

fields = get_weight_fields(weight_spec)
print(fields)  # {'action', 'algorithm'}

Workflow Integration

Weight specifications are used in workflow configurations for aggregation:

actions:
  merge_graphs:
    input: discovery_results.db
    weights:
      action:
        generate_graph: 1.0
        migrate_trace: 0.5
      algorithm:
        pc: 1.0
        fci: 0.8
    output: merged_graphs.db

Entries are weighted during aggregation based on their metadata values. This enables flexible weighting by source action, algorithm, sample size, or any other metadata field.