18375768062

Committed 09 Oct 2025 12:07PM UTC coverage: 92.132% (-0.7%) from 92.835%

Build # 18375768062

Build Type

Pull #549

github

Committed by

nielstron

Commit Message

Fixes for library exports

Pull Request Pull Request #549: Plutus V3 support

Run Details

1242 of 1458 branches covered (85.19%)

Branch coverage included in aggregate %.

38 of 47 new or added lines in 8 files covered. (80.85%)

27 existing lines in 4 files now uncovered.

4566 of 4846 relevant lines covered (94.22%)

4.71 hits per line

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

80.14

/opshin/__main__.py

import inspect

import argparse
import io
import logging
import os
import tempfile
from contextlib import redirect_stdout

import cbor2
import enum
import importlib
import json
import pathlib
import sys
import typing
import ast

import pycardano

import uplc
import uplc.ast
from uplc.cost_model import PlutusVersion

from . import (
    compiler,
    builder,
    prelude,
    __version__,
    __copyright__,
    Purpose,
    PlutusContract,
)
from .util import CompilerError, data_from_json, OPSHIN_LOG_HANDLER
from .prelude import ScriptContext
from .compiler_config import *
from uplc import cost_model


class Command(enum.Enum):
    compile_pluto = "compile_pluto"
    compile = "compile"
    eval = "eval"
    parse = "parse"
    eval_uplc = "eval_uplc"
    build = "build"
    lint = "lint"


def parse_uplc_param(param: str):
    if param.startswith("{"):
        try:
            return uplc.ast.data_from_json_dict(json.loads(param))
        except json.JSONDecodeError as e:
            raise ValueError(
                f"Invalid parameter for contract passed, expected JSON value, got {param}"
            ) from e
        except Exception as e:
            raise ValueError(
                f"Expected parameter for contract to be in valid Plutus data JSON format, got {param}"
            ) from e
    else:
        try:
            return uplc.ast.data_from_cbor(bytes.fromhex(param))
        except Exception as e:
            raise ValueError(
                "Expected hexadecimal CBOR representation of plutus datum but could not transform hex string to bytes."
            ) from e


def parse_plutus_param(annotation, param: str):
    try:
        if param.startswith("{"):
            try:
                param_dict = json.loads(param)
            except json.JSONDecodeError as e:
                raise ValueError(
                    f"Invalid parameter for contract passed, expected json value, got {param}"
                ) from e
            return plutus_data_from_json(annotation, param_dict)
        else:
            try:
                param_bytes = bytes.fromhex(param)
            except ValueError as e:
                raise ValueError(
                    "Expected hexadecimal CBOR representation of plutus datum but could not transform hex string to bytes."
                ) from e
            return plutus_data_from_cbor(annotation, param_bytes)
    except ValueError as e:
        raise ValueError(
            f"Could not parse parameter {param} as type {annotation}. Please provide the parameter either as JSON or CBOR (in hexadecimal notation). Detailed error: {e}"
        ) from e


def plutus_data_from_json(annotation: typing.Type, x: dict):
    try:
        if annotation == int:
            return int(x["int"])
        if annotation == bytes:
            return bytes.fromhex(x["bytes"])
        if annotation is None:
            return None
        if isinstance(annotation, typing._GenericAlias):
            # Annotation is a List or Dict
            if annotation._name == "List":
                annotation_ann = annotation.__dict__["__args__"][0]
                return [plutus_data_from_json(annotation_ann, k) for k in x["list"]]
            if annotation._name == "Dict":
                annotation_key, annotation_val = annotation.__dict__["__args__"]
                return {
                    plutus_data_from_json(
                        annotation_key, d["k"]
                    ): plutus_data_from_json(annotation_val, d["v"])
                    for d in x["map"]
                }
            if annotation.__origin__ == typing.Union:
                for ann in annotation.__dict__["__args__"]:
                    try:
                        return plutus_data_from_json(ann, x)
                    except (pycardano.DeserializeException, KeyError, ValueError):
                        pass
                raise ValueError(
                    f"Could not find matching type for {x} in {annotation}"
                )
            if annotation == pycardano.Datum:
                if "int" in x:
                    return int(x["int"])
                if "bytes" in x:
                    return bytes.fromhex(x["bytes"])
                if "constructor" in x:
                    return pycardano.RawCBOR(
                        uplc.ast.plutus_cbor_dumps(uplc.ast.data_from_json_dict(x))
                    )
                if "list" in x:
                    return [
                        plutus_data_from_json(pycardano.Datum, k) for k in x["list"]
                    ]
                if "map" in x:
                    return {
                        plutus_data_from_json(
                            pycardano.Datum, d["k"]
                        ): plutus_data_from_json(pycardano.Datum, d["v"])
                        for d in x["map"]
                    }
        if issubclass(annotation, pycardano.PlutusData):
            return annotation.from_dict(x)
    except (KeyError, ValueError):
        raise ValueError(
            f"Annotation {annotation} does not match provided plutus datum {json.dumps(x)}"
        )


