10843710178

Committed 13 Sep 2024 05:43AM UTC coverage: 89.626% (-0.5%) from 90.097%

Build # 10843710178

Build Type

Pull #14

github

Committed by

web-flow

Commit Message

Merge 880352d9a into f94a144d7

Pull Request Pull Request #14: roll down parent inheritance recursively

Run Details

1227 of 1368 branches covered (89.69%)

Branch coverage included in aggregate %.

27 of 36 new or added lines in 3 files covered. (75.0%)

1 existing line in 1 file now uncovered.

2324 of 2594 relevant lines covered (89.59%)

2.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.81

/nwb_linkml/src/nwb_linkml/adapters/namespaces.py

"""
Namespaces adapter

Wraps the :class:`nwb_schema_language.Namespaces` and other objects with convenience methods
for extracting information and generating translated schema
"""

import contextlib
from copy import copy
from pathlib import Path
from pprint import pformat
from typing import Dict, Generator, List, Optional

from linkml_runtime.dumpers import yaml_dumper
from linkml_runtime.linkml_model import Annotation, SchemaDefinition
from pydantic import Field, model_validator
import networkx as nx

from nwb_linkml.adapters.adapter import Adapter, BuildResult
from nwb_linkml.adapters.schema import SchemaAdapter
from nwb_linkml.lang_elements import NwbLangSchema
from nwb_linkml.ui import AdapterProgress
from nwb_schema_language import Namespaces


