Coverage for src / time_agnostic_library / agnostic_entity.py: 100%

1# SPDX-FileCopyrightText: 2021-2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5import re 

6 

7from time_agnostic_library.prov_entity import ProvEntity 

8from time_agnostic_library.sparql import Sparql, _n3_value 

9from time_agnostic_library.support import _cached_parse as _parse_datetime 

10from time_agnostic_library.support import convert_to_datetime 

11 

12_OPERATION_RE = re.compile(r"(DELETE|INSERT)\s+DATA", re.IGNORECASE) 

13_GRAPH_BLOCK_RE = re.compile(r"GRAPH\s*<([^>]+)>\s*\{", re.IGNORECASE) 

14 

15_RDF_TERM_RE = re.compile( 

16 r"<([^>]+)>" 

17 r'|"((?:[^"\\]|\\.)*)"\^\^<([^>]+)>' 

18 r'|"((?:[^"\\]|\\.)*)"@([a-zA-Z][\w-]*)' 

19 r'|"((?:[^"\\]|\\.)*)"' 

20 r"|'((?:[^'\\]|\\.)*)'" 

21 r"|(_:\S+)", 

22 re.DOTALL, 

23) 

24 

25_ESCAPE_CHAR_RE = re.compile(r"\\(.)") 

26_ESCAPE_CHAR_MAP = {"n": "\n", "r": "\r", "t": "\t"} 

27 

28_RDF_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type" 

29 

30_TRIPLE_LEN = 3 

31 

32 

33def _unescape_literal(s: str) -> str: 

34 if "\\" not in s: 

35 return s 

36 return _ESCAPE_CHAR_RE.sub( 

37 lambda m: _ESCAPE_CHAR_MAP.get(m.group(1), m.group(1)), s 

38 ) 

39 

40 

41def _normalize_literal(raw: str) -> str: 

42 unescaped = _unescape_literal(raw) 

43 return ( 

44 unescaped.replace("\\", "\\\\") 

45 .replace('"', '\\"') 

46 .replace("\n", "\\n") 

47 .replace("\r", "\\r") 

48 ) 

49 

50 

51def _regex_match_to_n3(match: re.Match) -> str: 

52 uri = match.group(1) 

53 if uri is not None: 

54 return f"<{uri}>" 

55 

56 typed_value = match.group(2) 

57 if typed_value is not None: 

58 return f'"{_normalize_literal(typed_value)}"^^<{match.group(3)}>' 

59 

60 lang_value = match.group(4) 

61 if lang_value is not None: 

62 return f'"{_normalize_literal(lang_value)}"@{match.group(5)}' 

63 

64 double_quoted = match.group(6) 

65 if double_quoted is not None: 

66 return f'"{_normalize_literal(double_quoted)}"' 

67 

68 single_quoted = match.group(7) 

69 if single_quoted is not None: 

70 return f'"{_normalize_literal(single_quoted)}"' 

71 

72 return match.group(8) 

73 

74 

75_BRACE_OR_QUOTE_RE = re.compile(r'[{}\'"]') 

76 

77 

78def _find_matching_close_brace(text: str, start: int) -> int: 

79 pos = start 

80 length = len(text) 

81 depth = 1 

82 while pos < length: 

83 m = _BRACE_OR_QUOTE_RE.search(text, pos) 

84 if m is None: 

85 return length 

86 pos = m.start() 

87 char = text[pos] 

88 if char == "{": 

89 depth += 1 

90 pos += 1 

91 elif char == "}": 

92 depth -= 1 

93 if depth == 0: 

94 return pos 

95 pos += 1 

96 else: 

97 quote_char = char 

98 pos += 1 

99 while pos < length: 

100 q = text.find(quote_char, pos) 

101 if q == -1: 

102 pos = length 

103 break 

104 num_backslashes = 0 

105 check = q - 1 

106 while check >= start and text[check] == "\\": 

107 num_backslashes += 1 

108 check -= 1 

109 if num_backslashes % 2 == 0: 

110 pos = q + 1 

111 break 

112 pos = q + 1 

113 return length 

114 

115 

116def _fast_parse_update( 

117 update_query: str, 

118) -> list[tuple[str, list[tuple[str, str, str, str]]]]: 

119 operations: list[tuple[str, list[tuple[str, str, str, str]]]] = [] 

120 operation_matches = list(_OPERATION_RE.finditer(update_query)) 

121 query_len = len(update_query) 

