Get neighbors via a relation. Returns (new_variable, "Observation: ...").
(self, variable: Union[Variable, str], relation: str)
| 88 | return None, rtn_str |
| 89 | |
| 90 | def get_neighbors(self, variable: Union[Variable, str], relation: str): |
| 91 | """ |
| 92 | Get neighbors via a relation. Returns (new_variable, "Observation: ..."). |
| 93 | """ |
| 94 | if not isinstance(variable, Variable): |
| 95 | if not re.match(r'^([mf])\.[\w_]+$', variable): |
| 96 | raise ValueError("get_neighbors: variable must be a variable or an entity") |
| 97 | |
| 98 | cache_key = (self.task_id, variable if isinstance(variable, str) else hash(variable)) |
| 99 | if cache_key not in relation_cache and variable not in variable_relations_cache: |
| 100 | self.get_relations(variable) |
| 101 | |
| 102 | relations_to_check = variable_relations_cache.get(variable, []) |
| 103 | if relation not in relations_to_check: |
| 104 | raise ValueError("get_neighbors: relation must be a relation of the variable") |
| 105 | |
| 106 | rtn_str = f"Observation: variable ##, which are instances of {range_info[relation]}" |
| 107 | base_program = variable.program if isinstance(variable, Variable) else variable |
| 108 | new_variable = Variable( |
| 109 | range_info[relation], |
| 110 | f"(JOIN {relation + '_inv'} {base_program})" |
| 111 | ) |
| 112 | return new_variable, rtn_str |
| 113 | |
| 114 | def intersection(self, variable1: Variable, variable2: Variable): |
| 115 | """ |
nothing calls this directly
no test coverage detected