|
| 1 | +import re |
| 2 | +import logging |
| 3 | +from dataclasses import dataclass |
| 4 | +from typing import List, Callable |
| 5 | + |
| 6 | +from linkml_runtime import SchemaView |
| 7 | +from linkml_runtime.linkml_model import SchemaDefinition, ClassDefinition, ClassDefinitionName, SlotDefinition |
| 8 | + |
| 9 | + |
| 10 | +@dataclass |
| 11 | +class GeneralSchemaEnhancer: |
| 12 | + """ |
| 13 | + Multiple methods for adding additional information to schemas |
| 14 | + """ |
| 15 | + |
| 16 | + def add_titles(self, schema: SchemaDefinition): |
| 17 | + """ |
| 18 | + Add titles to all elements if not present |
| 19 | +
|
| 20 | + :param schema: input schema, will be modified in place |
| 21 | + :return: |
| 22 | + """ |
| 23 | + sv = SchemaView(schema) |
| 24 | + pattern = re.compile(r'(?<!^)(?=[A-Z])') |
| 25 | + for e in sv.all_elements(): |
| 26 | + if e.title is not None: |
| 27 | + continue |
| 28 | + title = e.name.replace('_', ' ') |
| 29 | + title = pattern.sub(' ', title).lower() |
| 30 | + e.title = title |
| 31 | + |
| 32 | + |
| 33 | + def add_container(self, schema: SchemaDefinition, class_name: str = 'Container', |
| 34 | + force: bool = False) -> ClassDefinition: |
| 35 | + """ |
| 36 | + Adds a container class |
| 37 | +
|
| 38 | + :param schema: input schema, will be modified in place |
| 39 | + :param class_name: |
| 40 | + :param force: |
| 41 | + :return: container class |
| 42 | + """ |
| 43 | + sv = SchemaView(schema) |
| 44 | + tree_roots = [for c in sv.all_classes().values() if c.tree_root] |
| 45 | + if len(tree_roots) > 0: |
| 46 | + if force: |
| 47 | + logging.info(f'Forcing addition of containers') |
| 48 | + else: |
| 49 | + raise ValueError(f'Schema already has containers: {tree_roots}') |
| 50 | + container = ClassDefinition(class_name, tree_root=True) |
| 51 | + sv.add_class(container) |
| 52 | + self.add_index_slots(container.name) |
| 53 | + return container |
| 54 | + |
| 55 | + |
| 56 | + def add_index_slots(self, schema: SchemaDefinition, container_name: ClassDefinitionName, inlined_as_list = False, |
| 57 | + must_have_identifier = False, slot_name_func: Callable = None) -> List[SlotDefinition]: |
| 58 | + """ |
| 59 | + Adds index slots to a container pointing at all top-level classes |
| 60 | +
|
| 61 | + :param schema: input schema, will be modified in place |
| 62 | + :param container_name: |
| 63 | + :param inlined_as_list: |
| 64 | + :param must_have_identifier: |
| 65 | + :param slot_name_func: function to determine the name of the slot from the class |
| 66 | + :return: new slots |
| 67 | + """ |
| 68 | + sv = SchemaView(schema) |
| 69 | + container = sv.get_class(container_name) |
| 70 | + ranges = set() |
| 71 | + for cn in sv.all_classes(): |
| 72 | + for s in sv.class_induced_slots(cn): |
| 73 | + ranges.add(s.range) |
| 74 | + top_level_classes = [c for c in sv.all_classes().values() if not c.tree_root and c.name not in ranges] |
| 75 | + if must_have_identifier: |
| 76 | + top_level_classes = [c for c in top_level_classes if sv.get_identifier_slot(c.name) is not None] |
| 77 | + index_slots = [] |
| 78 | + for c in top_level_classes: |
| 79 | + has_identifier = sv.get_identifier_slot(c.name) |
| 80 | + if slot_name_func: |
| 81 | + sn = slot_name_func(c) |
| 82 | + else: |
| 83 | + sn = f'{c.name}_index' |
| 84 | + index_slot = SlotDefinition(sn, |
| 85 | + range=c.name, |
| 86 | + multivalued=True, |
| 87 | + inlined_as_list=not has_identifier or inlined_as_list) |
| 88 | + index_slots.append(index_slot) |
| 89 | + container.slots.append(index_slot.name) |
| 90 | + return index_slots |
| 91 | + |
| 92 | + def attributes_to_slots(self, schema: SchemaDefinition): |
| 93 | + sv = SchemaView(schema) |
| 94 | + new_slots = [] |
| 95 | + for c in sv.all_classes().values(): |
| 96 | + for a in c.attributes: |
| 97 | + new_slots.append(a) |
| 98 | + self.merge_slot_usage(sv, c, a) |
| 99 | + c.attributes = {} |
| 100 | + for slot in new_slots: |
| 101 | + if slot.name in sv.all_slots(): |
| 102 | + raise ValueError(f'Duplicate slot {slot.name}') |
| 103 | + sv.add_slot(slot) |
| 104 | + self.remove_redundant_slot_usage(schema) |
| 105 | + |
| 106 | + def merge_slot_usage(self, sv: SchemaView, cls: ClassDefinition, slot: SlotDefinition): |
| 107 | + if slot.name not in cls.slot_usage: |
| 108 | + cls.slot_usage[slot.name] = slot |
| 109 | + else: |
| 110 | + su = cls.slot_usage[slot.name] |
| 111 | + for k, v in vars(slot).items(): |
| 112 | + curr_v = getattr(su, k, None) |
| 113 | + if curr_v and curr_v != v: |
| 114 | + raise ValueError(f'Conflict in {cls.name}.{slot.name}, attr {k} {v} != {curr_v}') |
| 115 | + setattr(su, k, v) |
| 116 | + |
| 117 | + def remove_redundant_slot_usage(self, schema: SchemaDefinition): |
| 118 | + sv = SchemaView(schema) |
| 119 | + # TODO |
| 120 | + |
| 121 | + |
| 122 | + |
| 123 | + |
| 124 | + |
| 125 | + |
0 commit comments