122 

123 for i, operation_match in enumerate(operation_matches): 

124 operation_type = ( 

125 "DeleteData" 

126 if operation_match.group(1).upper() == "DELETE" 

127 else "InsertData" 

128 ) 

129 

130 op_start = operation_match.end() 

131 op_end = ( 

132 operation_matches[i + 1].start() 

133 if i + 1 < len(operation_matches) 

134 else query_len 

135 ) 

136 operation_body = update_query[op_start:op_end] 

137 

138 quads: list[tuple[str, str, str, str]] = [] 

139 for graph_match in _GRAPH_BLOCK_RE.finditer(operation_body): 

140 graph_n3 = f"<{graph_match.group(1)}>" 

141 triples_start = graph_match.end() 

142 triples_end = _find_matching_close_brace(operation_body, triples_start) 

143 triples_text = operation_body[triples_start:triples_end] 

144 

145 terms: list[str] = [] 

146 for m in _RDF_TERM_RE.finditer(triples_text): 

147 terms.append(_regex_match_to_n3(m)) 

148 if len(terms) == _TRIPLE_LEN: 

149 quads.append((terms[0], terms[1], terms[2], graph_n3)) 

150 terms.clear() 

151 

152 operations.append((operation_type, quads)) 

153 

154 return operations 

155 

156 

157def _compose_update_queries( 

158 update_queries: list[str], 

159) -> tuple[set[tuple[str, ...]], set[tuple[str, ...]]]: 

160 additions: set[tuple[str, ...]] = set() 

161 deletions: set[tuple[str, ...]] = set() 

162 for uq in update_queries: 

163 for op_type, quads in _fast_parse_update(uq): 

164 if op_type == "DeleteData": 

165 for quad in quads: 

166 if quad in additions: 

167 additions.discard(quad) 

168 else: 

169 deletions.add(quad) 

170 elif op_type == "InsertData": 

171 for quad in quads: 

172 if quad in deletions: 

173 deletions.discard(quad) 

174 else: 

175 additions.add(quad) 

176 return additions, deletions 

177 

178 

179CONFIG_PATH = "./config.json" 

180 

181 

182_GEN_AT_TIME_N3 = f"<{ProvEntity.iri_generated_at_time}>" 

183_HAS_UQ_N3 = f"<{ProvEntity.iri_has_update_query}>" 

184 

185 

186def _extract_snapshot_update_queries( 

187 quads: set[tuple[str, ...]], 

188) -> dict[str, str | None]: 

189 by_subject: dict[str, dict[str, str]] = {} 

190 for quad in quads: 

191 if quad[1] in (_GEN_AT_TIME_N3, _HAS_UQ_N3): 

192 by_subject.setdefault(quad[0], {})[quad[1]] = _n3_value(quad[2]) 

193 result: dict[str, str | None] = {} 

194 for props in by_subject.values(): 

195 if _GEN_AT_TIME_N3 in props: 

196 result[props[_GEN_AT_TIME_N3]] = props.get(_HAS_UQ_N3) 

197 return result 

198 

199 

200_PROV_PREFIX = ProvEntity.PROV 

201_RDF_TYPE_N3 = f"<{_RDF_TYPE}>" 

202 

203 

204def _find_related_object_uris(entity_uri: str, graphs: dict) -> set[str]: 

205 entity_n3 = f"<{entity_uri}>" 

206 result = set() 

207 for quad_set in graphs.values(): 

208 if quad_set is None: 

209 continue 

210 for quad in quad_set: 

211 if ( 

212 quad[0] == entity_n3 

213 and quad[2].startswith("<") 

214 and _PROV_PREFIX not in quad[1] 

215 and quad[1] != _RDF_TYPE_N3 

216 ): 

217 result.add(_n3_value(quad[2])) 

218 return result 

219 

220 

221class AgnosticEntity: 

222 def __init__( 

223 self, 

224 res: str, 

225 config: dict, 

226 *, 

227 include_related_objects: bool = False, 

228 include_merged_entities: bool = False, 

229 include_reverse_relations: bool = False, 

230 ): 

231 self.res = res 

232 self.include_related_objects = include_related_objects 

233 self.include_merged_entities = include_merged_entities 

234 self.include_reverse_relations = include_reverse_relations 

235 self.config = config 

236 

237 def get_history(self, *, include_prov_metadata: bool = False) -> tuple: 

