from __future__ import annotations
import importlib
import inspect
import os
import pathlib
import sys
import tempfile
from typing import List
import pkg_resources
from chialisp import compile_clvm as compile_clvm_rust
from chia.types.blockchain_format.program import Program
from chia.types.blockchain_format.serialized_program import SerializedProgram
from chia.util.lock import Lockfile
compile_clvm_py = None
recompile_requested = (
(os.environ.get("CHIA_DEV_COMPILE_CLVM_ON_IMPORT", "") != "") or ("pytest" in sys.modules)
) and os.environ.get("CHIA_DEV_COMPILE_CLVM_DISABLED", None) is None
def translate_path(p_):
p = str(p_)
if os.path.isdir(p):
return p
else:
module_object = importlib.import_module(p)
return os.path.dirname(inspect.getfile(module_object))
if "CLVM_TOOLS" in os.environ:
try:
from clvm_tools.clvmc import compile_clvm as compile_clvm_py_candidate
compile_clvm_py = compile_clvm_py_candidate
finally:
pass
def compile_clvm_in_lock(full_path: pathlib.Path, output: pathlib.Path, search_paths: List[pathlib.Path]):
treated_include_paths = list(map(translate_path, search_paths))
res = compile_clvm_rust(str(full_path), str(output), treated_include_paths)
if "CLVM_TOOLS" in os.environ and os.environ["CLVM_TOOLS"] == "check" and compile_clvm_py is not None:
def sha256file(f):
import hashlib
m = hashlib.sha256()
m.update(open(f).read().strip().encode("utf8"))
return m.hexdigest()
orig = "%s.orig" % output
compile_clvm_py(full_path, orig, search_paths=search_paths)
orig256 = sha256file(orig)
rs256 = sha256file(output)
if orig256 != rs256:
print("Compiled original %s: %s vs rust %s\n" % (full_path, orig256, rs256))
print("Aborting compilation due to mismatch with rust")
assert orig256 == rs256
else:
print("Compilation match %s: %s\n" % (full_path, orig256))
return res
def compile_clvm(full_path: pathlib.Path, output: pathlib.Path, search_paths: List[pathlib.Path] = []):
with Lockfile.create(pathlib.Path(tempfile.gettempdir()) / "clvm_compile" / full_path.name):
compile_clvm_in_lock(full_path, output, search_paths)
def load_serialized_clvm(
clvm_filename, package_or_requirement=__name__, include_standard_libraries: bool = False, recompile: bool = True
) -> SerializedProgram:
hex_filename = f"{clvm_filename}.hex"
if recompile:
try:
if pkg_resources.resource_exists(package_or_requirement, clvm_filename):
full_path = pathlib.Path(pkg_resources.resource_filename(package_or_requirement, clvm_filename))
output = full_path.parent / hex_filename
if not output.exists() or os.stat(full_path).st_mtime > os.stat(output).st_mtime:
search_paths = [full_path.parent]
if include_standard_libraries:
chia_puzzles_path = pathlib.Path(
pkg_resources.resource_filename(__name__, "__init__.py")
).parent
search_paths.append(chia_puzzles_path)
compile_clvm(full_path, output, search_paths=search_paths)
except NotImplementedError:
pass
clvm_hex = pkg_resources.resource_string(package_or_requirement, hex_filename).decode("utf8")
assert len(clvm_hex.strip()) != 0
clvm_blob = bytes.fromhex(clvm_hex)
return SerializedProgram.from_bytes(clvm_blob)
def load_clvm(
clvm_filename,
package_or_requirement=__name__,
include_standard_libraries: bool = False,
recompile: bool = True,
) -> Program:
return Program.from_bytes(
bytes(
load_serialized_clvm(
clvm_filename,
package_or_requirement=package_or_requirement,
include_standard_libraries=include_standard_libraries,
recompile=recompile,
)
)
)
def load_clvm_maybe_recompile(
clvm_filename,
package_or_requirement=__name__,
include_standard_libraries: bool = False,
recompile: bool = recompile_requested,
) -> Program:
return load_clvm(
clvm_filename=clvm_filename,
package_or_requirement=package_or_requirement,
include_standard_libraries=include_standard_libraries,
recompile=recompile,
)
def load_serialized_clvm_maybe_recompile(
clvm_filename,
package_or_requirement=__name__,
include_standard_libraries: bool = False,
recompile: bool = recompile_requested,
) -> SerializedProgram:
return load_serialized_clvm(
clvm_filename=clvm_filename,
package_or_requirement=package_or_requirement,
include_standard_libraries=include_standard_libraries,
recompile=recompile,
)