def plutus_data_from_cbor(annotation: typing.Type, x: bytes):
    try:
        if annotation in (int, bytes):
            res = cbor2.loads(x)
            if not isinstance(res, annotation):
                raise ValueError(
                    f"Expected {annotation} but got {type(x)} from {x.hex()}"
                )
            return res
        if annotation is None:
            if not x == cbor2.dumps(None):
                raise ValueError(f"Expected None but got {x.hex()}")
            return None
        if isinstance(annotation, typing._GenericAlias):
            # Annotation is a List or Dict
            if annotation.__origin__ == list:
                annotation_ann = annotation.__dict__["__args__"][0]
                return [
                    plutus_data_from_cbor(annotation_ann, cbor2.dumps(k))
                    for k in cbor2.loads(x)
                ]
            if annotation.__origin__ == dict:
                annotation_key, annotation_val = annotation.__dict__["__args__"]
                return {
                    plutus_data_from_cbor(
                        annotation_key, cbor2.dumps(k)
                    ): plutus_data_from_cbor(annotation_val, cbor2.dumps(v))
                    for k, v in cbor2.loads(x).items()
                }
            if annotation.__origin__ == typing.Union:
                for ann in annotation.__dict__["__args__"]:
                    try:
                        return plutus_data_from_cbor(ann, x)
                    except (pycardano.DeserializeException, ValueError):
                        pass
                raise ValueError(
                    f"Could not find matching type for {x.hex()} in {annotation}"
                )
        if issubclass(annotation, pycardano.PlutusData):
            return annotation.from_cbor(x)
    except (KeyError, ValueError):
        raise ValueError(
            f"Annotation {annotation} does not match provided plutus datum {x.hex()}"
        )


def check_params(
    command: Command,
    validator_args,
    return_type,
    validator_params,
):
    num_onchain_params = 1
    onchain_params = validator_args[-num_onchain_params:]
    param_types = validator_args[:-num_onchain_params]
    if return_type is not None:
        print(
            f"Warning: validator returns {return_type}, but it is recommended to return None. In PlutusV3, validators that do not return None always fail. This is most likely not what you want."
        )

    required_onchain_parameters = 1
    assert (
        len(onchain_params) == required_onchain_parameters
    ), f"""\
validator must expect {required_onchain_parameters} parameters at evaluation (on-chain), but was specified to have {len(onchain_params)}.
Make sure the validator expects just the script context."""

    if command in (Command.eval, Command.eval_uplc):
        assert len(validator_params) == len(param_types) + len(
            onchain_params
        ), f"validator expects {len(param_types) + len(onchain_params)} parameters for evaluation, but only got {len(validator_params)}."
    assert (
        onchain_params[-1][1] == ScriptContext
    ), f"Last parameter of the validator has to be ScriptContext, but is {onchain_params[-1][1].__name__} here."
    return onchain_params, param_types