class NamespacesAdapter(Adapter):
    """
    Translate a NWB Namespace to a LinkML Schema
    """

    namespaces: Namespaces
    schemas: List[SchemaAdapter]
    imported: List["NamespacesAdapter"] = Field(default_factory=list)

    _completed: bool = False
    """whether we have run the :meth:`.complete_namespace` method"""

    @classmethod
    def from_yaml(cls, path: Path) -> "NamespacesAdapter":
        """
        Create a NamespacesAdapter from a nwb schema language namespaces yaml file.

        Also attempts to provide imported implicitly imported schema (using the namespace key,
        rather than source, eg. with hdmf-common)
        """
        from nwb_linkml.io import schema as schema_io
        from nwb_linkml.providers.git import DEFAULT_REPOS

        ns_adapter = schema_io.load_namespace_adapter(path)

        # try and find imported schema

        need_imports = []
        for needed in ns_adapter.needed_imports.values():
            # try to locate imports implied by the namespace schema,
            # but are either not provided by the current namespace
            # or are otherwise already provided in `imported` by the loader function
            need_imports.extend(
                [
                    n
                    for n in needed
                    if n not in ns_adapter.needed_imports and n not in ns_adapter.versions
                ]
            )

        for needed in need_imports:
            if needed in DEFAULT_REPOS:
                needed_source_ns = DEFAULT_REPOS[needed].provide_from_git()
                needed_adapter = NamespacesAdapter.from_yaml(needed_source_ns)
                ns_adapter.imported.append(needed_adapter)

        ns_adapter.complete_namespaces()

        return ns_adapter

    def build(
        self, skip_imports: bool = False, progress: Optional[AdapterProgress] = None
    ) -> BuildResult:
        """
        Build the NWB namespace to the LinkML Schema
        """

        if not self._completed:
            self.complete_namespaces()

        sch_result = BuildResult()
        for sch in self.schemas:
            if progress is not None:
                with contextlib.suppress(KeyError):
                    # happens when we skip builds due to caching
                    progress.update(sch.namespace, action=sch.name)
            sch_result += sch.build()
            if progress is not None:
                with contextlib.suppress(KeyError):
                    # happens when we skip builds due to caching
                    progress.update(sch.namespace, advance=1)

        # recursive step
        if not skip_imports:
            for imported in self.imported:
                imported_build = imported.build(progress=progress)
                sch_result += imported_build

        # now generate the top-level namespaces that import everything
        for ns in self.namespaces.namespaces:

            # add in monkeypatch nwb types
            nwb_lang = copy(NwbLangSchema)
            nwb_lang.annotations.update(
                {
                    "is_namespace": Annotation(tag="is_namespace", value="False"),
                    "namespace": Annotation(tag="namespace", value=ns.name),
                }
            )
            lang_schema_name = ".".join([ns.name, "nwb.language"])
            nwb_lang.name = lang_schema_name
            sch_result.schemas.append(nwb_lang)

            ns_schemas = [sch.name for sch in self.schemas if sch.namespace == ns.name]
            ns_schemas.append(lang_schema_name)

            # also add imports bc, well, we need them
            if not skip_imports:
                ns_schemas.extend(
                    [ns.name for imported in self.imported for ns in imported.namespaces.namespaces]
                )
            ns_schema = SchemaDefinition(
                name=ns.name,
                id=ns.name,
                description=ns.doc,
                version=ns.version,
                imports=ns_schemas,
                annotations=[
                    {"tag": "is_namespace", "value": True},
                    {"tag": "namespace", "value": ns.name},
                ],
            )
            sch_result.schemas.append(ns_schema)

        return sch_result

    @model_validator(mode="after")
    def _populate_schema_namespaces(self) -> None:
        """
        annotate for each schema which namespace imports it
        """
        for sch in self.schemas:
            # imports seem to always be from same folder, so we can just use name part
            sch_name = sch.path.name
            # find which namespace imports this schema file
            for ns in self.namespaces.namespaces:
                sources = [sch.source for sch in ns.schema_]
                if sch_name in sources or sch.path.stem in sources:
                    sch.namespace = ns.name
                    sch.version = ns.version
                    break
        return self

    def complete_namespaces(self):
        """
        After loading the namespace, and after any imports have been added afterwards,
        this must be called to complete the definitions of the contained schema objects.

        This is not automatic because NWB doesn't have a formal dependency resolution system,
        so it is often impossible to know which imports are needed until after the namespace
        adapter has been instantiated.

        It **is** automatically called if it hasn't been already by the :meth:`.build` method.
        """
        self.populate_imports()
        self._roll_down_inheritance()

        for i in self.imported:
            i.complete_namespaces()

        self._completed = True

    def _roll_down_inheritance(self):
        """
        nwb-schema-language inheritance doesn't work like normal python inheritance -
        instead of inheriting everything at the 'top level' of a class, it also
        recursively merges all properties from the parent objects.

        References:
            https://github.com/NeurodataWithoutBorders/pynwb/issues/1954
        """
        pass

    def inheritance_graph(self) -> nx.DiGraph:
        """
        Make a graph of all ``neurodata_types`` in the namespace and imports such that
        each node contains the group or dataset it describes,
        and has directed edges pointing at all the classes that inherit from it.

        In the case that the inheriting class does not itself have a ``neurodata_type_def``,
        it is
        """
        g = nx.DiGraph()
        for sch in self.all_schemas():
            for cls in sch.created_classes:
                pass

    def find_type_source(self, name: str) -> SchemaAdapter:
        """
        Given some neurodata_type_inc, find the schema that it's defined in.

        Rather than returning as soon as a match is found, check all
        """
        # First check within the main schema
        internal_matches = []
        for schema in self.schemas:
            class_names = [cls.neurodata_type_def for cls in schema.created_classes]
            if name in class_names:
                internal_matches.append(schema)

        if len(internal_matches) > 1:
            raise KeyError(
                f"Found multiple schemas in namespace that define {name}:\ninternal:"
                f" {pformat(internal_matches)}\nimported:{pformat(internal_matches)}"
            )
        elif len(internal_matches) == 1:
            return internal_matches[0]

        import_matches = []
        for imported_ns in self.imported:
            for schema in imported_ns.schemas:
                class_names = [cls.neurodata_type_def for cls in schema.created_classes]
                if name in class_names:
                    import_matches.append(schema)

        if len(import_matches) > 1:
            raise KeyError(
                f"Found multiple schemas in namespace that define {name}:\ninternal:"
                f" {pformat(internal_matches)}\nimported:{pformat(import_matches)}"
            )
        elif len(import_matches) == 1:
            return import_matches[0]
        else:
            raise KeyError(f"No schema found that define {name}")

    def populate_imports(self) -> "NamespacesAdapter":
        """
        Populate the imports that are needed for each schema file

        This function adds a string version of imported schema assuming the
        generated schema will live in the same directory. If the path to
        the imported schema needs to be adjusted, that should happen elsewhere
        (eg in :class:`.LinkMLProvider`) because we shouldn't know about
        directory structure or anything like that here.
        """
        for sch in self.schemas:
            for needs in sch.needed_imports:
                # shouldn't be recursive references, since imports should form a tree
                depends_on = self.find_type_source(needs)
                if depends_on not in sch.imports:
                    sch.imports.append(depends_on)

        return self

    def to_yaml(self, base_dir: Path) -> None:
        """
        Build the schemas, saving them to ``yaml`` files according to
        their ``name``

        Args:
            base_dir (:class:`.Path`): Directory to save ``yaml`` files
        """
        schemas = self.build().schemas
        base_dir = Path(base_dir)

        base_dir.mkdir(exist_ok=True)

        for schema in schemas:
            output_file = base_dir / (schema.name + ".yaml")
            yaml_dumper.dump(schema, output_file)

    @property
    def needed_imports(self) -> Dict[str, List[str]]:
        """
        List of other, external namespaces that we need to import.
        Usually provided as schema with a namespace but not a source

        Returns:
            {'namespace_name': ['needed_import_0', ...]}
        """
        needed_imports = {}
        for a_ns in self.namespaces.namespaces:
            needed_imports[a_ns.name] = []
            for potential_import in a_ns.schema_:
                if potential_import.namespace and not potential_import.source:
                    needed_imports[a_ns.name].append(potential_import.namespace)

        return needed_imports

    @property
    def versions(self) -> Dict[str, str]:
        """
        versions for each namespace
        """
        versions = {ns.name: ns.version for ns in self.namespaces.namespaces}
        for imported in self.imported:
            versions.update(imported.versions)
        return versions

    def namespace_schemas(self, name: str) -> List[str]:
        """
        Get the schemas that are defined in a given namespace
        """
        ns = [ns for ns in self.namespaces.namespaces if ns.name == name]
        if len(ns) == 0:
            for imported in self.imported:
                ns = [ns for ns in imported.namespaces.namespaces if ns.name == name]
                if len(ns) > 0:
                    ns = ns[0]
                    break
            else:
                raise NameError(f"Couldn't find namespace {name}")
        else:
            ns = ns[0]

        schema_names = [sch.source for sch in ns.schema_ if sch.source is not None]
        return schema_names

    def schema_namespace(self, name: str) -> Optional[str]:
        """
        Inverse of :meth:`.namespace_schemas` - given a schema name, get the namespace it's in
        """
        for ns in self.namespaces.namespaces:
            sources = [sch.source for sch in ns.schema_ if sch.source is not None]
            if name in sources:
                return ns.name
        return None

    def all_schemas(self) -> Generator[SchemaAdapter, None, None]:
        """
        Iterator over all schemas including imports
        """
        for sch in self.schemas:
            yield sch
        for imported in self.imported:
            for sch in imported:
                yield sch

