| """Constraint checks shared by beam search and deterministic verification.""" |
|
|
| from __future__ import annotations |
|
|
| from graph.graph import GraphStore |
| from graph.types import ConstraintState, GraphPath |
|
|
|
|
| def path_satisfies_constraints( |
| path: GraphPath, constraints: ConstraintState, graph: GraphStore |
| ) -> bool: |
| hops = path.hop_count |
| if hops < constraints.min_hops or hops > constraints.max_hops: |
| return False |
| if constraints.exact_hops is not None and hops != constraints.exact_hops: |
| return False |
| relations = path.relation_sequence |
| if ( |
| constraints.required_relation_sequence |
| and relations != constraints.required_relation_sequence |
| ): |
| return False |
| if constraints.allowed_relations and not set(relations).issubset( |
| constraints.allowed_relations |
| ): |
| return False |
| if constraints.required_relations and not constraints.required_relations.issubset( |
| set(relations) |
| ): |
| return False |
| if constraints.forbidden_relations.intersection(relations): |
| return False |
| if constraints.require_unique_nodes: |
| nodes = ( |
| [path.steps[0].source_id, *(step.target_id for step in path.steps)] |
| if path.steps |
| else [] |
| ) |
| if len(nodes) != len(set(nodes)): |
| return False |
| intermediate_types = { |
| step.target_type for step in path.steps[:-1] if step.target_type is not None |
| } |
| if ( |
| constraints.required_intermediate_types |
| and not constraints.required_intermediate_types.issubset(intermediate_types) |
| ): |
| return False |
| if path.steps: |
| endpoint_type = path.steps[-1].target_type or graph.nodes.get( |
| path.endpoint_id or "" |
| ) |
| if ( |
| constraints.required_types |
| and endpoint_type not in constraints.required_types |
| ): |
| return False |
| if endpoint_type in constraints.forbidden_types: |
| return False |
| return True |
|
|