def perform_command(args):
    # generate the compiler config
    compiler_config = DEFAULT_CONFIG
    compiler_config = compiler_config.update(OPT_CONFIGS[args.opt_level])
    overrides = {}
    for k in ARGPARSE_ARGS.keys():
        if getattr(args, k) is not None:
            overrides[k] = getattr(args, k)
    compiler_config = compiler_config.update(CompilationConfig(**overrides))
    # configure logging
    if args.verbose:
        OPSHIN_LOG_HANDLER.setLevel(logging.DEBUG)
    lib = args.lib

    # execute the command
    command = Command(args.command)
    input_file = args.input_file if args.input_file != "-" else sys.stdin
    # read and import the contract
    with open(input_file, "r") as f:
        source_code = f.read()
    with tempfile.TemporaryDirectory(prefix="build") as tmpdir:
        tmp_input_file = pathlib.Path(tmpdir).joinpath("__tmp_opshin.py")
        with tmp_input_file.open("w") as fp:
            fp.write(source_code)
        sys.path.append(str(pathlib.Path(tmp_input_file).parent.absolute()))
        try:
            sc = importlib.import_module(pathlib.Path(tmp_input_file).stem)
        except Exception as e:
            # replace the traceback with an error pointing to the input file
            raise SyntaxError(
                f"Could not import the input file as python module. Make sure the input file is valid python code. Error: {e}",
            ) from e
        sys.path.pop()
    # load the passed parameters if not a lib
    try:
        argspec = inspect.signature(sc.validator if lib is None else getattr(sc, lib))
    except AttributeError:
        raise AssertionError(
            f"Contract has no function called '{'validator' if lib is None else lib}'. Make sure the compiled contract contains one function called 'validator'."
        )
    annotations = [
        (x.name, x.annotation or prelude.Anything) for x in argspec.parameters.values()
    ]
    return_annotation = (
        argspec.return_annotation
        if argspec.return_annotation is not argspec.empty
        else prelude.Anything
    )
    parsed_params = []
    uplc_params = []
    for i, (c, a) in enumerate(zip(annotations, args.args)):
        try:
            uplc_param = parse_uplc_param(a)
        except ValueError as e:
            raise ValueError(
                f"Could not parse parameter {i} ('{a}') as UPLC data. Please provide the parameter either as JSON or CBOR (in hexadecimal notation). Detailed error: {e}"
            ) from None
        uplc_params.append(uplc_param)
        try:
            param = parse_plutus_param(c[1], a)
        except ValueError as e:
            raise ValueError(
                f"Could not parse parameter {i} ('{a}') as type {c[1]}. Please provide the parameter either as JSON or CBOR (in hexadecimal notation). Detailed error: {e}"
            ) from None
        parsed_params.append(param)
    if lib is None:
        onchain_params, param_types = check_params(
            command,
            annotations,
            return_annotation,
            parsed_params,
        )
        assert (
            onchain_params
        ), "The validator function must have at least one on-chain parameter. You can also add `_:None`."

    py_ret = Command.eval
    if command == Command.eval:
        print("Python execution started")
        with redirect_stdout(open(os.devnull, "w")):
            try:
                py_ret = sc.validator(*parsed_params)
            except Exception as e:
                py_ret = e
        command = Command.eval_uplc

    source_ast = compiler.parse(source_code, filename=input_file)

    if command == Command.parse:
        print("Parsed successfully.")
        return

    try:
        code = compiler.compile(
            source_ast,
            filename=input_file,
            validator_function_name="validator" if lib is None else lib,
            # do not remove dead code when compiling a library - none of the code will be used
            config=compiler_config,
        )
    except CompilerError as c:
        # Generate nice error message from compiler error
        if not isinstance(c.node, ast.Module):
            source_seg = ast.get_source_segment(source_code, c.node)
            start_line = c.node.lineno - 1
            end_line = start_line + len(source_seg.splitlines())
            source_lines = "\n".join(source_code.splitlines()[start_line:end_line])
            pos_in_line = source_lines.find(source_seg)
        else:
            start_line = 0
            pos_in_line = 0
            source_lines = source_code.splitlines()[0]

        overwrite_syntaxerror = (
            len("SyntaxError: ") * "\b" if command != Command.lint else ""
        )
        err = SyntaxError(
            f"""\
{overwrite_syntaxerror}{c.orig_err.__class__.__name__}: {c.orig_err}
Note that opshin errors may be overly restrictive as they aim to prevent code with unintended consequences.
""",
            (
                args.input_file,
                start_line + 1,
                pos_in_line,
                source_lines,
            ),
            # we remove chaining so that users to not see the internal trace back,
        )
        err.orig_err = c.orig_err
        raise err

    if command == Command.compile_pluto:
        print(code.dumps())
        return
    code = pluthon.compile(code, config=compiler_config)

    # apply parameters from the command line to the contract (instantiates parameterized contract!)
    code = code.term
    # UPLC lambdas may only take one argument at a time, so we evaluate by repeatedly applying
    for d in uplc_params:
        code = uplc.ast.Apply(code, d)
    code = uplc.ast.Program((1, 0, 0), code)

    if command == Command.compile:
        print(code.dumps())
        return

    if command == Command.build:
        if lib is not None:
            raise ValueError(
                "Cannot build a library. Please remove the --lib flag when building a contract."
            )
        if args.output_directory == "":
            if args.input_file == "-":
                print(
                    "Please supply an output directory if no input file is specified."
                )
                exit(-1)
            target_dir = pathlib.Path("build") / pathlib.Path(input_file).stem
        else:
            target_dir = pathlib.Path(args.output_directory)
        built_code = builder._build(code)
        script_arts = PlutusContract(
            built_code,
            datum_type=onchain_params[0] if len(onchain_params) == 3 else None,
            redeemer_type=onchain_params[1 if len(onchain_params) == 3 else 0],
            parameter_types=param_types,
            purpose=(Purpose.any,),
            title=pathlib.Path(input_file).stem,
        )
        script_arts.dump(target_dir)

        print(f"Wrote script artifacts to {target_dir}/")
        return
    if command == Command.eval_uplc:
        print("UPLC execution started")
        assert isinstance(code, uplc.ast.Program)
        raw_ret = uplc.eval(
            code,
            cek_machine_cost_model=cost_model.default_cek_machine_cost_model_plutus_v3(),
            builtin_cost_model=cost_model.default_builtin_cost_model_plutus_v3(),
        )
        print("------LOGS--------")
        if raw_ret.logs:
            for log in raw_ret.logs:
                print(" > " + log)
        else:
            print("No logs")
        print("------COST--------")
        print(f"CPU: {raw_ret.cost.cpu} | MEM: {raw_ret.cost.memory}")
        if isinstance(raw_ret.result, Exception):
            print("----EXCEPTION-----")
            ret = raw_ret.result
        else:
            print("-----SUCCESS------")
            ret = uplc.dumps(raw_ret.result)
        print(str(ret), end="")
        if not isinstance(py_ret, Command):
            print(" (Python: " + str(py_ret) + ")", end="")
        print()