238 if ( 

239 self.include_related_objects 

240 or self.include_merged_entities 

241 or self.include_reverse_relations 

242 ): 

243 histories = {} 

244 self._collect_all_related_entities_histories( 

245 histories, include_prov_metadata=include_prov_metadata 

246 ) 

247 return self._get_merged_histories( 

248 histories, include_prov_metadata=include_prov_metadata 

249 ) 

250 entity_history = self._get_entity_current_state( 

251 include_prov_metadata=include_prov_metadata 

252 ) 

253 entity_history = self._get_old_graphs(entity_history) 

254 for uri, time_dict in entity_history[0].items(): 

255 for ts, quad_set in time_dict.items(): 

256 if quad_set is None: 

257 entity_history[0][uri][ts] = set() 

258 return tuple(entity_history) 

259 

260 def get_histories_by_entity(self, *, include_prov_metadata: bool = False) -> tuple: 

261 histories = {} 

262 self._collect_all_related_entities_histories( 

263 histories, include_prov_metadata=include_prov_metadata 

264 ) 

265 entity_histories = {} 

266 metadata = {} 

267 for entity_uri, (entity_history_dict, entity_metadata) in histories.items(): 

268 entity_states = entity_history_dict[entity_uri] 

269 if not entity_states: 

270 continue 

271 entity_histories[entity_uri] = { 

272 timestamp: quad_set if quad_set is not None else set() 

273 for timestamp, quad_set in entity_states.items() 

274 } 

275 if include_prov_metadata and entity_metadata: 

276 metadata[entity_uri] = entity_metadata[entity_uri] 

277 return entity_histories, metadata 

278 

279 def _collect_all_related_entities_histories( 

280 self, histories: dict, *, include_prov_metadata: bool 

281 ) -> None: 

282 main_entity = AgnosticEntity( 

283 self.res, 

284 self.config, 

285 include_related_objects=False, 

286 include_merged_entities=False, 

287 include_reverse_relations=False, 

288 ) 

289 entity_history = main_entity._get_entity_current_state( 

290 include_prov_metadata=include_prov_metadata 

291 ) 

292 entity_history = main_entity._get_old_graphs(entity_history) 

293 histories[self.res] = (entity_history[0], entity_history[1]) 

294 

295 processed_entities = {self.res} 

296 

297 if self.include_related_objects: 

298 self._collect_related_objects_recursively( 

299 self.res, 

300 processed_entities, 

301 histories, 

302 include_prov_metadata=include_prov_metadata, 

303 ) 

304 

305 if self.include_merged_entities: 

306 self._collect_merged_entities_recursively( 

307 self.res, 

308 processed_entities, 

309 histories, 

310 include_prov_metadata=include_prov_metadata, 

311 ) 

312 

313 if self.include_reverse_relations: 

314 self._collect_reverse_relations_recursively( 

315 self.res, 

316 processed_entities, 

317 histories, 

318 include_prov_metadata=include_prov_metadata, 

319 ) 

320 

321 def _collect_related_objects_recursively( 

322 self, 

323 entity_uri: str, 

324 processed_entities: set[str], 

325 histories: dict, 

326 *, 

327 include_prov_metadata: bool, 

328 depth: int | None = None, 

329 ) -> None: 

330 if depth is not None and depth <= 0: 

331 return 

332 

333 next_depth = None if depth is None else depth - 1 

334 

335 entity_graphs = ( 

336 histories[entity_uri][0][entity_uri] if entity_uri in histories else None 

337 ) 

338 if not entity_graphs: 

339 return 

340 

341 for obj_uri in _find_related_object_uris(entity_uri, entity_graphs): 

342 if obj_uri not in processed_entities: 

343 processed_entities.add(obj_uri) 

344 agnostic_entity = AgnosticEntity( 

345 obj_uri, 

346 self.config, 

347 include_related_objects=False, 

348 include_merged_entities=False, 

349 include_reverse_relations=False, 

350 ) 

351 entity_history = agnostic_entity._get_entity_current_state( 

352 include_prov_metadata=include_prov_metadata 

353 ) 

354 entity_history = agnostic_entity._get_old_graphs(entity_history) 

355 histories[obj_uri] = (entity_history[0], entity_history[1]) 

356 self._collect_related_objects_recursively( 

357 obj_uri, 

358 processed_entities, 

359 histories, 

360 include_prov_metadata=include_prov_metadata, 

361 depth=next_depth, 

362 ) 

363 

