|
| 1 | +import json |
1 | 2 | from pathlib import Path |
2 | 3 | from typing import IO, Dict, Iterable, Iterator, Union |
3 | 4 |
|
4 | 5 | from basyx.aas import model |
5 | 6 | from basyx.aas.model import provider as sdk_provider |
6 | 7 |
|
7 | | -import app.adapter as adapter |
8 | 8 | from app.model import descriptor |
9 | 9 |
|
10 | 10 | PathOrIO = Union[Path, IO] |
@@ -51,29 +51,35 @@ def __iter__(self) -> Iterator[_DESCRIPTOR_TYPE]: |
51 | 51 | return iter(self._backend.values()) |
52 | 52 |
|
53 | 53 |
|
| 54 | +_DESCRIPTOR_KEY_TO_CLS = ( |
| 55 | + ("assetAdministrationShellDescriptors", descriptor.AssetAdministrationShellDescriptor), |
| 56 | + ("submodelDescriptors", descriptor.SubmodelDescriptor), |
| 57 | +) |
| 58 | + |
| 59 | + |
54 | 60 | def load_directory(directory: Union[Path, str]) -> DictDescriptorStore: |
55 | 61 | """ |
56 | | - Create a new :class:`~basyx.aas.model.provider.DictIdentifiableStore` and use it to load Asset Administration Shell |
57 | | - and Submodel files in ``AASX``, ``JSON`` and ``XML`` format from a given directory into memory. Additionally, load |
58 | | - all embedded supplementary files into a new :class:`~basyx.aas.adapter.aasx.DictSupplementaryFileContainer`. |
59 | | -
|
60 | | - :param directory: :class:`~pathlib.Path` or ``str`` pointing to the directory containing all Asset Administration |
61 | | - Shell and Submodel files to load |
62 | | - :return: Tuple consisting of a :class:`~basyx.aas.model.provider.DictIdentifiableStore` and a |
63 | | - :class:`~basyx.aas.adapter.aasx.DictSupplementaryFileContainer` containing all loaded data |
64 | | - """ |
| 62 | + Load AAS/Submodel descriptor JSON files from a directory into a :class:`DictDescriptorStore`. |
65 | 63 |
|
66 | | - dict_descriptor_store: DictDescriptorStore = DictDescriptorStore() |
| 64 | + :param directory: Path to the directory containing JSON descriptor files |
| 65 | + :return: Populated :class:`DictDescriptorStore` |
| 66 | + """ |
| 67 | + from app.adapter import ServerAASFromJsonDecoder |
67 | 68 |
|
| 69 | + store = DictDescriptorStore() |
68 | 70 | directory = Path(directory) |
69 | 71 |
|
70 | 72 | for file in directory.iterdir(): |
71 | | - if not file.is_file(): |
| 73 | + if not file.is_file() or file.suffix.lower() != ".json": |
72 | 74 | continue |
73 | | - |
74 | | - suffix = file.suffix.lower() |
75 | | - if suffix == ".json": |
76 | | - with open(file) as f: |
77 | | - adapter.read_server_aas_json_file_into(dict_descriptor_store, f) |
78 | | - |
79 | | - return dict_descriptor_store |
| 75 | + with open(file) as f: |
| 76 | + data = json.load(f, cls=ServerAASFromJsonDecoder) |
| 77 | + for key, cls in _DESCRIPTOR_KEY_TO_CLS: |
| 78 | + for item in data.get(key, []): |
| 79 | + if isinstance(item, cls): |
| 80 | + try: |
| 81 | + store.add(item) |
| 82 | + except KeyError: |
| 83 | + pass |
| 84 | + |
| 85 | + return store |
0 commit comments