def parse_args():
    a = argparse.ArgumentParser(
        description="An evaluator and compiler from python into UPLC. Translate imperative programs into functional quasi-assembly. Flags allow setting fine-grained compiler options. All flags can be turned off via -fno-<flag>."
    )
    a.add_argument(
        "command",
        type=str,
        choices=Command.__members__.keys(),
        help="The command to execute on the input file.",
    )
    a.add_argument(
        "input_file", type=str, help="The input program to parse. Set to - for stdin."
    )
    a.add_argument(
        "--lib",
        const="validator",
        default=None,
        nargs="?",
        type=str,
        help="Indicates that the input file should compile to a generic function, reusable by other contracts (not a smart contract itself). Discards corresponding typechecks. An optional name of the function to export can be given, by default it is validator. Use -fwrap_input and -fwrap_output to control whether the function expects or returns PlutusData (otherwise BuiltIn).",
    )
    a.add_argument(
        "-o",
        "--output-directory",
        default="",
        type=str,
        help="The output directory for artefacts of the build command. Defaults to the filename of the compiled contract. of the compiled contract.",
    )
    a.add_argument(
        "args",
        nargs="*",
        default=[],
        help="Input parameters for the validator (parameterizes the contract for compile/build). Either json or CBOR notation.",
    )
    a.add_argument(
        "--output-format-json",
        action="store_true",
        help="Changes the output of the Linter to a json format.",
    )
    a.add_argument(
        "--version",
        action="version",
        version=f"opshin {__version__} {__copyright__}",
    )
    a.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Enable verbose logging.",
    )
    a.add_argument(
        "--recursion-limit",
        default=max(sys.getrecursionlimit(), 4000),
        help="Modify the recursion limit (necessary for larger UPLC programs)",
        type=int,
    )
    for k, v in ARGPARSE_ARGS.items():
        alts = v.pop("__alts__", [])
        type = v.pop("type", None)
        if type is None:
            a.add_argument(
                f"-f{k.replace('_', '-')}",
                *alts,
                **v,
                action="store_true",
                dest=k,
                default=None,
            )
            a.add_argument(
                f"-fno-{k.replace('_', '-')}",
                action="store_false",
                help=argparse.SUPPRESS,
                dest=k,
                default=None,
            )
        else:
            a.add_argument(
                f"-f{k.replace('_', '-')}",
                *alts,
                **v,
                type=type,
                dest=k,
                default=None,
            )

    a.add_argument(
        f"-O",
        type=int,
        help=f"Optimization level from 0 (no optimization) to 3 (aggressive optimization). Defaults to 1.",
        default=2,
        choices=range(len(OPT_CONFIGS)),
        dest="opt_level",
    )
    return a.parse_args()


def main():
    args = parse_args()
    sys.setrecursionlimit(args.recursion_limit)
    if Command(args.command) != Command.lint:
        OPSHIN_LOG_HANDLER.setFormatter(
            logging.Formatter(
                f"%(levelname)s for {args.input_file}:%(lineno)d %(message)s"
            )
        )
        perform_command(args)
    else:
        OPSHIN_LOG_HANDLER.stream = sys.stdout
        if args.output_format_json:
            OPSHIN_LOG_HANDLER.setFormatter(
                logging.Formatter(
                    '{"line":%(lineno)d,"column":%(col_offset)d,"error_class":"%(levelname)s","message":"%(message)s"}'
                )
            )
        else:
            OPSHIN_LOG_HANDLER.setFormatter(
                logging.Formatter(
                    args.input_file
                    + ":%(lineno)d:%(col_offset)d:%(levelname)s: %(message)s"
                )
            )

        try:
            perform_command(args)
        except Exception as e:
            error_class_name = e.__class__.__name__
            message = str(e)
            if isinstance(e, SyntaxError):
                start_line = e.lineno
                pos_in_line = e.offset
                if hasattr(e, "orig_err"):
                    error_class_name = e.orig_err.__class__.__name__
                    message = str(e.orig_err)
            else:
                start_line = 1
                pos_in_line = 1
            if args.output_format_json:
                print(
                    convert_linter_to_json(
                        line=start_line,
                        column=pos_in_line,
                        error_class=error_class_name,
                        message=message,
                    )
                )
            else:
                print(
                    f"{args.input_file}:{start_line}:{pos_in_line}: {error_class_name}: {message}"
                )


def convert_linter_to_json(
    line: int,
    column: int,
    error_class: str,
    message: str,
):
    # output in lists
    return json.dumps(
        [
            {
                "line": line,
                "column": column,
                "error_class": error_class,
                "message": message,
            }
        ]
    )


if __name__ == "__main__":
    main()

