25099405401

Committed 29 Apr 2026 08:45AM UTC coverage: 83.835% (-0.1%) from 83.966%

Build # 25099405401

Build Type

Pull #616

github

Committed by

web-flow

Commit Message

Merge f17e6e3cd into 7cd0b694a

Pull Request Pull Request #616: Discontinue OPC-UA generator

Coverage Stats

30111 of 35917 relevant lines covered (83.83%)

3.35 hits per line

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

69.37

/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
import aas_core_codegen.protobuf.main as protobuf_main
import aas_core_codegen.python_protobuf.main as python_protobuf_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"
    PROTOBUF = "protobuf"
    PYTHON = "python"
    PYTHON_PROTOBUF = "python_protobuf"
    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)

    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)

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

aas-core-works / aas-core-codegen / 25099405401

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