21706826598

Committed 05 Feb 2026 09:52AM UTC coverage: 82.275% (+0.01%) from 82.263%

Build # 21706826598

Build Type

push

github

Committed by

web-flow

Commit Message

Cache meta-models optionally (#579)

We introduce the caching of meta-models so that the re-runs of the
codegen use the temporary directory of the OS to store and load the
already parsed meta-models. This can speed up the runs significantly.In
particular, this is useful during development as test iterations go from
15 seconds to 5 seconds on i7-6600U CPU @ 2.60GHz.

Coverage Stats

30 of 31 new or added lines in 2 files covered. (96.77%)

1 existing line in 1 file now uncovered.

28654 of 34827 relevant lines covered (82.28%)

3.29 hits per line

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

72.95

/aas_core_codegen/main.py

"""Generate implementations and schemas based on an AAS meta-model."""

import argparse
import enum
import pathlib
import sys
from typing import TextIO

import aas_core_codegen
import aas_core_codegen.csharp.main as csharp_main
import aas_core_codegen.cpp.main as cpp_main
import aas_core_codegen.golang.main as golang_main
import aas_core_codegen.java.main as java_main
import aas_core_codegen.jsonschema.main as jsonschema_main
import aas_core_codegen.python.main as python_main
import aas_core_codegen.rdf_shacl.main as rdf_shacl_main
import aas_core_codegen.typescript.main as typescript_main
import aas_core_codegen.xsd.main as xsd_main
import aas_core_codegen.jsonld.main as jsonld_main
import aas_core_codegen.protobuf.main as protobuf_main
import aas_core_codegen.python_protobuf.main as python_protobuf_main
import aas_core_codegen.opcua.main as opcua_main
from aas_core_codegen import run, specific_implementations
from aas_core_codegen.common import LinenoColumner, assert_never

assert aas_core_codegen.__doc__ == __doc__


class Target(enum.Enum):
    """List available target implementations."""

    CSHARP = "csharp"
    CPP = "cpp"
    GOLANG = "golang"
    JAVA = "java"
    JSONSCHEMA = "jsonschema"
    PYTHON = "python"
    TYPESCRIPT = "typescript"
    RDF_SHACL = "rdf_shacl"
    XSD = "xsd"
    JSONLD_CONTEXT = "jsonld_context"
    PROTOBUF = "protobuf"
    PYTHON_PROTOBUF = "python_protobuf"
    OPCUA = "opcua"


class Parameters:
    """Represent the program parameters."""

    def __init__(
        self,
        model_path: pathlib.Path,
        target: Target,
        snippets_dir: pathlib.Path,
        output_dir: pathlib.Path,
        cache_model: bool = False,
    ) -> None:
        """Initialize with the given values."""
        self.model_path = model_path
        self.target = target
        self.snippets_dir = snippets_dir
        self.output_dir = output_dir
        self.cache_model = True


# noinspection SpellCheckingInspection
def execute(params: Parameters, stdout: TextIO, stderr: TextIO) -> int:
    """Run the program."""
    # region Basic checks
    if not params.model_path.exists():
        stderr.write(f"The --model_path does not exist: {params.model_path}\n")
        return 1

    if not params.model_path.is_file():
        stderr.write(
            f"The --model_path does not point to a file: {params.model_path}\n"
        )
        return 1

    if not params.snippets_dir.exists():
        stderr.write(f"The --snippets_dir does not exist: {params.snippets_dir}\n")
        return 1

    if not params.snippets_dir.is_dir():
        stderr.write(
            f"The --snippets_dir does not point to a directory: "
            f"{params.snippets_dir}\n"
        )
        return 1

    if not params.output_dir.exists():
        params.output_dir.mkdir(parents=True, exist_ok=True)
    else:
        if not params.output_dir.is_dir():
            stderr.write(
                f"The --output_dir does not point to a directory: "
                f"{params.output_dir}\n"
            )
            return 1

    # endregion

    # region Parse and understand

    spec_impls, spec_impls_errors = specific_implementations.read_from_directory(
        snippets_dir=params.snippets_dir
    )

    if spec_impls_errors:
        run.write_error_report(
            message="Failed to resolve the implementation-specific snippets",
            errors=spec_impls_errors,
            stderr=stderr,
        )
        return 1

    assert spec_impls is not None

    symbol_table_atok, error_message = run.load_model(
        model_path=params.model_path, cache_model=params.cache_model
    )
    if error_message is not None:
        stderr.write(error_message)
        return 1
    assert symbol_table_atok is not None

    ir_symbol_table, atok = symbol_table_atok

    # endregion

    # region Dispatch

    lineno_columner = LinenoColumner(atok=atok)

    run_context = run.Context(
        model_path=params.model_path,
        symbol_table=ir_symbol_table,
        spec_impls=spec_impls,
        lineno_columner=lineno_columner,
        output_dir=params.output_dir,
    )

    if params.target is Target.CSHARP:
        return csharp_main.execute(context=run_context, stdout=stdout, stderr=stderr)

    elif params.target is Target.CPP:
        return cpp_main.execute(context=run_context, stdout=stdout, stderr=stderr)

    elif params.target is Target.GOLANG:
        return golang_main.execute(context=run_context, stdout=stdout, stderr=stderr)

    elif params.target is Target.JAVA:
        return java_main.execute(context=run_context, stdout=stdout, stderr=stderr)

    elif params.target is Target.JSONSCHEMA:
        return jsonschema_main.execute(
            context=run_context, stdout=stdout, stderr=stderr
        )

    elif params.target is Target.PYTHON:
        return python_main.execute(context=run_context, stdout=stdout, stderr=stderr)

    elif params.target is Target.TYPESCRIPT:
        return typescript_main.execute(
            context=run_context, stdout=stdout, stderr=stderr
        )

    elif params.target is Target.RDF_SHACL:
        return rdf_shacl_main.execute(context=run_context, stdout=stdout, stderr=stderr)

    elif params.target is Target.XSD:
        return xsd_main.execute(context=run_context, stdout=stdout, stderr=stderr)

    elif params.target is Target.JSONLD_CONTEXT:
        return jsonld_main.execute(context=run_context, stdout=stdout, stderr=stderr)

    elif params.target is Target.PROTOBUF:
        return protobuf_main.execute(run_context, stdout=stdout, stderr=stderr)

    elif params.target is Target.PYTHON_PROTOBUF:
        return python_protobuf_main.execute(run_context, stdout=stdout, stderr=stderr)

    elif params.target is Target.OPCUA:
        return opcua_main.execute(run_context, stdout=stdout, stderr=stderr)

    else:
        assert_never(params.target)

    # endregion

    stdout.write(f"Code generated to: {params.output_dir}\n")
    return 0


def main(prog: str) -> int:
    """
    Execute the main routine.

    :param prog: name of the program to be displayed in the help
    :return: exit code
    """
    parser = argparse.ArgumentParser(prog=prog, description=__doc__)
    parser.add_argument("--model_path", help="path to the meta-model", required=True)
    parser.add_argument(
        "--snippets_dir",
        help="path to the directory containing implementation-specific code snippets",
        required=True,
    )
    parser.add_argument(
        "--output_dir", help="path to the generated code", required=True
    )
    parser.add_argument(
        "--target",
        help="target language or schema",
        required=True,
        choices=[literal.value for literal in Target],
    )
    parser.add_argument(
        "--cache_model",
        help=(
            "If set, cache the parsed meta-model in the temporary directory of your OS. "
            "This makes the repeated runs faster as the meta-model does not need to be "
            "reparsed."
        ),
        action="store_true",
    )
    parser.add_argument(
        "--version", help="show the current version and exit", action="store_true"
    )

    # NOTE (mristin, 2022-01-14):
    # The module ``argparse`` is not flexible enough to understand special options such
    # as ``--version`` so we manually hard-wire.
    if "--version" in sys.argv and "--help" not in sys.argv:
        print(aas_core_codegen.__version__)
        return 0

    args = parser.parse_args()

    target_to_str = {literal.value: literal for literal in Target}

    params = Parameters(
        model_path=pathlib.Path(args.model_path),
        target=target_to_str[args.target],
        snippets_dir=pathlib.Path(args.snippets_dir),
        output_dir=pathlib.Path(args.output_dir),
        cache_model=bool(args.cache_model),
    )

    return execute(params=params, stdout=sys.stdout, stderr=sys.stderr)


def entry_point() -> int:
    """Provide an entry point for a console script."""
    return main(prog="aas-core-codegen")


if __name__ == "__main__":
    sys.exit(main(prog="aas-core-codegen"))

1	"""Generate implementations and schemas based on an AAS meta-model."""
2
3	import argparse	4✔
4	import enum	4✔
5	import pathlib	4✔
6	import sys	4✔
7	from typing import TextIO	4✔
8
9	import aas_core_codegen	4✔
10	import aas_core_codegen.csharp.main as csharp_main	4✔
11	import aas_core_codegen.cpp.main as cpp_main	4✔
12	import aas_core_codegen.golang.main as golang_main	4✔
13	import aas_core_codegen.java.main as java_main	4✔
14	import aas_core_codegen.jsonschema.main as jsonschema_main	4✔
15	import aas_core_codegen.python.main as python_main	4✔
16	import aas_core_codegen.rdf_shacl.main as rdf_shacl_main	4✔
17	import aas_core_codegen.typescript.main as typescript_main	4✔
18	import aas_core_codegen.xsd.main as xsd_main	4✔
19	import aas_core_codegen.jsonld.main as jsonld_main	4✔
20	import aas_core_codegen.protobuf.main as protobuf_main	4✔
21	import aas_core_codegen.python_protobuf.main as python_protobuf_main	4✔
22	import aas_core_codegen.opcua.main as opcua_main	4✔
23	from aas_core_codegen import run, specific_implementations	4✔
24	from aas_core_codegen.common import LinenoColumner, assert_never	4✔
25
26	assert aas_core_codegen.__doc__ == __doc__	4✔
27
28
29	class Target(enum.Enum):	4✔
30	"""List available target implementations."""
31
32	CSHARP = "csharp"	4✔
33	CPP = "cpp"	4✔
34	GOLANG = "golang"	4✔
35	JAVA = "java"	4✔
36	JSONSCHEMA = "jsonschema"	4✔
37	PYTHON = "python"	4✔
38	TYPESCRIPT = "typescript"	4✔
39	RDF_SHACL = "rdf_shacl"	4✔
40	XSD = "xsd"	4✔
41	JSONLD_CONTEXT = "jsonld_context"	4✔
42	PROTOBUF = "protobuf"	4✔
43	PYTHON_PROTOBUF = "python_protobuf"	4✔
44	OPCUA = "opcua"	4✔
45
46
47	class Parameters:	4✔
48	"""Represent the program parameters."""
49
50	def __init__(	4✔
51	self,
52	model_path: pathlib.Path,
53	target: Target,
54	snippets_dir: pathlib.Path,
55	output_dir: pathlib.Path,
56	cache_model: bool = False,
57	) -> None:
58	"""Initialize with the given values."""
59	self.model_path = model_path	4✔
60	self.target = target	4✔
61	self.snippets_dir = snippets_dir	4✔
62	self.output_dir = output_dir	4✔
63	self.cache_model = True	4✔
64
65
66	# noinspection SpellCheckingInspection
67	def execute(params: Parameters, stdout: TextIO, stderr: TextIO) -> int:	4✔
68	"""Run the program."""
69	# region Basic checks
70	if not params.model_path.exists():	4✔
71	stderr.write(f"The --model_path does not exist: {params.model_path}\n")	×
72	return 1	×
73
74	if not params.model_path.is_file():	4✔
75	stderr.write(	×
76	f"The --model_path does not point to a file: {params.model_path}\n"
77	)
78	return 1	×
79
80	if not params.snippets_dir.exists():	4✔
81	stderr.write(f"The --snippets_dir does not exist: {params.snippets_dir}\n")	×
82	return 1	×
83
84	if not params.snippets_dir.is_dir():	4✔
85	stderr.write(	×
86	f"The --snippets_dir does not point to a directory: "
87	f"{params.snippets_dir}\n"
88	)
89	return 1	×
90
91	if not params.output_dir.exists():	4✔
92	params.output_dir.mkdir(parents=True, exist_ok=True)	×
93	else:
94	if not params.output_dir.is_dir():	4✔
95	stderr.write(	×
96	f"The --output_dir does not point to a directory: "
97	f"{params.output_dir}\n"
98	)
99	return 1	×
100
101	# endregion
102
103	# region Parse and understand
104
105	spec_impls, spec_impls_errors = specific_implementations.read_from_directory(	4✔
106	snippets_dir=params.snippets_dir
107	)
108
109	if spec_impls_errors:	4✔
110	run.write_error_report(	×
111	message="Failed to resolve the implementation-specific snippets",
112	errors=spec_impls_errors,
113	stderr=stderr,
114	)
115	return 1	×
116
117	assert spec_impls is not None	4✔
118
119	symbol_table_atok, error_message = run.load_model(	4✔
120	model_path=params.model_path, cache_model=params.cache_model
121	)
122	if error_message is not None:	4✔
123	stderr.write(error_message)	4✔
124	return 1	4✔
125	assert symbol_table_atok is not None	4✔
126
127	ir_symbol_table, atok = symbol_table_atok	4✔
128
129	# endregion
130
131	# region Dispatch
132
133	lineno_columner = LinenoColumner(atok=atok)	4✔
134
135	run_context = run.Context(	4✔
136	model_path=params.model_path,
137	symbol_table=ir_symbol_table,
138	spec_impls=spec_impls,
139	lineno_columner=lineno_columner,
140	output_dir=params.output_dir,
141	)
142
143	if params.target is Target.CSHARP:	4✔
144	return csharp_main.execute(context=run_context, stdout=stdout, stderr=stderr)	4✔
145
146	elif params.target is Target.CPP:	4✔
147	return cpp_main.execute(context=run_context, stdout=stdout, stderr=stderr)	4✔
148
149	elif params.target is Target.GOLANG:	4✔
150	return golang_main.execute(context=run_context, stdout=stdout, stderr=stderr)	4✔
151
152	elif params.target is Target.JAVA:	4✔
153	return java_main.execute(context=run_context, stdout=stdout, stderr=stderr)	4✔
154
155	elif params.target is Target.JSONSCHEMA:	4✔
156	return jsonschema_main.execute(	4✔
157	context=run_context, stdout=stdout, stderr=stderr
158	)
159
160	elif params.target is Target.PYTHON:	4✔
161	return python_main.execute(context=run_context, stdout=stdout, stderr=stderr)	4✔
162
163	elif params.target is Target.TYPESCRIPT:	4✔
164	return typescript_main.execute(	4✔
165	context=run_context, stdout=stdout, stderr=stderr
166	)
167
168	elif params.target is Target.RDF_SHACL:	4✔
169	return rdf_shacl_main.execute(context=run_context, stdout=stdout, stderr=stderr)	4✔
170
171	elif params.target is Target.XSD:	4✔
172	return xsd_main.execute(context=run_context, stdout=stdout, stderr=stderr)	4✔
173
174	elif params.target is Target.JSONLD_CONTEXT:	4✔
175	return jsonld_main.execute(context=run_context, stdout=stdout, stderr=stderr)	×
176
177	elif params.target is Target.PROTOBUF:	4✔
178	return protobuf_main.execute(run_context, stdout=stdout, stderr=stderr)	4✔
179
180	elif params.target is Target.PYTHON_PROTOBUF:	4✔
181	return python_protobuf_main.execute(run_context, stdout=stdout, stderr=stderr)	4✔
182
183	elif params.target is Target.OPCUA:	4✔
184	return opcua_main.execute(run_context, stdout=stdout, stderr=stderr)	4✔
185
186	else:
187	assert_never(params.target)	×
188
189	# endregion
190
191	stdout.write(f"Code generated to: {params.output_dir}\n")	×
192	return 0	×
193
194
195	def main(prog: str) -> int:	4✔
196	"""
197	Execute the main routine.
198
199	:param prog: name of the program to be displayed in the help
200	:return: exit code
201	"""
202	parser = argparse.ArgumentParser(prog=prog, description=__doc__)	×
203	parser.add_argument("--model_path", help="path to the meta-model", required=True)	×
204	parser.add_argument(	×
205	"--snippets_dir",
206	help="path to the directory containing implementation-specific code snippets",
207	required=True,
208	)
209	parser.add_argument(	×
210	"--output_dir", help="path to the generated code", required=True
211	)
212	parser.add_argument(	×
213	"--target",
214	help="target language or schema",
215	required=True,
216	choices=[literal.value for literal in Target],
217	)
NEW 218	parser.add_argument(	×
219	"--cache_model",
220	help=(
221	"If set, cache the parsed meta-model in the temporary directory of your OS. "
222	"This makes the repeated runs faster as the meta-model does not need to be "
223	"reparsed."
224	),
225	action="store_true",
226	)
UNCOV 227	parser.add_argument(	×
228	"--version", help="show the current version and exit", action="store_true"
229	)
230
231	# NOTE (mristin, 2022-01-14):
232	# The module ``argparse`` is not flexible enough to understand special options such
233	# as ``--version`` so we manually hard-wire.
234	if "--version" in sys.argv and "--help" not in sys.argv:	×
235	print(aas_core_codegen.__version__)	×
236	return 0	×
237
238	args = parser.parse_args()	×
239
240	target_to_str = {literal.value: literal for literal in Target}	×
241
242	params = Parameters(	×
243	model_path=pathlib.Path(args.model_path),
244	target=target_to_str[args.target],
245	snippets_dir=pathlib.Path(args.snippets_dir),
246	output_dir=pathlib.Path(args.output_dir),
247	cache_model=bool(args.cache_model),
248	)
249
250	return execute(params=params, stdout=sys.stdout, stderr=sys.stderr)	×
251
252
253	def entry_point() -> int:	4✔
254	"""Provide an entry point for a console script."""
255	return main(prog="aas-core-codegen")	×
256
257
258	if __name__ == "__main__":	4✔
259	sys.exit(main(prog="aas-core-codegen"))	×

aas-core-works / aas-core-codegen / 21706826598

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