364 def _collect_merged_entities_recursively( 

365 self, 

366 entity_uri: str, 

367 processed_entities: set[str], 

368 histories: dict, 

369 *, 

370 include_prov_metadata: bool, 

371 depth: int | None = None, 

372 ) -> None: 

373 if depth is not None and depth <= 0: 

374 return 

375 

376 next_depth = None if depth is None else depth - 1 

377 

378 merged_entities = self._find_merged_entities(entity_uri) 

379 

380 for merged_entity_uri in merged_entities: 

381 if merged_entity_uri not in processed_entities: 

382 processed_entities.add(merged_entity_uri) 

383 agnostic_entity = AgnosticEntity( 

384 merged_entity_uri, 

385 self.config, 

386 include_related_objects=False, 

387 include_merged_entities=False, 

388 include_reverse_relations=False, 

389 ) 

390 entity_history = agnostic_entity._get_entity_current_state( 

391 include_prov_metadata=include_prov_metadata 

392 ) 

393 entity_history = agnostic_entity._get_old_graphs(entity_history) 

394 histories[merged_entity_uri] = (entity_history[0], entity_history[1]) 

395 self._collect_merged_entities_recursively( 

396 merged_entity_uri, 

397 processed_entities, 

398 histories, 

399 include_prov_metadata=include_prov_metadata, 

400 depth=next_depth, 

401 ) 

402 

403 def _collect_reverse_relations_recursively( 

404 self, 

405 entity_uri: str, 

406 processed_entities: set[str], 

407 histories: dict, 

408 *, 

409 include_prov_metadata: bool, 

410 depth: int | None = None, 

411 ) -> None: 

412 if depth is not None and depth <= 0: 

413 return 

414 

415 next_depth = None if depth is None else depth - 1 

416 

417 reverse_related_entities = self._find_reverse_related_entities(entity_uri) 

418 

419 for reverse_entity_uri in reverse_related_entities: 

420 if reverse_entity_uri not in processed_entities: 

421 processed_entities.add(reverse_entity_uri) 

422 agnostic_entity = AgnosticEntity( 

423 reverse_entity_uri, 

424 self.config, 

425 include_related_objects=False, 

426 include_merged_entities=False, 

427 include_reverse_relations=False, 

428 ) 

429 entity_history = agnostic_entity._get_entity_current_state( 

430 include_prov_metadata=include_prov_metadata 

431 ) 

432 entity_history = agnostic_entity._get_old_graphs(entity_history) 

433 histories[reverse_entity_uri] = (entity_history[0], entity_history[1]) 

434 self._collect_reverse_relations_recursively( 

435 reverse_entity_uri, 

436 processed_entities, 

437 histories, 

438 include_prov_metadata=include_prov_metadata, 

439 depth=next_depth, 

440 ) 

441 

442 def _get_merged_histories( 

443 self, histories: dict, *, include_prov_metadata: bool 

444 ) -> tuple: 

445 entity_histories = {} 

446 metadata = {} 

447 for entity_uri, (entity_history_dict, entity_metadata) in histories.items(): 

448 entity_histories[entity_uri] = entity_history_dict[entity_uri] 

449 if include_prov_metadata and entity_metadata: 

450 metadata[entity_uri] = entity_metadata[entity_uri] 

451 

452 main_entity_times = sorted( 

453 entity_histories[self.res].keys(), key=_parse_datetime 

454 ) 

455 

456 merged_histories = {self.res: {}} 

457 

458 related_sorted_times = {} 

459 for entity_uri, entity_history in entity_histories.items(): 

460 if entity_uri == self.res: 

461 continue 

462 related_sorted_times[entity_uri] = sorted( 

463 ((t, _parse_datetime(t)) for t in entity_history), key=lambda x: x[1] 

464 ) 

465 

466 for timestamp in main_entity_times: 

467 merged_set = set(entity_histories[self.res][timestamp]) 

468 timestamp_dt = _parse_datetime(timestamp) 

469 

470 for entity_uri, sorted_times in related_sorted_times.items(): 

471 relevant_time = None 

472 for etime, etime_dt in sorted_times: 

473 if etime_dt <= timestamp_dt: 

474 relevant_time = etime 

475 else: 

476 break 

477 if relevant_time: 

478 merged_set.update(entity_histories[entity_uri][relevant_time]) 

479 

480 merged_histories[self.res][timestamp] = merged_set 

481 

482 return merged_histories, metadata 

483 

