|
1 | | -import re |
2 | | -import logging |
3 | 1 | 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 | 2 |
|
9 | 3 |
|
10 | 4 | @dataclass |
11 | 5 | class GeneralSchemaEnhancer: |
12 | 6 | """ |
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 |
| 7 | + Main functions have moved to core linkml, see https://github.com/linkml/linkml/pull/854 |
54 | 8 |
|
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 |
| 9 | + This is currently a stub for future enhancements |
| 10 | + """ |
| 11 | + pass |
120 | 12 |
|
121 | 13 |
|
122 | 14 |
|
|
0 commit comments