1	"""
2	Namespaces adapter
3
4	Wraps the :class:`nwb_schema_language.Namespaces` and other objects with convenience methods
5	for extracting information and generating translated schema
6	"""
7
8	import contextlib	3✔
9	from copy import copy	3✔
10	from pathlib import Path	3✔
11	from pprint import pformat	3✔
12	from typing import Dict, Generator, List, Optional	3✔
13
14	from linkml_runtime.dumpers import yaml_dumper	3✔
15	from linkml_runtime.linkml_model import Annotation, SchemaDefinition	3✔
16	from pydantic import Field, model_validator	3✔
17	import networkx as nx	3✔
18
19	from nwb_linkml.adapters.adapter import Adapter, BuildResult	3✔
20	from nwb_linkml.adapters.schema import SchemaAdapter	3✔
21	from nwb_linkml.lang_elements import NwbLangSchema	3✔
22	from nwb_linkml.ui import AdapterProgress	3✔
23	from nwb_schema_language import Namespaces	3✔
24
25
26	class NamespacesAdapter(Adapter):	3✔
27	"""
28	Translate a NWB Namespace to a LinkML Schema
29	"""
30
31	namespaces: Namespaces	3✔
32	schemas: List[SchemaAdapter]	3✔
33	imported: List["NamespacesAdapter"] = Field(default_factory=list)	3✔
34
35	_completed: bool = False	3✔
36	"""whether we have run the :meth:`.complete_namespace` method"""	3✔
37
38	@classmethod	3✔
39	def from_yaml(cls, path: Path) -> "NamespacesAdapter":	3✔
40	"""
41	Create a NamespacesAdapter from a nwb schema language namespaces yaml file.
42
43	Also attempts to provide imported implicitly imported schema (using the namespace key,
44	rather than source, eg. with hdmf-common)
45	"""
46	from nwb_linkml.io import schema as schema_io	3✔
47	from nwb_linkml.providers.git import DEFAULT_REPOS	3✔
48
49	ns_adapter = schema_io.load_namespace_adapter(path)	3✔
50
51	# try and find imported schema
52
53	need_imports = []	3✔
54	for needed in ns_adapter.needed_imports.values():	3✔
55	# try to locate imports implied by the namespace schema,
56	# but are either not provided by the current namespace
57	# or are otherwise already provided in `imported` by the loader function
58	need_imports.extend(	3✔
59	[
60	n
61	for n in needed
62	if n not in ns_adapter.needed_imports and n not in ns_adapter.versions
63	]
64	)
65
66	for needed in need_imports:	3✔
67	if needed in DEFAULT_REPOS:	×
68	needed_source_ns = DEFAULT_REPOS[needed].provide_from_git()	×
69	needed_adapter = NamespacesAdapter.from_yaml(needed_source_ns)	×
70	ns_adapter.imported.append(needed_adapter)	×
71
72	ns_adapter.complete_namespaces()	3✔
73
74	return ns_adapter	3✔
75
76	def build(	3✔
77	self, skip_imports: bool = False, progress: Optional[AdapterProgress] = None
78	) -> BuildResult:
79	"""
80	Build the NWB namespace to the LinkML Schema
81	"""
82
83	if not self._completed:	3✔
84	self.complete_namespaces()	3✔
85
86	sch_result = BuildResult()	3✔
87	for sch in self.schemas:	3✔
88	if progress is not None:	3✔
89	with contextlib.suppress(KeyError):	3✔
90	# happens when we skip builds due to caching
91	progress.update(sch.namespace, action=sch.name)	3✔
92	sch_result += sch.build()	3✔
93	if progress is not None:	3✔
94	with contextlib.suppress(KeyError):	3✔
95	# happens when we skip builds due to caching
96	progress.update(sch.namespace, advance=1)	3✔
97
98	# recursive step
99	if not skip_imports:	3✔
100	for imported in self.imported:	3✔
101	imported_build = imported.build(progress=progress)	3✔
102	sch_result += imported_build	3✔
103
104	# now generate the top-level namespaces that import everything
105	for ns in self.namespaces.namespaces:	3✔
106
107	# add in monkeypatch nwb types
108	nwb_lang = copy(NwbLangSchema)	3✔
109	nwb_lang.annotations.update(	3✔
110	{
111	"is_namespace": Annotation(tag="is_namespace", value="False"),
112	"namespace": Annotation(tag="namespace", value=ns.name),
113	}
114	)
115	lang_schema_name = ".".join([ns.name, "nwb.language"])	3✔
116	nwb_lang.name = lang_schema_name	3✔
117	sch_result.schemas.append(nwb_lang)	3✔
118
119	ns_schemas = [sch.name for sch in self.schemas if sch.namespace == ns.name]	3✔
120	ns_schemas.append(lang_schema_name)	3✔
121
122	# also add imports bc, well, we need them
123	if not skip_imports:	3✔
124	ns_schemas.extend(	3✔
125	[ns.name for imported in self.imported for ns in imported.namespaces.namespaces]
126	)
127	ns_schema = SchemaDefinition(	3✔
128	name=ns.name,
129	id=ns.name,
130	description=ns.doc,
131	version=ns.version,
132	imports=ns_schemas,
133	annotations=[
134	{"tag": "is_namespace", "value": True},
135	{"tag": "namespace", "value": ns.name},
136	],
137	)
138	sch_result.schemas.append(ns_schema)	3✔
139
140	return sch_result	3✔
141
142	@model_validator(mode="after")	3✔
143	def _populate_schema_namespaces(self) -> None:	3✔
144	"""
145	annotate for each schema which namespace imports it
146	"""
147	for sch in self.schemas:	3✔
148	# imports seem to always be from same folder, so we can just use name part
149	sch_name = sch.path.name	3✔
150	# find which namespace imports this schema file
151	for ns in self.namespaces.namespaces:	3✔
152	sources = [sch.source for sch in ns.schema_]	3✔
153	if sch_name in sources or sch.path.stem in sources:	3✔
154	sch.namespace = ns.name	3✔
155	sch.version = ns.version	3✔
156	break	3✔
157	return self	3✔
158
159	def complete_namespaces(self):	3✔
160	"""
161	After loading the namespace, and after any imports have been added afterwards,
162	this must be called to complete the definitions of the contained schema objects.
163
164	This is not automatic because NWB doesn't have a formal dependency resolution system,
165	so it is often impossible to know which imports are needed until after the namespace
166	adapter has been instantiated.
167
168	It is automatically called if it hasn't been already by the :meth:`.build` method.
169	"""
170	self.populate_imports()	3✔
171	self._roll_down_inheritance()	3✔
172
173	for i in self.imported:	3✔
174	i.complete_namespaces()	3✔
175
176	self._completed = True	3✔
177
178	def _roll_down_inheritance(self):	3✔
179	"""
180	nwb-schema-language inheritance doesn't work like normal python inheritance -
181	instead of inheriting everything at the 'top level' of a class, it also
182	recursively merges all properties from the parent objects.
183
184	References:
185	https://github.com/NeurodataWithoutBorders/pynwb/issues/1954
186	"""
187	pass	3✔
188
189	def inheritance_graph(self) -> nx.DiGraph:	3✔
190	"""
191	Make a graph of all ``neurodata_types`` in the namespace and imports such that
192	each node contains the group or dataset it describes,
193	and has directed edges pointing at all the classes that inherit from it.
194
195	In the case that the inheriting class does not itself have a ``neurodata_type_def``,
196	it is
197	"""
NEW 198	g = nx.DiGraph()	×
NEW 199	for sch in self.all_schemas():	×
NEW 200	for cls in sch.created_classes:	×
NEW 201	pass	×
202
203	def find_type_source(self, name: str) -> SchemaAdapter:	3✔
204	"""
205	Given some neurodata_type_inc, find the schema that it's defined in.
206
207	Rather than returning as soon as a match is found, check all
208	"""
209	# First check within the main schema
210	internal_matches = []	3✔
211	for schema in self.schemas:	3✔
212	class_names = [cls.neurodata_type_def for cls in schema.created_classes]	3✔
213	if name in class_names:	3✔
214	internal_matches.append(schema)	3✔
215
216	if len(internal_matches) > 1:	3✔
217	raise KeyError(	×
218	f"Found multiple schemas in namespace that define {name}:\ninternal:"
219	f" {pformat(internal_matches)}\nimported:{pformat(internal_matches)}"
220	)
221	elif len(internal_matches) == 1:	3✔
222	return internal_matches[0]	3✔
223
224	import_matches = []	3✔
225	for imported_ns in self.imported:	3✔
226	for schema in imported_ns.schemas:	3✔
227	class_names = [cls.neurodata_type_def for cls in schema.created_classes]	3✔
228	if name in class_names:	3✔
229	import_matches.append(schema)	3✔
230
231	if len(import_matches) > 1:	3✔
232	raise KeyError(	×
233	f"Found multiple schemas in namespace that define {name}:\ninternal:"
234	f" {pformat(internal_matches)}\nimported:{pformat(import_matches)}"
235	)
236	elif len(import_matches) == 1:	3✔
237	return import_matches[0]	3✔
238	else:
239	raise KeyError(f"No schema found that define {name}")	×
240
241	def populate_imports(self) -> "NamespacesAdapter":	3✔
242	"""
243	Populate the imports that are needed for each schema file
244
245	This function adds a string version of imported schema assuming the
246	generated schema will live in the same directory. If the path to
247	the imported schema needs to be adjusted, that should happen elsewhere
248	(eg in :class:`.LinkMLProvider`) because we shouldn't know about
249	directory structure or anything like that here.
250	"""
251	for sch in self.schemas:	3✔
252	for needs in sch.needed_imports:	3✔
253	# shouldn't be recursive references, since imports should form a tree
254	depends_on = self.find_type_source(needs)	3✔
255	if depends_on not in sch.imports:	3✔
256	sch.imports.append(depends_on)	3✔
257
258	return self	3✔
259
260	def to_yaml(self, base_dir: Path) -> None:	3✔
261	"""
262	Build the schemas, saving them to ``yaml`` files according to
263	their ``name``
264
265	Args:
266	base_dir (:class:`.Path`): Directory to save ``yaml`` files
267	"""
268	schemas = self.build().schemas	×
269	base_dir = Path(base_dir)	×
270
271	base_dir.mkdir(exist_ok=True)	×
272
273	for schema in schemas:	×
274	output_file = base_dir / (schema.name + ".yaml")	×
275	yaml_dumper.dump(schema, output_file)	×
276
277	@property	3✔
278	def needed_imports(self) -> Dict[str, List[str]]:	3✔
279	"""
280	List of other, external namespaces that we need to import.
281	Usually provided as schema with a namespace but not a source
282
283	Returns:
284	{'namespace_name': ['needed_import_0', ...]}
285	"""
286	needed_imports = {}	3✔
287	for a_ns in self.namespaces.namespaces:	3✔
288	needed_imports[a_ns.name] = []	3✔
289	for potential_import in a_ns.schema_:	3✔
290	if potential_import.namespace and not potential_import.source:	3✔
291	needed_imports[a_ns.name].append(potential_import.namespace)	3✔
292
293	return needed_imports	3✔
294
295	@property	3✔
296	def versions(self) -> Dict[str, str]:	3✔
297	"""
298	versions for each namespace
299	"""
300	versions = {ns.name: ns.version for ns in self.namespaces.namespaces}	3✔
301	for imported in self.imported:	3✔
302	versions.update(imported.versions)	3✔
303	return versions	3✔
304
305	def namespace_schemas(self, name: str) -> List[str]:	3✔
306	"""
307	Get the schemas that are defined in a given namespace
308	"""
309	ns = [ns for ns in self.namespaces.namespaces if ns.name == name]	3✔
310	if len(ns) == 0:	3✔
311	for imported in self.imported:	×
312	ns = [ns for ns in imported.namespaces.namespaces if ns.name == name]	×
313	if len(ns) > 0:	×
314	ns = ns[0]	×
315	break	×
316	else:
317	raise NameError(f"Couldn't find namespace {name}")	×
318	else:
319	ns = ns[0]	3✔
320
321	schema_names = [sch.source for sch in ns.schema_ if sch.source is not None]	3✔
322	return schema_names	3✔
323
324	def schema_namespace(self, name: str) -> Optional[str]:	3✔
325	"""
326	Inverse of :meth:`.namespace_schemas` - given a schema name, get the namespace it's in
327	"""
328	for ns in self.namespaces.namespaces:	×
329	sources = [sch.source for sch in ns.schema_ if sch.source is not None]	×
330	if name in sources:	×
331	return ns.name	×
332	return None	×
333
334	def all_schemas(self) -> Generator[SchemaAdapter, None, None]:	3✔
335	"""
336	Iterator over all schemas including imports
337	"""
NEW 338	for sch in self.schemas:	×
NEW 339	yield sch	×
NEW 340	for imported in self.imported:	×
NEW 341	for sch in imported:	×
NEW 342	yield sch	×

p2p-ld / nwb-linkml / 10843710178

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous