more

clvm/CLVMObject.py

@@ -44,7 +44,9 @@ def __new__(
                 raise ValueError("tuples must be of size 2, cannot create CLVMObject from: %s" % str(v))
             self.pair = v
             self.atom = None
-        else:
+        elif isinstance(v, bytes):
             self.atom = v
             self.pair = None
+        else:
+            raise ValueError(f"cannot create CLVMObject from: {v!r}")
         return self

clvm/SExp.py

@@ -178,18 +178,18 @@ def as_bin(self) -> bytes:
         return f.getvalue()
 
     @classmethod
-    def to(class_, v: CastableType) -> "SExp":
-        if isinstance(v, class_):
+    def to(cls: typing.Type[_T_SExp], v: CastableType) -> _T_SExp:
+        if isinstance(v, cls):
             return v
 
         if looks_like_clvm_object(v):
             # TODO: maybe this can be done more cleanly
-            return class_(typing.cast(CLVMObjectLike, v))
+            return cls(typing.cast(CLVMObjectLike, v))
 
         # this will lazily convert elements
-        return class_(to_sexp_type(v))
+        return cls(to_sexp_type(v))
 
-    def cons(self: _T_SExp, right) -> _T_SExp:
+    def cons(self: _T_SExp, right: _T_SExp) -> _T_SExp:
         return self.to((self, right))
 
     def first(self: _T_SExp) -> _T_SExp:
@@ -214,9 +214,9 @@ def as_iter(self: _T_SExp) -> typing.Iterable[_T_SExp]:
             yield v.first()
             v = v.rest()
 
-    def __eq__(self, other: CastableType) -> bool:
+    def __eq__(self, other: object) -> bool:
         try:
-            other = self.to(other)
+            other = self.to(typing.cast(CastableType, other))
             to_compare_stack = [(self, other)]
             while to_compare_stack:
                 s1, s2 = to_compare_stack.pop()

clvm/as_python.py

@@ -1,19 +1,27 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import Callable, List, Tuple, TYPE_CHECKING, Union
 
 if TYPE_CHECKING:
     from clvm.SExp import SExp
 
+OpCallable = Callable[["OpStackType", "ValStackType"], None]
+
+ValStackType = List[SExp]
+OpStackType = List[OpCallable]
+
+# TODO: hum...
+PythonType = Union[int, bytes, str, List["PythonType"], Tuple["PythonType", "PythonType"]]
+
 
 def as_python(sexp: SExp):
-    def _roll(op_stack, val_stack):
+    def _roll(op_stack: OpStackType, value_stack: ValStackType) -> None:
         v1 = val_stack.pop()
         v2 = val_stack.pop()
         val_stack.append(v1)
         val_stack.append(v2)
 
-    def _make_tuple(op_stack, val_stack):
+    def _make_tuple(op_stack: OpStackType, value_stack: ValStackType) -> None:
         left = val_stack.pop()
         right = val_stack.pop()
         if right == b"":
@@ -24,7 +32,7 @@ def _make_tuple(op_stack, val_stack):
         else:
             val_stack.append((left, right))
 
-    def _as_python(op_stack, val_stack):
+    def _as_python(op_stack: OpStackType, value_stack: ValStackType) -> None:
         t = val_stack.pop()
         pair = t.as_pair()
         if pair:
@@ -36,10 +44,10 @@ def _as_python(op_stack, val_stack):
             val_stack.append(left)
             val_stack.append(right)
         else:
-            val_stack.append(t.as_atom())
+            val_stack.append(t.atom)
 
-    op_stack = [_as_python]
-    val_stack = [sexp]
+    op_stack: OpStackType = [_as_python]
+    val_stack: ValStackType = [sexp]
     while op_stack:
         op_f = op_stack.pop()
         op_f(op_stack, val_stack)

clvm/core_ops.py

@@ -46,7 +46,7 @@ def op_rest(args: _T_SExp) -> Tuple[int, _T_SExp]:
     return REST_COST, args.first().rest()
 
 
-def op_listp(args: _T_SExp) -> Tuple[int, _T_SExp]:
+def op_listp(args: _T_SExp) -> Tuple[int, SExp]:
     if args.list_len() != 1:
         raise EvalError("l takes exactly 1 argument", args)
     return LISTP_COST, args.true if args.first().listp() else args.false
@@ -59,15 +59,17 @@ def op_raise(args: _T_SExp) -> Tuple[int, _T_SExp]:
         raise EvalError("clvm raise", args)
 
 
-def op_eq(args: _T_SExp) -> Tuple[int, _T_SExp]:
+def op_eq(args: _T_SExp) -> Tuple[int, SExp]:
     if args.list_len() != 2:
         raise EvalError("= takes exactly 2 arguments", args)
     a0 = args.first()
     a1 = args.rest().first()
     if a0.pair or a1.pair:
         raise EvalError("= on list", a0 if a0.pair else a1)
     b0 = a0.as_atom()
+    assert b0 is not None
     b1 = a1.as_atom()
+    assert b1 is not None
     cost = EQ_BASE_COST
     cost += (len(b0) + len(b1)) * EQ_COST_PER_BYTE
     return cost, (args.true if b0 == b1 else args.false)

clvm/more_ops.py

@@ -224,7 +224,7 @@ def op_gr_bytes(args: SExp) -> typing.Tuple[int, SExp]:
     return cost, args.true if b0 > b1 else args.false
 
 
-def op_pubkey_for_exp(args: _T_SExp) -> typing.Tuple[_T_SExp, _T_SExp]:
+def op_pubkey_for_exp(args: _T_SExp) -> typing.Tuple[int, _T_SExp]:
     ((i0, l0),) = args_as_int_list("pubkey_for_exp", args, 1)
     i0 %= 0x73EDA753299D7D483339D80809A1D80553BDA402FFFE5BFEFFFFFFFF00000001
     exponent = PrivateKey.from_bytes(i0.to_bytes(32, "big"))
@@ -258,7 +258,8 @@ def op_strlen(args: _T_SExp) -> typing.Tuple[int, _T_SExp]:
     a0 = args.first()
     if a0.pair:
         raise EvalError("strlen on list", a0)
-    size = len(a0.as_atom())
+    assert a0.atom is not None
+    size = len(a0.atom)
     cost = STRLEN_BASE_COST + size * STRLEN_COST_PER_BYTE
     return malloc_cost(cost, args.to(size))
 
@@ -272,6 +273,7 @@ def op_substr(args: _T_SExp) -> typing.Tuple[int, _T_SExp]:
         raise EvalError("substr on list", a0)
 
     s0 = a0.as_atom()
+    assert s0 is not None
 
     if arg_count == 2:
         i1, = list(args_as_int32("substr", args.rest()))
@@ -292,7 +294,8 @@ def op_concat(args: _T_SExp) -> typing.Tuple[int, _T_SExp]:
     for arg in args.as_iter():
         if arg.pair:
             raise EvalError("concat on list", arg)
-        s.write(arg.as_atom())
+        assert arg.atom is not None
+        s.write(arg.atom)
         cost += CONCAT_COST_PER_ARG
     r = s.getvalue()
     cost += len(r) * CONCAT_COST_PER_BYTE
@@ -322,6 +325,7 @@ def op_lsh(args: _T_SExp) -> typing.Tuple[int, _T_SExp]:
         raise EvalError("shift too large", args.to(i1))
     # we actually want i0 to be an *unsigned* int
     a0 = args.first().as_atom()
+    assert a0 is not None
     i0 = int.from_bytes(a0, "big", signed=False)
     if i1 >= 0:
         r = i0 << i1
@@ -350,23 +354,23 @@ def binop_reduction(
 
 
 def op_logand(args: _T_SExp) -> typing.Tuple[int, _T_SExp]:
-    def binop(a, b):
+    def binop(a: int, b: int) -> int:
         a &= b
         return a
 
     return binop_reduction("logand", -1, args, binop)
 
 
 def op_logior(args: _T_SExp) -> typing.Tuple[int, _T_SExp]:
-    def binop(a, b):
+    def binop(a: int, b: int) -> int:
         a |= b
         return a
 
     return binop_reduction("logior", 0, args, binop)
 
 
 def op_logxor(args: _T_SExp) -> typing.Tuple[int, _T_SExp]:
-    def binop(a, b):
+    def binop(a: int, b: int) -> int:
         a ^= b
         return a
 
@@ -411,7 +415,7 @@ def op_all(args: _T_SExp) -> typing.Tuple[int, _T_SExp]:
     return cost, args.to(r)
 
 
-def op_softfork(args: SExp) -> typing.Tuple[int, bool]:
+def op_softfork(args: SExp) -> typing.Tuple[int, SExp]:
     if args.list_len() < 1:
         raise EvalError("softfork takes at least 1 argument", args)
     a = args.first()

clvm/operators.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Dict, Tuple, Type, TypeVar
+from typing import Dict, Iterator, Optional, Tuple, Type, TypeVar
 
 from typing_extensions import Protocol
 
@@ -28,22 +28,16 @@
 KEYWORDS = (
     # core opcodes 0x01-x08
     ". q a i c f r l x "
-
     # opcodes on atoms as strings 0x09-0x0f
     "= >s sha256 substr strlen concat . "
-
     # opcodes on atoms as ints 0x10-0x17
     "+ - * / divmod > ash lsh "
-
     # opcodes on atoms as vectors of bools 0x18-0x1c
     "logand logior logxor lognot . "
-
     # opcodes for bls 1381 0x1d-0x1f
     "point_add pubkey_for_exp . "
-
     # bool opcodes 0x20-0x23
     "not any all . "
-
     # misc 0x24
     "softfork "
 ).split()
@@ -68,11 +62,12 @@
 }
 
 
-def args_len(op_name, args: SExp):
+def args_len(op_name: str, args: SExp) -> Iterator[int]:
     for arg in args.as_iter():
         if arg.pair:
             raise EvalError("%s requires int args" % op_name, arg)
-        yield len(arg.as_atom())
+        assert arg.atom is not None
+        yield len(arg.atom)
 
 
 # unknown ops are reserved if they start with 0xffff
@@ -102,6 +97,7 @@ def args_len(op_name, args: SExp):
 # this means that unknown ops where cost_function is 1, 2, or 3, may still be
 # fatal errors if the arguments passed are not atoms.
 
+
 def default_unknown_op(op: bytes, args: SExp) -> Tuple[int, SExp]:
     # any opcode starting with ffff is reserved (i.e. fatal error)
     # opcodes are not allowed to be empty
@@ -158,6 +154,7 @@ def default_unknown_op(op: bytes, args: SExp) -> Tuple[int, SExp]:
             if arg.pair:
                 raise EvalError("unknown op on list", arg)
             cost += CONCAT_COST_PER_ARG
+            assert arg.atom is not None
             length += len(arg.atom)
         cost += length * CONCAT_COST_PER_BYTE
 
@@ -169,7 +166,13 @@ def default_unknown_op(op: bytes, args: SExp) -> Tuple[int, SExp]:
 
 
 class OperatorProtocol(Protocol):
-    def __call__(self, op: bytes, args: SExp) -> Tuple[int, SExp]: ...
+    def __call__(self, args: SExp) -> Tuple[int, SExp]:
+        ...
+
+
+class UnknownOperatorProtocol(Protocol):
+    def __call__(self, op: bytes, args: SExp) -> Tuple[int, SExp]:
+        ...
 
 
 _T_OperatorDict = TypeVar("_T_OperatorDict", bound="OperatorDict")
@@ -181,25 +184,41 @@ class OperatorDict(Dict[bytes, OperatorProtocol]):
     operators can be added dynamically.
     """
 
-    unknown_op_handler: OperatorProtocol
-    quote_atom: int
-    apply_atom: int
-
-    # TODO: how do you create an instance if that requires passing in an instance?
-    def __new__(cls: Type[_T_OperatorDict], d: Dict[bytes, OperatorProtocol], *args: object, **kwargs) -> _T_OperatorDict:
+    unknown_op_handler: UnknownOperatorProtocol
+    quote_atom: bytes
+    apply_atom: bytes
+
+    # TODO: can we remove the args and kwargs?
+    # TODO: hint the overloads
+    def __new__(
+        cls: Type[_T_OperatorDict],
+        d: Dict[bytes, OperatorProtocol],
+        *args: object,
+        quote: Optional[bytes] = None,
+        apply: Optional[bytes] = None,
+        unknown_op_handler: UnknownOperatorProtocol = default_unknown_op,
+        **kwargs: object,
+    ) -> _T_OperatorDict:
         """
         `quote_atom` and `apply_atom` must be set
         `unknown_op_handler` has a default implementation
         We do not check if quote and apply are distinct
         We do not check if the opcode values for quote and apply exist in the passed-in dict
         """
         self = super().__new__(cls, d)
-        self.quote_atom = kwargs["quote"] if "quote" in kwargs else d.quote_atom
-        self.apply_atom = kwargs["apply"] if "apply" in kwargs else d.apply_atom
-        if "unknown_op_handler" in kwargs:
-            self.unknown_op_handler = kwargs["unknown_op_handler"]
+
+        if quote is None:
+            assert isinstance(d, OperatorDict)
+            self.quote_atom = d.quote_atom
         else:
-            self.unknown_op_handler = default_unknown_op
+            self.quote_atom = quote
+
+        if apply is None:
+            assert isinstance(d, OperatorDict)
+            self.apply_atom = d.apply_atom
+        else:
+            self.apply_atom = apply
+
         return self
 
     def __call__(self, op: bytes, arguments: SExp) -> Tuple[int, SExp]:
@@ -214,6 +233,8 @@ def __call__(self, op: bytes, arguments: SExp) -> Tuple[int, SExp]:
 APPLY_ATOM = KEYWORD_TO_ATOM["a"]
 
 OPERATOR_LOOKUP = OperatorDict(
-    operators_for_module(KEYWORD_TO_ATOM, core_ops, OP_REWRITE), quote=QUOTE_ATOM, apply=APPLY_ATOM
+    operators_for_module(KEYWORD_TO_ATOM, core_ops, OP_REWRITE),
+    quote=QUOTE_ATOM,
+    apply=APPLY_ATOM,
 )
 OPERATOR_LOOKUP.update(operators_for_module(KEYWORD_TO_ATOM, more_ops, OP_REWRITE))