1	import inspect	5✔
2
3	import argparse	5✔
4	import io	5✔
5	import logging	5✔
6	import os	5✔
7	import tempfile	5✔
8	from contextlib import redirect_stdout	5✔
9
10	import cbor2	5✔
11	import enum	5✔
12	import importlib	5✔
13	import json	5✔
14	import pathlib	5✔
15	import sys	5✔
16	import typing	5✔
17	import ast	5✔
18
19	import pycardano	5✔
20
21	import uplc	5✔
22	import uplc.ast	5✔
23	from uplc.cost_model import PlutusVersion	5✔
24
25	from . import (	5✔
26	compiler,
27	builder,
28	prelude,
29	__version__,
30	__copyright__,
31	Purpose,
32	PlutusContract,
33	)
34	from .util import CompilerError, data_from_json, OPSHIN_LOG_HANDLER	5✔
35	from .prelude import ScriptContext	5✔
36	from .compiler_config import *	5✔
37	from uplc import cost_model	5✔
38
39
40	class Command(enum.Enum):	5✔
41	compile_pluto = "compile_pluto"	5✔
42	compile = "compile"	5✔
43	eval = "eval"	5✔
44	parse = "parse"	5✔
45	eval_uplc = "eval_uplc"	5✔
46	build = "build"	5✔
47	lint = "lint"	5✔
48
49
50	def parse_uplc_param(param: str):	5✔
51	if param.startswith("{"):	5✔
52	try:	5✔
53	return uplc.ast.data_from_json_dict(json.loads(param))	5✔
54	except json.JSONDecodeError as e:	5✔
55	raise ValueError(	5✔
56	f"Invalid parameter for contract passed, expected JSON value, got {param}"
57	) from e
58	except Exception as e:	×
59	raise ValueError(	×
60	f"Expected parameter for contract to be in valid Plutus data JSON format, got {param}"
61	) from e
62	else:
63	try:	5✔
64	return uplc.ast.data_from_cbor(bytes.fromhex(param))	5✔
65	except Exception as e:	5✔
66	raise ValueError(	5✔
67	"Expected hexadecimal CBOR representation of plutus datum but could not transform hex string to bytes."
68	) from e
69
70
71	def parse_plutus_param(annotation, param: str):	5✔
72	try:	5✔
73	if param.startswith("{"):	5✔
74	try:	5✔
75	param_dict = json.loads(param)	5✔
76	except json.JSONDecodeError as e:	5✔
77	raise ValueError(	5✔
78	f"Invalid parameter for contract passed, expected json value, got {param}"
79	) from e
80	return plutus_data_from_json(annotation, param_dict)	5✔
81	else:
82	try:	5✔
83	param_bytes = bytes.fromhex(param)	5✔
84	except ValueError as e:	5✔
85	raise ValueError(	5✔
86	"Expected hexadecimal CBOR representation of plutus datum but could not transform hex string to bytes."
87	) from e
88	return plutus_data_from_cbor(annotation, param_bytes)	5✔
89	except ValueError as e:	5✔
90	raise ValueError(	5✔
91	f"Could not parse parameter {param} as type {annotation}. Please provide the parameter either as JSON or CBOR (in hexadecimal notation). Detailed error: {e}"
92	) from e
93
94
95	def plutus_data_from_json(annotation: typing.Type, x: dict):	5✔
96	try:	5✔
97	if annotation == int:	5✔
98	return int(x["int"])	5✔
99	if annotation == bytes:	5✔
100	return bytes.fromhex(x["bytes"])	5✔
101	if annotation is None:	5✔
102	return None	5✔
103	if isinstance(annotation, typing._GenericAlias):	5✔
104	# Annotation is a List or Dict
105	if annotation._name == "List":	5✔
106	annotation_ann = annotation.__dict__["__args__"][0]	5✔
107	return [plutus_data_from_json(annotation_ann, k) for k in x["list"]]	5✔
108	if annotation._name == "Dict":	5✔
109	annotation_key, annotation_val = annotation.__dict__["__args__"]	5✔
110	return {	5✔
111	plutus_data_from_json(
112	annotation_key, d["k"]
113	): plutus_data_from_json(annotation_val, d["v"])
114	for d in x["map"]
115	}
116	if annotation.__origin__ == typing.Union:	5!
117	for ann in annotation.__dict__["__args__"]:	5✔
118	try:	5✔
119	return plutus_data_from_json(ann, x)	5✔
120	except (pycardano.DeserializeException, KeyError, ValueError):	5✔
121	pass	5✔
122	raise ValueError(	5✔
123	f"Could not find matching type for {x} in {annotation}"
124	)
125	if annotation == pycardano.Datum:	×
126	if "int" in x:	×
127	return int(x["int"])	×
128	if "bytes" in x:	×
129	return bytes.fromhex(x["bytes"])	×
130	if "constructor" in x:	×
131	return pycardano.RawCBOR(	×
132	uplc.ast.plutus_cbor_dumps(uplc.ast.data_from_json_dict(x))
133	)
134	if "list" in x:	×
135	return [	×
136	plutus_data_from_json(pycardano.Datum, k) for k in x["list"]
137	]
138	if "map" in x:	×
139	return {	×
140	plutus_data_from_json(
141	pycardano.Datum, d["k"]
142	): plutus_data_from_json(pycardano.Datum, d["v"])
143	for d in x["map"]
144	}
145	if issubclass(annotation, pycardano.PlutusData):	5✔
146	return annotation.from_dict(x)	5✔
147	except (KeyError, ValueError):	5✔
148	raise ValueError(	5✔
149	f"Annotation {annotation} does not match provided plutus datum {json.dumps(x)}"
150	)
151
152
153	def plutus_data_from_cbor(annotation: typing.Type, x: bytes):	5✔
154	try:	5✔
155	if annotation in (int, bytes):	5✔
156	res = cbor2.loads(x)	5✔
157	if not isinstance(res, annotation):	5✔
158	raise ValueError(	5✔
159	f"Expected {annotation} but got {type(x)} from {x.hex()}"
160	)
161	return res	5✔
162	if annotation is None:	5✔
163	if not x == cbor2.dumps(None):	5✔
164	raise ValueError(f"Expected None but got {x.hex()}")	5✔
165	return None	5✔
166	if isinstance(annotation, typing._GenericAlias):	5✔
167	# Annotation is a List or Dict
168	if annotation.__origin__ == list:	5✔
169	annotation_ann = annotation.__dict__["__args__"][0]	5✔
170	return [	5✔
171	plutus_data_from_cbor(annotation_ann, cbor2.dumps(k))
172	for k in cbor2.loads(x)
173	]
174	if annotation.__origin__ == dict:	5✔
175	annotation_key, annotation_val = annotation.__dict__["__args__"]	5✔
176	return {	5✔
177	plutus_data_from_cbor(
178	annotation_key, cbor2.dumps(k)
179	): plutus_data_from_cbor(annotation_val, cbor2.dumps(v))
180	for k, v in cbor2.loads(x).items()
181	}
182	if annotation.__origin__ == typing.Union:	5!
183	for ann in annotation.__dict__["__args__"]:	5✔
184	try:	5✔
185	return plutus_data_from_cbor(ann, x)	5✔
186	except (pycardano.DeserializeException, ValueError):	5✔
187	pass	5✔
188	raise ValueError(	5✔
189	f"Could not find matching type for {x.hex()} in {annotation}"
190	)
191	if issubclass(annotation, pycardano.PlutusData):	5✔
192	return annotation.from_cbor(x)	5✔
193	except (KeyError, ValueError):	5✔
194	raise ValueError(	5✔
195	f"Annotation {annotation} does not match provided plutus datum {x.hex()}"
196	)
197
198
199	def check_params(	5✔
200	command: Command,
201	validator_args,
202	return_type,
203	validator_params,
204	):
205	num_onchain_params = 1	5✔
206	onchain_params = validator_args[-num_onchain_params:]	5✔
207	param_types = validator_args[:-num_onchain_params]	5✔
208	if return_type is not None:	5✔
209	print(	5✔
210	f"Warning: validator returns {return_type}, but it is recommended to return None. In PlutusV3, validators that do not return None always fail. This is most likely not what you want."
211	)
212
213	required_onchain_parameters = 1	5✔
214	assert (	5✔
215	len(onchain_params) == required_onchain_parameters
216	), f"""\
217	validator must expect {required_onchain_parameters} parameters at evaluation (on-chain), but was specified to have {len(onchain_params)}.
218	Make sure the validator expects just the script context."""
219
220	if command in (Command.eval, Command.eval_uplc):	5✔
221	assert len(validator_params) == len(param_types) + len(	5✔
222	onchain_params
223	), f"validator expects {len(param_types) + len(onchain_params)} parameters for evaluation, but only got {len(validator_params)}."
224	assert (	5✔
225	onchain_params[-1][1] == ScriptContext
226	), f"Last parameter of the validator has to be ScriptContext, but is {onchain_params[-1][1].__name__} here."
227	return onchain_params, param_types	5✔
228
229
230	def perform_command(args):	5✔
231	# generate the compiler config
232	compiler_config = DEFAULT_CONFIG	5✔
233	compiler_config = compiler_config.update(OPT_CONFIGS[args.opt_level])	5✔
234	overrides = {}	5✔
235	for k in ARGPARSE_ARGS.keys():	5✔
236	if getattr(args, k) is not None:	5!
UNCOV 237	overrides[k] = getattr(args, k)	×
238	compiler_config = compiler_config.update(CompilationConfig(**overrides))	5✔
239	# configure logging
240	if args.verbose:	5!
241	OPSHIN_LOG_HANDLER.setLevel(logging.DEBUG)	×
242	lib = args.lib	5✔
243
244	# execute the command
245	command = Command(args.command)	5✔
246	input_file = args.input_file if args.input_file != "-" else sys.stdin	5✔
247	# read and import the contract
248	with open(input_file, "r") as f:	5✔
249	source_code = f.read()	5✔
250	with tempfile.TemporaryDirectory(prefix="build") as tmpdir:	5✔
251	tmp_input_file = pathlib.Path(tmpdir).joinpath("__tmp_opshin.py")	5✔
252	with tmp_input_file.open("w") as fp:	5✔
253	fp.write(source_code)	5✔
254	sys.path.append(str(pathlib.Path(tmp_input_file).parent.absolute()))	5✔
255	try:	5✔
256	sc = importlib.import_module(pathlib.Path(tmp_input_file).stem)	5✔
257	except Exception as e:	×
258	# replace the traceback with an error pointing to the input file
259	raise SyntaxError(	×
260	f"Could not import the input file as python module. Make sure the input file is valid python code. Error: {e}",
261	) from e
262	sys.path.pop()	5✔
263	# load the passed parameters if not a lib
264	try:	5✔
265	argspec = inspect.signature(sc.validator if lib is None else getattr(sc, lib))	5✔
NEW 266	except AttributeError:	×
NEW 267	raise AssertionError(	×
268	f"Contract has no function called '{'validator' if lib is None else lib}'. Make sure the compiled contract contains one function called 'validator'."
269	)
270	annotations = [	5✔
271	(x.name, x.annotation or prelude.Anything) for x in argspec.parameters.values()
272	]
273	return_annotation = (	5✔
274	argspec.return_annotation
275	if argspec.return_annotation is not argspec.empty
276	else prelude.Anything
277	)
278	parsed_params = []	5✔
279	uplc_params = []	5✔
280	for i, (c, a) in enumerate(zip(annotations, args.args)):	5✔
281	try:	5✔
282	uplc_param = parse_uplc_param(a)	5✔
NEW 283	except ValueError as e:	×
NEW 284	raise ValueError(	×
285	f"Could not parse parameter {i} ('{a}') as UPLC data. Please provide the parameter either as JSON or CBOR (in hexadecimal notation). Detailed error: {e}"
286	) from None
287	uplc_params.append(uplc_param)	5✔
288	try:	5✔
289	param = parse_plutus_param(c[1], a)	5✔
NEW 290	except ValueError as e:	×
NEW 291	raise ValueError(	×
292	f"Could not parse parameter {i} ('{a}') as type {c[1]}. Please provide the parameter either as JSON or CBOR (in hexadecimal notation). Detailed error: {e}"
293	) from None
294	parsed_params.append(param)	5✔
295	if lib is None:	5✔
296	onchain_params, param_types = check_params(	5✔
297	command,
298	annotations,
299	return_annotation,
300	parsed_params,
301	)
302	assert (	5✔
303	onchain_params
304	), "The validator function must have at least one on-chain parameter. You can also add `_:None`."
305
306	py_ret = Command.eval	5✔
307	if command == Command.eval:	5✔
308	print("Python execution started")	5✔
309	with redirect_stdout(open(os.devnull, "w")):	5✔
310	try:	5✔
311	py_ret = sc.validator(*parsed_params)	5✔
312	except Exception as e:	×
313	py_ret = e	×
314	command = Command.eval_uplc	5✔
315
316	source_ast = compiler.parse(source_code, filename=input_file)	5✔
317
318	if command == Command.parse:	5!
319	print("Parsed successfully.")	×
320	return	×
321
322	try:	5✔
323	code = compiler.compile(	5✔
324	source_ast,
325	filename=input_file,
326	validator_function_name="validator" if lib is None else lib,
327	# do not remove dead code when compiling a library - none of the code will be used
328	config=compiler_config,
329	)
UNCOV 330	except CompilerError as c:	×
331	# Generate nice error message from compiler error
UNCOV 332	if not isinstance(c.node, ast.Module):	×
UNCOV 333	source_seg = ast.get_source_segment(source_code, c.node)	×
UNCOV 334	start_line = c.node.lineno - 1	×
UNCOV 335	end_line = start_line + len(source_seg.splitlines())	×
UNCOV 336	source_lines = "\n".join(source_code.splitlines()[start_line:end_line])	×
UNCOV 337	pos_in_line = source_lines.find(source_seg)	×
338	else:
339	start_line = 0	×
340	pos_in_line = 0	×
341	source_lines = source_code.splitlines()[0]	×
342
UNCOV 343	overwrite_syntaxerror = (	×
344	len("SyntaxError: ") * "\b" if command != Command.lint else ""
345	)
UNCOV 346	err = SyntaxError(	×
347	f"""\
348	{overwrite_syntaxerror}{c.orig_err.__class__.__name__}: {c.orig_err}
349	Note that opshin errors may be overly restrictive as they aim to prevent code with unintended consequences.
350	""",
351	(
352	args.input_file,
353	start_line + 1,
354	pos_in_line,
355	source_lines,
356	),
357	# we remove chaining so that users to not see the internal trace back,
358	)
UNCOV 359	err.orig_err = c.orig_err	×
NEW 360	raise err	×
361
362	if command == Command.compile_pluto:	5✔
363	print(code.dumps())	5✔
364	return	5✔
365	code = pluthon.compile(code, config=compiler_config)	5✔
366
367	# apply parameters from the command line to the contract (instantiates parameterized contract!)
368	code = code.term	5✔
369	# UPLC lambdas may only take one argument at a time, so we evaluate by repeatedly applying
370	for d in uplc_params:	5✔
371	code = uplc.ast.Apply(code, d)	5✔
372	code = uplc.ast.Program((1, 0, 0), code)	5✔
373
374	if command == Command.compile:	5✔
375	print(code.dumps())	5✔
376	return	5✔
377
378	if command == Command.build:	5✔
379	if lib is not None:	5!
NEW 380	raise ValueError(	×
381	"Cannot build a library. Please remove the --lib flag when building a contract."
382	)
383	if args.output_directory == "":	5!
384	if args.input_file == "-":	5!
385	print(	×
386	"Please supply an output directory if no input file is specified."
387	)
388	exit(-1)	×
389	target_dir = pathlib.Path("build") / pathlib.Path(input_file).stem	5✔
390	else:
391	target_dir = pathlib.Path(args.output_directory)	×
392	built_code = builder._build(code)	5✔
393	script_arts = PlutusContract(	5✔
394	built_code,
395	datum_type=onchain_params[0] if len(onchain_params) == 3 else None,
396	redeemer_type=onchain_params[1 if len(onchain_params) == 3 else 0],
397	parameter_types=param_types,
398	purpose=(Purpose.any,),
399	title=pathlib.Path(input_file).stem,
400	)
401	script_arts.dump(target_dir)	5✔
402
403	print(f"Wrote script artifacts to {target_dir}/")	5✔
404	return	5✔
405	if command == Command.eval_uplc:	5✔
406	print("UPLC execution started")	5✔
407	assert isinstance(code, uplc.ast.Program)	5✔
408	raw_ret = uplc.eval(	5✔
409	code,
410	cek_machine_cost_model=cost_model.default_cek_machine_cost_model_plutus_v3(),
411	builtin_cost_model=cost_model.default_builtin_cost_model_plutus_v3(),
412	)
413	print("------LOGS--------")	5✔
414	if raw_ret.logs:	5!
415	for log in raw_ret.logs:	×
416	print(" > " + log)	×
417	else:
418	print("No logs")	5✔
419	print("------COST--------")	5✔
420	print(f"CPU: {raw_ret.cost.cpu} \| MEM: {raw_ret.cost.memory}")	5✔
421	if isinstance(raw_ret.result, Exception):	5!
422	print("----EXCEPTION-----")	×
423	ret = raw_ret.result	×
424	else:
425	print("-----SUCCESS------")	5✔
426	ret = uplc.dumps(raw_ret.result)	5✔
427	print(str(ret), end="")	5✔
428	if not isinstance(py_ret, Command):	5✔
429	print(" (Python: " + str(py_ret) + ")", end="")	5✔
430	print()	5✔
431
432
433	def parse_args():	5✔
434	a = argparse.ArgumentParser(	5✔
435	description="An evaluator and compiler from python into UPLC. Translate imperative programs into functional quasi-assembly. Flags allow setting fine-grained compiler options. All flags can be turned off via -fno-<flag>."
436	)
437	a.add_argument(	5✔
438	"command",
439	type=str,
440	choices=Command.__members__.keys(),
441	help="The command to execute on the input file.",
442	)
443	a.add_argument(	5✔
444	"input_file", type=str, help="The input program to parse. Set to - for stdin."
445	)
446	a.add_argument(	5✔
447	"--lib",
448	const="validator",
449	default=None,
450	nargs="?",
451	type=str,
452	help="Indicates that the input file should compile to a generic function, reusable by other contracts (not a smart contract itself). Discards corresponding typechecks. An optional name of the function to export can be given, by default it is validator. Use -fwrap_input and -fwrap_output to control whether the function expects or returns PlutusData (otherwise BuiltIn).",
453	)
454	a.add_argument(	5✔
455	"-o",
456	"--output-directory",
457	default="",
458	type=str,
459	help="The output directory for artefacts of the build command. Defaults to the filename of the compiled contract. of the compiled contract.",
460	)
461	a.add_argument(	5✔
462	"args",
463	nargs="*",
464	default=[],
465	help="Input parameters for the validator (parameterizes the contract for compile/build). Either json or CBOR notation.",
466	)
467	a.add_argument(	5✔
468	"--output-format-json",
469	action="store_true",
470	help="Changes the output of the Linter to a json format.",
471	)
472	a.add_argument(	5✔
473	"--version",
474	action="version",
475	version=f"opshin {__version__} {__copyright__}",
476	)
477	a.add_argument(	5✔
478	"-v",
479	"--verbose",
480	action="store_true",
481	help="Enable verbose logging.",
482	)
483	a.add_argument(	5✔
484	"--recursion-limit",
485	default=max(sys.getrecursionlimit(), 4000),
486	help="Modify the recursion limit (necessary for larger UPLC programs)",
487	type=int,
488	)
489	for k, v in ARGPARSE_ARGS.items():	5✔
490	alts = v.pop("__alts__", [])	5✔
491	type = v.pop("type", None)	5✔
492	if type is None:	5✔
493	a.add_argument(	5✔
494	f"-f{k.replace('_', '-')}",
495	*alts,
496	**v,
497	action="store_true",
498	dest=k,
499	default=None,
500	)
501	a.add_argument(	5✔
502	f"-fno-{k.replace('_', '-')}",
503	action="store_false",
504	help=argparse.SUPPRESS,
505	dest=k,
506	default=None,
507	)
508	else:
509	a.add_argument(	5✔
510	f"-f{k.replace('_', '-')}",
511	*alts,
512	**v,
513	type=type,
514	dest=k,
515	default=None,
516	)
517
518	a.add_argument(	5✔
519	f"-O",
520	type=int,
521	help=f"Optimization level from 0 (no optimization) to 3 (aggressive optimization). Defaults to 1.",
522	default=2,
523	choices=range(len(OPT_CONFIGS)),
524	dest="opt_level",
525	)
526	return a.parse_args()	5✔
527
528
529	def main():	5✔
530	args = parse_args()	5✔
531	sys.setrecursionlimit(args.recursion_limit)	5✔
532	if Command(args.command) != Command.lint:	5✔
533	OPSHIN_LOG_HANDLER.setFormatter(	5✔
534	logging.Formatter(
535	f"%(levelname)s for {args.input_file}:%(lineno)d %(message)s"
536	)
537	)
538	perform_command(args)	5✔
539	else:
540	OPSHIN_LOG_HANDLER.stream = sys.stdout	5✔
541	if args.output_format_json:	5✔
542	OPSHIN_LOG_HANDLER.setFormatter(	5✔
543	logging.Formatter(
544	'{"line":%(lineno)d,"column":%(col_offset)d,"error_class":"%(levelname)s","message":"%(message)s"}'
545	)
546	)
547	else:
548	OPSHIN_LOG_HANDLER.setFormatter(	5✔
549	logging.Formatter(
550	args.input_file
551	+ ":%(lineno)d:%(col_offset)d:%(levelname)s: %(message)s"
552	)
553	)
554
555	try:	5✔
556	perform_command(args)	5✔
557	except Exception as e:	5✔
558	error_class_name = e.__class__.__name__	5✔
559	message = str(e)	5✔
560	if isinstance(e, SyntaxError):	5!
UNCOV 561	start_line = e.lineno	×
UNCOV 562	pos_in_line = e.offset	×
UNCOV 563	if hasattr(e, "orig_err"):	×
UNCOV 564	error_class_name = e.orig_err.__class__.__name__	×
UNCOV 565	message = str(e.orig_err)	×
566	else:
567	start_line = 1	5✔
568	pos_in_line = 1	5✔
569	if args.output_format_json:	5✔
570	print(	5✔
571	convert_linter_to_json(
572	line=start_line,
573	column=pos_in_line,
574	error_class=error_class_name,
575	message=message,
576	)
577	)
578	else:
579	print(	5✔
580	f"{args.input_file}:{start_line}:{pos_in_line}: {error_class_name}: {message}"
581	)
582
583
584	def convert_linter_to_json(	5✔
585	line: int,
586	column: int,
587	error_class: str,
588	message: str,
589	):
590	# output in lists
591	return json.dumps(	5✔
592	[
593	{
594	"line": line,
595	"column": column,
596	"error_class": error_class,
597	"message": message,
598	}
599	]
600	)
601
602
603	if __name__ == "__main__":
604	main()

OpShin / opshin / 18375768062

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