Coverage for rdgai/apparatus.py: 98.38%

1import math

2from typing import Optional

3from pathlib import Path

4from dataclasses import dataclass, field

5from lxml.etree import _Element as Element

6from lxml.etree import _ElementTree as ElementTree

7from lxml import etree as ET

8from rich.console import Console

9import functools

10import Levenshtein

11import numpy as np

13# from .relations import Relation, get_reading_identifier

14from .tei import read_tei, find_elements, extract_text, find_parent, find_element, write_tei, make_nc_name, get_language, get_reading_identifier, extract_text_siblings

15from .mapper import Mapper

17@dataclass

18class Reading():

19 element: Element

20 app:"App"

21 n: str = field(default=None)

22 text: str = field(default=None)

23 witnesses: list[str] = field(default_factory=list)

25 def __post_init__(self):

26 self.n = get_reading_identifier(self.element)

27 self.text = extract_text(self.element).strip()

28 self.witnesses = self.element.attrib.get("wit", "").split()

30 def __str__(self):

31 return self.text or 'OMIT'

33 def witnesses_str(self) -> str:

34 return " ".join(self.witnesses)

36 def __hash__(self):

37 return hash(self.element)

39 def text_in_context(self, text="") -> str:

40 return self.app.text_in_context(text or self.text)

43@dataclass

44class RelationType():

45 element: Element

46 name: str

47 description: str

48 inverse: Optional['RelationType'] = None

49 pairs: set['Pair'] = field(default_factory=set)

51 def __str__(self):

52 return self.name

54 def __repr__(self) -> str:

55 return str(self)

57 def __eq__(self, other):

58 if isinstance(other, RelationType):

59 return (self.name, self.element, self.description) == (other.name, other.element, other.description)

60 return False

62 def __hash__(self):

63 return hash((self.name, self.element, self.description))

65 def str_with_description(self) -> str:

66 result = self.name

67 if self.description:

68 result += f": {self.description}"

69 return result

71 def pairs_sorted(self, exclude_rdgai:bool = False) -> list['Pair']:

72 pairs = sorted(self.pairs, key=lambda pair: (str(pair.active.app), pair.active.n, pair.passive.n))

73 if exclude_rdgai:

74 pairs = [pair for pair in pairs if not pair.rdgai_responsible()]

75 return pairs

77 def get_inverse(self) -> 'RelationType':

78 return self.inverse if self.inverse else self

80 @functools.lru_cache(maxsize=None)

81 def representative_examples(self, k:int, random_state:int=42) -> list['Pair']:

83 def find_representative_examples(pairs_list:list[Pair], k:int, random_state:int=42):

84 import kmedoids

85 if len(pairs_list) <= k:

86 return pairs_list

87 distance_matrix = np.zeros((len(pairs_list), len(pairs_list)))

88 for index1, pair in enumerate(pairs_list):

89 for index2 in range(index1+1, len(pairs_list)):

90 other_pair = pairs_list[index2]

91 active_text_distance = Levenshtein.distance(pair.active.text, other_pair.active.text)

92 passive_text_distance = Levenshtein.distance(pair.passive.text, other_pair.passive.text)

93 distance = active_text_distance + passive_text_distance

94 distance_matrix[index1, index2] = distance

95 distance_matrix[index2, index1] = distance

97 result = kmedoids.fasterpam(distance_matrix, k, random_state=random_state, init="build")

99 return [pairs_list[index] for index in result.medoids]

100

101 pairs_list = self.pairs_sorted(exclude_rdgai=True)

102 pairs_with_descriptions = [pair for pair in pairs_list if pair.has_description()]

103 representative_pairs = []

104 if pairs_with_descriptions:

105 representative_pairs = find_representative_examples(pairs_with_descriptions, k, random_state=random_state)

106

107 if len(representative_pairs) < k:

108 pairs_without_descriptions = [pair for pair in pairs_list if not pair.has_description()]

109 additional_pairs = find_representative_examples(pairs_without_descriptions, k-len(representative_pairs), random_state=random_state)

110 representative_pairs.extend(additional_pairs)

111

112 return representative_pairs

113

114

115@dataclass

116class Pair():

117 active: Reading

118 passive: Reading

119 types: set[RelationType] = field(default_factory=set)

120

121 def __post_init__(self):

122 for relation_type in self.types:

123 relation_type.pairs.add(self)

124

125 def __str__(self):

126 return f"{self.active} ➞ {self.passive}"

127

128 def print(self, console):

129 console.print(f"[bold red]{self.app}[/bold red]: [green]{self.active}[/green] [red]➞[/red] [green]{self.passive}[/green]")

130

131 def reading_transition_str(self) -> str:

132 return f"{self.active or 'OMISSION'} → {self.passive or 'OMISSION'}"

133

134 def __repr__(self) -> str:

135 return str(self)

136

137 @property

138 def app(self) -> "App":

139 # assert self.active.app == self.passive.app

140 return self.active.app

141

142 def __hash__(self):

143 return hash((self.active, self.passive))

144

145 def app_element(self) -> Element:

146 return find_parent(self.active.element, "app")

147

148 def relation_elements(self) -> list[Element]:

149 list_relation = find_element(self.app_element(), ".//listRelation[@type='transcriptional']")

150 if list_relation is None:

151 return []

152

153 return find_elements(list_relation, f".//relation[@active='{self.active.n}'][@passive='{self.passive.n}']")

154

155 def element_for_type(self, type:RelationType) -> Element|None:

156 for relation in self.relation_elements():

157 if f"#{type.name}" in relation.attrib.get("ana").split():

158 return relation

159 return None

160

161 def get_inverse(self) -> "Pair":

162 found_pair = None

163 for pair in self.app.pairs:

164 if pair.active == self.passive and pair.passive == self.active:

165 found_pair = pair

166 break

167 assert found_pair is not None, f"No inverse pair found for {self}"

168 return found_pair

169

170 def add_type_with_inverse(self, type:RelationType, responsible:str|None=None, description:str="", inverse_description:str="") -> Element:

171 relation = self.add_type(type, responsible=responsible, description=description)

172 inverse = self.get_inverse()

173 inverse.add_type(type.get_inverse(), responsible=responsible, description=inverse_description)

174 return relation

175

176 def add_type(self, type:RelationType, responsible:str|None=None, description:str="") -> Element:

177 self.types.add(type)

178 type.pairs.add(self)

179

180 # Check if the relation already exists

181 relation = self.element_for_type(type)

182 if relation is not None:

183 return relation

184

185 list_relation = find_element(self.app_element(), ".//listRelation[@type='transcriptional']")

186 if list_relation is None:

187 list_relation = ET.SubElement(self.app_element(), "listRelation", attrib={"type":"transcriptional"})

188

189 relation = find_element(list_relation, f".//relation[@active='{self.active.n}'][@passive='{self.passive.n}']")

190 if relation is not None:

191 if type.name not in relation.attrib.get("ana").split():

192 relation.attrib["ana"] += f" #{type.name}"

193 else:

194 relation = ET.SubElement(list_relation, "relation", attrib={"active":self.active.n, "passive":self.passive.n, "ana":f"#{type.name}"})

195

196 if responsible is not None:

197 relation.set("resp", responsible)

198

199 self.add_description(description, relation)

200

201 return relation

202

203 def remove_description(self):

204 for relation in self.relation_elements():

205 for desc in find_elements(relation, ".//desc"):

206 relation.remove(desc)

207

208 def add_description(self, description:str, relation:Element|None=None):

209 if relation is None:

210 relation_elements = self.relation_elements()

211

212 if len(relation_elements) == 0:

213 list_relation = find_element(self.app_element(), ".//listRelation[@type='transcriptional']")

214 if list_relation is None:

215 list_relation = ET.SubElement(self.app_element(), "listRelation", attrib={"type":"transcriptional"})

216 relation = ET.SubElement(list_relation, "relation", attrib={"active":self.active.n, "passive":self.passive.n})

217 else:

218 relation = relation_elements[0]

219

220 description = description.strip()

221 if description:

222 description_element = find_element(relation, ".//desc")

223 if description_element is None:

224 description_element = ET.SubElement(relation, "desc")

225

226 description_element.text = description

227

228 def remove_type(self, relation_type:RelationType):

229 if relation_type in self.types:

230 self.types.remove(relation_type)

231

232 if self in relation_type.pairs:

233 relation_type.pairs.remove(self)

234

235 list_relation = find_element(self.app_element(), ".//listRelation[@type='transcriptional']")

236 for relation in find_elements(list_relation, f".//relation[@active='{self.active.n}'][@passive='{self.passive.n}']"):

237 if f"#{relation_type.name}" in relation.attrib.get("ana").split():

238 relation.attrib['ana'] = " ".join([ana for ana in relation.attrib.get("ana").split() if ana != f"#{relation_type.name}"])

239 if not relation.attrib.get("ana"):

240 relation.getparent().remove(relation)

241

242 def remove_type_with_inverse(self, relation_type:RelationType):

243 self.remove_type(relation_type)

244 inverse = self.get_inverse()

245 inverse.remove_type(relation_type.get_inverse())

246

247 def remove_all_types(self):

