-
Notifications
You must be signed in to change notification settings - Fork 2k
/
load_clvm.py
161 lines (128 loc) · 5.57 KB
/
load_clvm.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
from __future__ import annotations
import importlib
import inspect
import os
import pathlib
import sys
import tempfile
from typing import List
import importlib_resources
from clvm_tools_rs 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
here_name = __name__.rpartition(".")[0]
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))
# Handle optional use of python clvm_tools if available and requested
if "CLVM_TOOLS" in os.environ:
from clvm_tools.clvmc import compile_clvm as compile_clvm_py_candidate
compile_clvm_py = compile_clvm_py_candidate
def compile_clvm_in_lock(full_path: pathlib.Path, output: pathlib.Path, search_paths: List[pathlib.Path]):
# Compile using rust (default)
# Ensure path translation is done in the idiomatic way currently
# expected. It can use either a filesystem path or name a python
# module.
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:
# Simple helper to read the compiled output
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(f"Compiled original {full_path}: {orig256} vs rust {rs256}\n")
print("Aborting compilation due to mismatch with rust")
assert orig256 == rs256
else:
print(f"Compilation match {full_path}: {orig256}\n")
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=here_name, include_standard_libraries: bool = True, recompile: bool = True
) -> SerializedProgram:
"""
This function takes a .clsp file in the given package and compiles it to a
.clsp.hex file if the .hex file is missing or older than the .clsp file, then
returns the contents of the .hex file as a `Program`.
clvm_filename: file name
package_or_requirement: usually `__name__` if the clvm file is in the same package
"""
hex_filename = f"{clvm_filename}.hex"
# Set the CHIA_DEV_COMPILE_CLVM_ON_IMPORT environment variable to anything except
# "" or "0" to trigger automatic recompilation of the Chialisp on load.
resources = importlib_resources.files(package_or_requirement)
if recompile and not getattr(sys, "frozen", False):
full_path = resources.joinpath(clvm_filename)
if full_path.exists():
# Establish whether the size is zero on entry
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:
# we can't get the dir, but we can get a file then get its parent.
chia_puzzles_path = pathlib.Path(__file__).parent
search_paths.append(chia_puzzles_path)
compile_clvm(full_path, output, search_paths=search_paths)
clvm_path = resources.joinpath(hex_filename)
clvm_hex = clvm_path.read_text(encoding="utf-8")
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=here_name,
include_standard_libraries: bool = True,
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=here_name,
include_standard_libraries: bool = True,
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=here_name,
include_standard_libraries: bool = True,
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,
)