25100358556

Committed 29 Apr 2026 09:08AM UTC coverage: 83.998% (+0.2%) from 83.835%

Build # 25100358556

Build Type

Pull #617

github

Committed by

web-flow

Commit Message

Merge df3d59e01 into c22bac8b7

Pull Request Pull Request #617: Discontinue protobuf generator

Coverage Stats

29191 of 34752 relevant lines covered (84.0%)

3.36 hits per line

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

66.99

/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.typescript.main as typescript_main
import aas_core_codegen.xsd.main as xsd_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."""

    CPP = "cpp"
    CSHARP = "csharp"
    GOLANG = "golang"
    JAVA = "java"
    JSONSCHEMA = "jsonschema"
    PYTHON = "python"
    TYPESCRIPT = "typescript"
    XSD = "xsd"


assert [target.name for target in Target] == sorted(
    another_target.name for another_target in Target
), f"{[target.name for target in Target]=}, {sorted(another_target.name for another_target in Target)=}"


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

aas-core-works / aas-core-codegen / 25100358556

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