25098769931

Committed 29 Apr 2026 08:30AM UTC coverage: 83.966% (-0.02%) from 83.987%

Build # 25098769931

Build Type

Pull #615

github

Committed by

web-flow

Commit Message

Merge 1321aff3e into c752b149a

Pull Request Pull Request #615: Discontinue RDF+SHACL and JSON-LD generators

Coverage Stats

30593 of 36435 relevant lines covered (83.97%)

3.36 hits per line

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

70.43

/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
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."""

    CPP = "cpp"
    CSHARP = "csharp"
    GOLANG = "golang"
    JAVA = "java"
    JSONSCHEMA = "jsonschema"
    OPCUA = "opcua"
    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)

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

aas-core-works / aas-core-codegen / 25098769931

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