484 def get_state_at_time( 

485 self, 

486 time: tuple[str | None, str | None], 

487 *, 

488 include_prov_metadata: bool = False, 

489 ) -> tuple: 

490 if ( 

491 self.include_related_objects 

492 or self.include_merged_entities 

493 or self.include_reverse_relations 

494 ): 

495 histories = {} 

496 self._collect_all_related_entities_states_at_time( 

497 histories, time, include_prov_metadata=include_prov_metadata 

498 ) 

499 return self._get_merged_histories_at_time( 

500 histories, include_prov_metadata=include_prov_metadata 

501 ) 

502 return self._get_entity_state_at_time( 

503 time, include_prov_metadata=include_prov_metadata 

504 ) 

505 

506 def get_delta( 

507 self, 

508 time_start: str, 

509 time_end: str, 

510 ) -> tuple[set[tuple[str, ...]], set[tuple[str, ...]]]: 

511 is_quadstore = self.config["provenance"]["is_quadstore"] 

512 graph_statement = f"GRAPH <{self.res}/prov/>" if is_quadstore else "" 

513 query_snapshots = f""" 

514 SELECT ?time ?updateQuery 

515 WHERE {{ 

516 {graph_statement} 

517 {{ 

518 ?snapshot <{ProvEntity.iri_specialization_of}> <{self.res}>; 

519 <{ProvEntity.iri_generated_at_time}> ?time. 

520 OPTIONAL {{ 

521 ?snapshot <{ProvEntity.iri_has_update_query}> ?updateQuery. 

522 }} 

523 }} 

524 }} 

525 """ 

526 results = Sparql(query_snapshots, config=self.config).run_select_query() 

527 bindings = results["results"]["bindings"] 

528 if not bindings: 

529 return set(), set() 

530 start_dt = _parse_datetime(time_start) 

531 end_dt = _parse_datetime(time_end) 

532 parsed = [(b, _parse_datetime(b["time"]["value"])) for b in bindings] 

533 first_snapshot_dt = min(dt for _, dt in parsed) 

534 if first_snapshot_dt > start_dt: 

535 entity_graphs, _, _ = self._get_entity_state_at_time( 

536 (time_end, time_end), include_prov_metadata=False 

537 ) 

538 if not entity_graphs: 

539 return set(), set() 

540 state_at_end = next(iter(entity_graphs.values())) 

541 return state_at_end, set() 

542 relevant = sorted( 

543 ( 

544 (b, dt) 

545 for b, dt in parsed 

546 if start_dt < dt <= end_dt 

547 and "updateQuery" in b 

548 and "value" in b["updateQuery"] 

549 ), 

550 key=lambda x: x[1], 

551 ) 

552 return _compose_update_queries([b["updateQuery"]["value"] for b, _ in relevant]) 

553 

554 def _collect_all_related_entities_states_at_time( 

555 self, 

556 histories: dict, 

557 time: tuple[str | None, str | None], 

558 *, 

559 include_prov_metadata: bool, 

560 ) -> None: 

561 main_entity = AgnosticEntity( 

562 self.res, 

563 self.config, 

564 include_related_objects=False, 

565 include_merged_entities=False, 

566 include_reverse_relations=False, 

567 ) 

568 entity_graphs, entity_snapshots, other_snapshots_metadata = ( 

569 main_entity._get_entity_state_at_time( 

570 time, include_prov_metadata=include_prov_metadata 

571 ) 

572 ) 

573 histories[self.res] = ( 

574 entity_graphs, 

575 entity_snapshots, 

576 other_snapshots_metadata, 

577 ) 

578 

579 processed_entities = {self.res} 

580 

581 if self.include_related_objects: 

582 self._collect_related_objects_states_at_time( 

583 self.res, 

584 processed_entities, 

585 histories, 

586 time, 

587 include_prov_metadata=include_prov_metadata, 

588 ) 

589 

590 if self.include_merged_entities: 

591 self._collect_merged_entities_states_at_time( 

592 self.res, 

593 processed_entities, 

594 histories, 

595 time, 

596 include_prov_metadata=include_prov_metadata, 

597 ) 

598 

599 if self.include_reverse_relations: 

600 self._collect_reverse_relations_states_at_time( 

601 self.res, 

602 processed_entities, 

603 histories, 

604 time, 

605 include_prov_metadata=include_prov_metadata, 

606 ) 

607 