248 for relation_type in set(self.types):

249 self.remove_type_with_inverse(relation_type)

250

251 def rdgai_responsible(self) -> bool:

252 for element in self.relation_elements():

253 if element.attrib.get('resp', '') == '#rdgai':

254 return True

255 return False

256

257 def relation_type_names(self) -> set[str]:

258 return set(type.name for type in self.types)

259

260 def has_description(self) -> bool:

261 for relation in self.relation_elements():

262 if find_element(relation, ".//desc") is not None:

263 return True

264 return False

265

266 def get_description(self) -> str:

267 description = ""

268 for relation_element in self.relation_elements():

269 for desc in find_elements(relation_element, ".//desc"):

270 description += "\n" + extract_text(desc)

271 return description.strip()

272

273

274@dataclass

275class App():

276 element: Element

277 doc: "Doc"

278 readings: list[Reading] = field(default_factory=list)

279 pairs: list[Pair] = field(default_factory=list)

280 non_redundant_pairs: list[Pair] = field(default_factory=list)

281 id_to_reading: dict[str,Reading] = field(default_factory=dict)

282

283 def __post_init__(self):

284 for reading in find_elements(self.element, ".//rdg"):

285 self.readings.append(Reading(reading, app=self))

286

287 # Build list of relation elements

288 relation_elements = []

289 for list_relation in find_elements(self.element, ".//listRelation[@type='transcriptional']"):

290 for relation_element in find_elements(list_relation, ".//relation"):

291 relation_elements.append(relation_element)

292

293 # Build list of relation pairs

294 active_visited = set()

295 for active in self.readings:

296 active_visited.add(active)

297 for passive in self.readings:

298 if active == passive:

299 continue

300

301 types = set()

302 for relation_element in relation_elements:

303 if relation_element.attrib.get("active") == active.n and relation_element.attrib.get("passive") == passive.n:

304 for ana in relation_element.attrib.get("ana", "").split():

305 if ana.startswith("#"):

306 ana = ana[1:]

307 if ana:

308 types.add(ana)

309

310 pair_relation_types = set()

311 for type_name in types:

312 relation_type = self.doc.relation_types[type_name] if type_name in self.doc.relation_types else self.doc.add_relation_type(type_name)

313 pair_relation_types.add(relation_type)

314

315 pair = Pair(active=active, passive=passive, types=pair_relation_types)

316 self.pairs.append(pair)

317 if passive not in active_visited:

318 self.non_redundant_pairs.append(pair)

319

320 for relation_type in pair_relation_types:

321 relation_type.pairs.add(pair)

322

323 assert len(self.pairs) == len(self.non_redundant_pairs) * 2

324

325 self.id_to_reading = {reading.n: reading for reading in self.readings}

326

327 def __getitem__(self, key):

328 return self.id_to_reading[key]

329

330 def __len__(self):

331 return len(self.readings)

332

333 def get_classified_pairs(self, redundant:bool=True) -> list[Pair]:

334 pairs = self.pairs if redundant else self.non_redundant_pairs

335 return [pair for pair in pairs if len(pair.types) > 0]

336

337 def get_unclassified_pairs(self, redundant:bool=True) -> list[Pair]:

338 pairs = self.pairs if redundant else self.non_redundant_pairs

339 return [pair for pair in pairs if len(pair.types) == 0]

340

341 def __hash__(self):

342 return hash(self.element)

343

344 def __str__(self):

345 name = self.element.attrib.get('{http://www.w3.org/XML/1998/namespace}id', '')

346 if not name:

347 name = self.element.attrib.get('n', '')

348

349 if not name:

350 ab = self.ab()

351 if ab is not None:

352 for index, app in enumerate(find_elements(ab, ".//app")):

353 if app == self.element:

354 name = make_nc_name(f"{self.ab_name()}-{index+1}")

355 self.element.attrib['{http://www.w3.org/XML/1998/namespace}id'] = name

356 break

357 if not name:

358 root = self.element.getroottree().getroot()

359 for index, app in enumerate(find_elements(root, ".//app")):

360 if app == self.element:

361 name = make_nc_name(f"app-{index+1}")

362 self.element.attrib['{http://www.w3.org/XML/1998/namespace}id'] = name

363 break

364 return str(name).replace(" ", "_").replace(":", "_")

365

366 def ab(self) -> Element|None:

367 return find_parent(self.element, "ab")

368

369 def ab_name(self) -> str:

370 ab = self.ab()

371 if ab is None:

372 return ""

373 return ab.attrib.get("n", "")

374

375 def text_before(self) -> str:

376 ab = self.ab()

377 if ab is None:

378 return extract_text_siblings(self.element, "milestone", truncate=100, preceding=True)

379

380 items = []

381 for child in ab:

382 if child == self.element:

383 break

384 child_text = extract_text(child)

385 if child_text:

386 items.append(child_text)

387

388 text = " ".join(items)

389 return text.strip()

390

391 def text_in_context(self, text="") -> str:

392 return f"{self.text_before()} {self.text_with_signs(text)} {self.text_after()}".strip()

393

394 def text(self) -> str:

395 return extract_text(self.element)

396

397 def text_with_signs(self, text="") -> str:

398 text = text or self.text()

399 if not text:

400 return "⸆"

401 return f"⸂{text}⸃"

402

403 def text_after(self) -> str:

404 ab = self.ab()

405 if ab is None:

406 return extract_text_siblings(self.element, "milestone", truncate=100)

407

408 items = []

409 reached_element = False

410 for child in ab:

411 if reached_element:

412 child_text = extract_text(child)

413 if child_text:

414 items.append(child_text)

415 if child == self.element:

416 reached_element = True

417

418 text = " ".join(items)

419 return text.strip()

420

421 def entropy(self) -> float:

422 counts = [len(reading.witnesses) for reading in self.readings if len(reading.witnesses) > 0]

423 total = sum(counts)

424 probabilities = [count / total for count in counts]

425 entropy = sum(-p * math.log2(p) for p in probabilities)

426

427 return entropy

428

429

430@dataclass

431class Doc():

432 path: Path

433 tree: ElementTree = field(default=None)

434 apps: list[App] = field(default_factory=list)

435 relation_types: dict[str,RelationType] = field(default_factory=dict)

436 id_to_app: dict[str,App] = field(default_factory=dict)

437

438 def __post_init__(self):

439 self.tree = read_tei(self.path)

440 self.relation_types = self.get_relation_types()

441

442 for app_element in find_elements(self.tree, ".//app"):

443 app = App(app_element, doc=self)

444 self.apps.append(app)

445

446 self.id_to_app = {app.__str__(): app for app in self.apps}

447

448 def __getitem__(self, key):

449 return self.id_to_app[key]

450

451 def __len__(self):

452 return len(self.apps)

453

454 def get_interpgrp(self) -> Element:

455 text = find_element(self.tree, ".//text")

456 interp_group = find_element(text, ".//interpGrp[@type='transcriptional']")

457 if interp_group is None:

458 interp_group = ET.Element("interpGrp", attrib={"type":"transcriptional"})

459 text.insert(0, interp_group)

460

461 return interp_group

462

463 def add_relation_type(self, name:str, description:str="") -> RelationType:

464 if name in self.relation_types:

465 assert self.relation_types[name].description == description, f"RelationType {name} already exists with a different description."

466 return self.relation_types[name]

467

468 interp_group = self.get_interpgrp()

469 interp = find_element(interp_group, f".//interp[@xml:id='{name}']")

470 if interp is None:

471 interp = ET.Element("interp", attrib={"{http://www.w3.org/XML/1998/namespace}id":name})

472 interp_group.append(interp)

473

474 relation_type = RelationType(name=name, element=interp, description="")

475 self.relation_types[name] = relation_type

476 return relation_type

477

478 def __str__(self):

479 return str(self.path)

480

481 def __repr__(self) -> str:

482 return str(self)

483

484 def write(self, output:str|Path):

485 write_tei(self.tree, output)

486

487 @property

488 def language(self):

489 return get_language(self.tree)

490

491 def get_relation_types(self, categories_to_ignore:list[str]|None=None) -> list[RelationType]:

492 interp_group = self.get_interpgrp()

493 categories_to_ignore = categories_to_ignore or []

494

495 relation_types = dict()

496 assert interp_group is not None, "No interpGrp of type='transcriptional' found in TEI file."

497

498 for interp in find_elements(interp_group, "./interp"):

499 name = interp.attrib.get("{http://www.w3.org/XML/1998/namespace}id", "")

500 if name in categories_to_ignore: continue

501

502 description = extract_text(interp).strip()

503 relation_types[name] = RelationType(name=name, element=interp, description=description)

504

505 # get corresponding relations

506 for category in relation_types.values():

507 inverse_name = category.element.attrib.get("corresp", "")

508 if inverse_name.startswith("#"):

509 inverse_name = inverse_name[1:]

510

511 if inverse_name in relation_types:

512 inverse = relation_types[inverse_name]

513 category.inverse = inverse

514 if inverse.inverse is None:

515 inverse.inverse = category

516 else:

517 assert inverse.inverse == category, f"Inverse category {inverse} already has an inverse {inverse.inverse}."

