cursus.steps.scripts.label_ruleset_generation

Ruleset Generation Script

Validates and optimizes user-defined classification rules following the cursus framework pattern for transparent, maintainable rule-based classification.

class ValidationResult(valid=True)

Bases: object

Result of a validation check.

to_dict()

Convert to dictionary for JSON serialization.

class RulesetLabelValidator

Bases: object

Validates output labels match configuration (extended for multilabel).

validate_labels(ruleset)

Validates all output_label values in rules. Extended to support multilabel mode.

Parameters:

ruleset (dict) – Input ruleset configuration

Returns:

ValidationResult with label validation status

Return type:

ValidationResult

class RuleCoverageValidator

Bases: object

Validates that all label columns have at least one rule.

validate(label_config, rules)

Validates rule coverage for all label columns.

Checks: - Each label column has at least one rule targeting it - Warns about orphan label columns

Parameters:

• label_config (dict) – Label configuration

• rules (List[dict]) – List of rule definitions

Returns:

ValidationResult with coverage validation status

Return type:

ValidationResult

class RulesetLogicValidator

Bases: object

Validates rule logic for errors.

validate_logic(ruleset)

Validates rule logic for common errors.

Parameters:

ruleset (dict) – Input ruleset configuration

Returns:

ValidationResult with logic validation status

Return type:

ValidationResult

calculate_complexity(condition)

Calculate complexity score for a condition.

Parameters:

condition (dict) – Condition expression

Returns:

Complexity score (higher = more complex)

Return type:

int

extract_all_fields(condition)

Recursively extract all field names from a condition.

Parameters:

condition (dict) – Condition expression (may be nested)

Returns:

List of unique field names used

Return type:

List[str]

extract_fields_and_values(condition)

Recursively extract field names and their used values from conditions.

Parameters:

condition (dict) – Condition expression (may be nested)

Returns:

Dictionary mapping field names to list of values seen in conditions

Return type:

Dict[str, List[Any]]

infer_field_type(values)

Infer field type from values used in conditions.

Parameters:

values (List[Any]) – List of values seen for a field

Returns:

‘string’, ‘int’, ‘float’, or ‘bool’

Return type:

Inferred type

infer_field_config_from_rules(rules, log=<built-in function print>)

Infer complete field configuration from rule definitions.

Analyzes all rules to extract: - Field names used - Field types based on values - Field usage statistics

Parameters:

• rules (List[dict]) – List of rule definitions

• log (Callable[[str], None]) – Logging function

Returns:

{

“required_fields”: [], # Empty when inferred “optional_fields”: […], # All discovered fields “field_types”: {…} # Inferred types

}

Return type:

Complete field_config dictionary with structure

analyze_field_usage(rules)

Analyze which fields are used most frequently across rules.

Parameters:

rules (List[dict]) – List of rule definitions

Returns:

Dictionary mapping field names to usage count

Return type:

Dict[str, int]

optimize_ruleset(ruleset, enable_complexity=True, enable_field_grouping=False, log=<built-in function print>)

Optimize ruleset using multiple strategies.

Parameters:

• ruleset (dict) – Input ruleset with unoptimized rules

• enable_complexity (bool) – Enable complexity-based ordering

• enable_field_grouping (bool) – Enable field usage grouping

• log (Callable[[str], None]) – Logging function

Returns:

Optimized ruleset with reordered rules

Return type:

dict

main(input_paths, output_paths, environ_vars, job_args=None, logger=None)

Main logic for ruleset generation and validation.

Parameters:

• input_paths (Dict[str, str]) – Dictionary with input paths

• output_paths (Dict[str, str]) – Dictionary with output paths

• environ_vars (Dict[str, str]) – Environment variables

• job_args (Namespace | None) – Command line arguments

• logger (Callable[[str], None] | None) – Optional logger function

Returns:

Dictionary with processing results

Return type:

Dict[str, Any]