608 def _collect_related_objects_states_at_time( 

609 self, 

610 entity_uri: str, 

611 processed_entities: set[str], 

612 histories: dict, 

613 time: tuple[str | None, str | None], 

614 *, 

615 include_prov_metadata: bool, 

616 depth: int | None = None, 

617 ) -> None: 

618 if depth is not None and depth <= 0: 

619 return 

620 

621 next_depth = None if depth is None else depth - 1 

622 

623 entity_graphs = histories[entity_uri][0] if entity_uri in histories else None 

624 if not entity_graphs: 

625 return 

626 

627 for obj_uri in _find_related_object_uris(entity_uri, entity_graphs): 

628 if obj_uri not in processed_entities: 

629 processed_entities.add(obj_uri) 

630 agnostic_entity = AgnosticEntity( 

631 obj_uri, 

632 self.config, 

633 include_related_objects=False, 

634 include_merged_entities=False, 

635 include_reverse_relations=False, 

636 ) 

637 entity_graphs_new, entity_snapshots, other_snapshots_metadata = ( 

638 agnostic_entity._get_entity_state_at_time( 

639 time, include_prov_metadata=include_prov_metadata 

640 ) 

641 ) 

642 histories[obj_uri] = ( 

643 entity_graphs_new, 

644 entity_snapshots, 

645 other_snapshots_metadata, 

646 ) 

647 self._collect_related_objects_states_at_time( 

648 obj_uri, 

649 processed_entities, 

650 histories, 

651 time, 

652 include_prov_metadata=include_prov_metadata, 

653 depth=next_depth, 

654 ) 

655 

656 def _collect_merged_entities_states_at_time( 

657 self, 

658 entity_uri: str, 

659 processed_entities: set[str], 

660 histories: dict, 

661 time: tuple[str | None, str | None], 

662 *, 

663 include_prov_metadata: bool, 

664 depth: int | None = None, 

665 ) -> None: 

666 if depth is not None and depth <= 0: 

667 return 

668 

669 next_depth = None if depth is None else depth - 1 

670 

671 merged_entities = self._find_merged_entities(entity_uri) 

672 

673 for merged_entity_uri in merged_entities: 

674 if merged_entity_uri not in processed_entities: 

675 processed_entities.add(merged_entity_uri) 

676 agnostic_entity = AgnosticEntity( 

677 merged_entity_uri, 

678 self.config, 

679 include_related_objects=False, 

680 include_merged_entities=False, 

681 include_reverse_relations=False, 

682 ) 

683 entity_graphs, entity_snapshots, other_snapshots_metadata = ( 

684 agnostic_entity._get_entity_state_at_time( 

685 time, include_prov_metadata=include_prov_metadata 

686 ) 

687 ) 

688 histories[merged_entity_uri] = ( 

689 entity_graphs, 

690 entity_snapshots, 

691 other_snapshots_metadata, 

692 ) 

693 self._collect_merged_entities_states_at_time( 

694 merged_entity_uri, 

695 processed_entities, 

696 histories, 

697 time, 

698 include_prov_metadata=include_prov_metadata, 

699 depth=next_depth, 

700 ) 

701 

702 def _collect_reverse_relations_states_at_time( 

703 self, 

704 entity_uri: str, 

705 processed_entities: set[str], 

706 histories: dict, 

707 time: tuple[str | None, str | None], 

708 *, 

709 include_prov_metadata: bool, 

710 depth: int | None = None, 

711 ) -> None: 

712 if depth is not None and depth <= 0: 

713 return 

714 

715 next_depth = None if depth is None else depth - 1 

716 

717 reverse_related_entities = self._find_reverse_related_entities(entity_uri) 

718 

719 for reverse_entity_uri in reverse_related_entities: 

720 if reverse_entity_uri not in processed_entities: 

721 processed_entities.add(reverse_entity_uri) 

722 agnostic_entity = AgnosticEntity( 

723 reverse_entity_uri, 

724 self.config, 

725 include_related_objects=False, 

726 include_merged_entities=False, 

727 include_reverse_relations=False, 

728 ) 

729 entity_graphs, entity_snapshots, other_snapshots_metadata = ( 

730 agnostic_entity._get_entity_state_at_time( 

731 time, include_prov_metadata=include_prov_metadata 

732 ) 

733 ) 

734 histories[reverse_entity_uri] = ( 

735 entity_graphs, 

736 entity_snapshots, 

737 other_snapshots_metadata, 

738 ) 