518

519 return relation_types

520

521 def get_classified_pairs(self, redundant:bool=True) -> list[Pair]:

522 pairs = []

523 for app in self.apps:

524 pairs.extend(app.get_classified_pairs(redundant=redundant))

525

526 return pairs

527

528 def get_unclassified_pairs(self, redundant:bool=True) -> list[Pair]:

529 pairs = []

530 for app in self.apps:

531 pairs.extend(app.get_unclassified_pairs(redundant=redundant))

532

533 return pairs

534

535 def print_classified_pairs(self, console:Console|None=None) -> None:

536 console = console or Console()

537 for relation_type in self.relation_types.values():

538 console.rule(str(relation_type))

539 console.print(relation_type.description, style="grey46")

540 for pair in relation_type.pairs_sorted():

541 pair.print(console)

542

543 console.print("")

544

545 def render_html(self, output:Path|None=None, all_apps:bool=False) -> str:

546 from flask import Flask, request, render_template

547

548 mapper = Mapper()

549 app = Flask(__name__)

550

551 with app.app_context():

552 html = render_template('server.html', doc=self, mapper=mapper, all_apps=all_apps)

553

554 if output:

555 output.parent.mkdir(parents=True, exist_ok=True)

556 output.write_text(html)

557

558 return html

559

560 def flask_app(self, output:Path, all_apps:bool=False):

561 from flask import Flask, request, render_template

562

563 self.write(output)

564 mapper = Mapper()

565

566 app = Flask(__name__)

567

568 @app.route("/")

569 def root():

570 return render_template('server.html', doc=self, mapper=mapper, all_apps=all_apps)

571

572 @app.route("/api/relation-type", methods=['POST'])

573 def api_relation_type():

574 data = request.get_json()

575

576 relation_type = mapper.obj(data['relation_type'])

577 assert isinstance(relation_type, RelationType), f"Expected RelationType, got {type(relation_type)}"

578

579 pair = mapper.obj(data['pair'])

580 assert isinstance(pair, Pair), f"Expected Pair, got {type(pair)}"

581

582 try:

583 if data['operation'] == 'remove':

584 print('remove', relation_type)

585 pair.remove_type_with_inverse(relation_type)

586 elif data['operation'] == 'add':

587 print('add', relation_type)

588 pair.add_type_with_inverse(relation_type)

589 else:

590 raise ValueError(f"Unknown operation {data['operation']}")

591

592 print('write', output)

593 self.write(output)

594 return "Success", 200

595 except Exception as e:

596 print(str(e))

597 return str(e), 400

598

599 return "Failed", 400

600

601 @app.route("/api/desc", methods=['POST'])

602 def desc():

603 data = request.get_json()

604

605 pair = mapper.obj(data['pair'])

606 assert isinstance(pair, Pair), f"Expected Pair, got {type(pair)}"

607

608 try:

609 if data['operation'] == 'remove':

610 pair.remove_description()

611 elif data['operation'] == 'add':

612 pair.add_description(data['description'])

613 else:

614 raise ValueError(f"Unknown operation {data['operation']}")

615

616 print('write', output)

617 self.write(output)

618 return "Success", 200

619 except Exception as e:

620 print(str(e))

621 return str(e), 400

622

623 return "Failed", 400

624

625 return app

626 # app.run(debug=True, use_reloader=True)

627

628 def clean(self, output:Path|None=None):

629 """ Cleans a TEI XML file for common errors. """

630

631 # find all listRelation elements

632 list_relations = find_elements(self.tree, ".//listRelation")

633 for list_relation in list_relations:

634 relations_so_far = set()

635 for relation in find_elements(list_relation, ".//relation"):

636 # make sure that relation elements have a # at the start of the ana attribute

637 if not relation.attrib['ana'].startswith("#"):

638 relation.attrib['ana'] = f"#{relation.attrib['ana']}"

639

640 relations_so_far.add( (relation.attrib['active'], relation.attrib['passive']) )

641

642 # consolidate duplicate relations

643 for active, passive in relations_so_far:

644 relations = find_elements(list_relation, f".//relation[@active='{active}'][@passive='{passive}']")

645 if len(relations) > 1:

646 analytic_set = set()

647 for relation in relations:

648 analytic_set.update(relation.attrib['ana'].split())

649

650 for relation in relations[1:]:

651 list_relation.remove(relation)

652

653 relations[0].attrib['ana'] = " ".join(sorted(analytic_set))

654

655 if output:

656 output = Path(output)

657 output.parent.mkdir(parents=True, exist_ok=True)

658 print("Writing to", output)

659 self.write(output)