Get all relations of a variable (variable = program derivation or entity id). Returns (None, "Observation: [...]").
(self, variable: Union[Variable, str])
| 52 | return results |
| 53 | |
| 54 | def get_relations(self, variable: Union[Variable, str]): |
| 55 | """ |
| 56 | Get all relations of a variable (variable = program derivation or entity id). |
| 57 | Returns (None, "Observation: [...]"). |
| 58 | """ |
| 59 | if not isinstance(variable, Variable): |
| 60 | if not re.match(r'^([mf])\.[\w_]+$', variable): |
| 61 | raise ValueError("get_relations: variable must be a variable or an entity") |
| 62 | |
| 63 | cache_key = (self.task_id, variable if isinstance(variable, str) else hash(variable)) |
| 64 | |
| 65 | if cache_key in relation_cache: |
| 66 | out_relations = relation_cache[cache_key] |
| 67 | else: |
| 68 | if isinstance(variable, Variable): |
| 69 | program = variable.program |
| 70 | processed_code = postprocess_raw_code(program) |
| 71 | sparql_query = lisp_to_sparql(processed_code) |
| 72 | |
| 73 | clauses = sparql_query.split("\n") |
| 74 | new_clauses = [clauses[0], "SELECT DISTINCT ?rel\nWHERE {\n?x ?rel ?obj .\n{"] |
| 75 | new_clauses.extend(clauses[1:]) |
| 76 | new_clauses.append("}\n}") |
| 77 | new_query = '\n'.join(new_clauses) |
| 78 | |
| 79 | out_relations = self.sparql_executor.execute_query(new_query) |
| 80 | else: |
| 81 | out_relations = self.sparql_executor.get_out_relations(variable) |
| 82 | |
| 83 | out_relations = list(set(out_relations).intersection(set(relations))) |
| 84 | relation_cache[cache_key] = out_relations |
| 85 | |
| 86 | rtn_str = f"Observation: [{', '.join(out_relations)}]" |
| 87 | variable_relations_cache[variable] = out_relations |
| 88 | return None, rtn_str |
| 89 | |
| 90 | def get_neighbors(self, variable: Union[Variable, str], relation: str): |
| 91 | """ |
no test coverage detected