739 self._collect_reverse_relations_states_at_time( 

740 reverse_entity_uri, 

741 processed_entities, 

742 histories, 

743 time, 

744 include_prov_metadata=include_prov_metadata, 

745 depth=next_depth, 

746 ) 

747 

748 def _get_merged_histories_at_time( 

749 self, histories: dict, *, include_prov_metadata: bool 

750 ) -> tuple: 

751 entity_histories = {} 

752 entity_snapshots_metadata = {} 

753 other_snapshots_metadata = {} if include_prov_metadata else None 

754 

755 for entity_uri, ( 

756 entity_graphs, 

757 entity_snapshots, 

758 other_snapshots, 

759 ) in histories.items(): 

760 entity_histories[entity_uri] = entity_graphs 

761 entity_snapshots_metadata[entity_uri] = entity_snapshots 

762 if ( 

763 include_prov_metadata 

764 and other_snapshots 

765 and other_snapshots_metadata is not None 

766 ): 

767 other_snapshots_metadata[entity_uri] = other_snapshots 

768 

769 main_entity_times = sorted( 

770 set(entity_histories[self.res].keys()), key=_parse_datetime 

771 ) 

772 

773 merged_histories = {self.res: {}} 

774 

775 related_sorted_times = {} 

776 for entity_uri, graphs_at_times in entity_histories.items(): 

777 if entity_uri == self.res: 

778 continue 

779 related_sorted_times[entity_uri] = sorted( 

780 ((t, _parse_datetime(t)) for t in graphs_at_times), key=lambda x: x[1] 

781 ) 

782 

783 for timestamp in main_entity_times: 

784 merged_set = set(entity_histories[self.res][timestamp]) 

785 timestamp_dt = _parse_datetime(timestamp) 

786 

787 for entity_uri, sorted_times in related_sorted_times.items(): 

788 graphs_at_times = entity_histories[entity_uri] 

789 if timestamp in graphs_at_times: 

790 related_quads = graphs_at_times[timestamp] 

791 else: 

792 relevant_time = None 

793 for rt, rt_dt in sorted_times: 

794 if rt_dt <= timestamp_dt: 

795 relevant_time = rt 

796 else: 

797 break 

798 if relevant_time: 

799 related_quads = graphs_at_times[relevant_time] 

800 else: 

801 continue 

802 merged_set.update(related_quads) 

803 

804 merged_histories[self.res][timestamp] = merged_set 

805 

806 return merged_histories, entity_snapshots_metadata, other_snapshots_metadata 

807 

808 def _get_entity_state_at_time( 

809 self, time: tuple[str | None, str | None], *, include_prov_metadata: bool 

810 ) -> tuple: 

811 other_snapshots_metadata = {} 

812 is_quadstore = self.config["provenance"]["is_quadstore"] 

813 graph_statement = f"GRAPH <{self.res}/prov/>" if is_quadstore else "" 

814 if include_prov_metadata: 

815 query_snapshots = f""" 

816 SELECT ?snapshot ?time ?responsibleAgent ?updateQuery 

817 ?primarySource ?description ?invalidatedAtTime 

818 WHERE {{ 

819 {graph_statement} 

820 {{ 

821 ?snapshot <{ProvEntity.iri_specialization_of}> <{self.res}>; 

822 <{ProvEntity.iri_generated_at_time}> ?time; 

823 <{ProvEntity.iri_was_attributed_to}> ?responsibleAgent. 

824 OPTIONAL {{ 

825 ?snapshot <{ProvEntity.iri_invalidated_at_time}> 

826 ?invalidatedAtTime. 

827 }} 

828 OPTIONAL {{ 

829 ?snapshot <{ProvEntity.iri_description}> ?description. 

830 }} 

831 OPTIONAL {{ 

832 ?snapshot <{ProvEntity.iri_has_update_query}> ?updateQuery. 

833 }} 

834 OPTIONAL {{ 

835 ?snapshot <{ProvEntity.iri_had_primary_source}> 

836 ?primarySource. 

837 }} 

838 }} 

839 }} 

840 """ 

841 else: 

842 query_snapshots = f""" 

843 SELECT ?snapshot ?time ?updateQuery 

844 WHERE {{ 

845 {graph_statement} 

846 {{ 

847 ?snapshot <{ProvEntity.iri_specialization_of}> <{self.res}>; 

848 <{ProvEntity.iri_generated_at_time}> ?time. 

849 OPTIONAL {{