generated from dopt-python/py311-cython
229 lines
5.7 KiB
Python
229 lines
5.7 KiB
Python
"""placeholder module for compilation"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import hashlib
|
|
import struct
|
|
import time
|
|
|
|
import cython
|
|
import dopt_basics.datetime
|
|
|
|
# from cython.cimports.cpython.ref import Py_INCREF, PyObject
|
|
|
|
PyStringC = cython.struct(
|
|
ptr=cython.p_char,
|
|
length=cython.Py_ssize_t,
|
|
)
|
|
|
|
|
|
@cython.cfunc
|
|
def PyStringC_to_unicode(
|
|
string: PyStringC,
|
|
) -> str:
|
|
return string.ptr[: string.length].decode("UTF-8", "strict")
|
|
|
|
|
|
@cython.cfunc
|
|
def PyStringC_to_bytes(
|
|
string: PyStringC,
|
|
) -> bytes:
|
|
return cython.cast(bytes, string.ptr[: string.length])
|
|
|
|
|
|
@cython.cfunc
|
|
def timestamp_to_datetime(
|
|
ts: cython.double,
|
|
) -> datetime.datetime:
|
|
return datetime.datetime.fromtimestamp(ts, dopt_basics.datetime.TIMEZONE_UTC)
|
|
|
|
|
|
@cython.cclass
|
|
class StringHolder:
|
|
c = cython.declare(PyStringC, visibility="readonly")
|
|
_py_data = cython.declare(object, visibility="private")
|
|
|
|
def __cinit__(self, data: str):
|
|
"""accepting Python str
|
|
|
|
Parameters
|
|
----------
|
|
data : str
|
|
_description_
|
|
"""
|
|
if not isinstance(data, str):
|
|
raise TypeError("Data not 'str'")
|
|
|
|
self._py_data = data.encode("UTF-8")
|
|
self.c = PyStringC(self._py_data, len(self._py_data))
|
|
|
|
@property
|
|
def py_string(self):
|
|
return PyStringC_to_unicode(self.c)
|
|
|
|
@property
|
|
def py_bytes(self):
|
|
return PyStringC_to_bytes(self.c)
|
|
|
|
|
|
# BlockC = cython.struct(
|
|
# index=cython.ulonglong,
|
|
# timestamp=cython.double,
|
|
# data=PyStringC,
|
|
# previous_hash=PyStringC,
|
|
# nonce=cython.ulonglong,
|
|
# )
|
|
|
|
|
|
@cython.cfunc
|
|
def float_to_bytes(num: cython.double):
|
|
return struct.pack(">d", num)
|
|
|
|
|
|
@cython.cclass
|
|
class Block:
|
|
_index = cython.declare(cython.ulong, visibility="private")
|
|
_timestamp = cython.declare(cython.double, visibility="private")
|
|
_data = cython.declare(StringHolder, visibility="private")
|
|
_prev_hash = cython.declare(StringHolder, visibility="private")
|
|
_nonce = cython.declare(cython.ulong, visibility="private")
|
|
_hash = cython.declare(StringHolder, visibility="private")
|
|
|
|
def __init__(
|
|
self,
|
|
index: int,
|
|
data: str,
|
|
previous_hash: str,
|
|
nonce: int,
|
|
):
|
|
self._index = index
|
|
# self._timestamp = dopt_basics.datetime.current_time_tz().timestamp()
|
|
self._timestamp = datetime.datetime(2025, 12, 1, 12, 0, 0).timestamp()
|
|
self._data = StringHolder.__new__(StringHolder, data)
|
|
self._prev_hash = StringHolder.__new__(StringHolder, previous_hash)
|
|
self._nonce = nonce
|
|
self._hash = StringHolder.__new__(StringHolder, "")
|
|
|
|
@cython.ccall
|
|
def _perform_hash(self):
|
|
parts = bytearray()
|
|
parts.extend(self._index.to_bytes(8, "big"))
|
|
parts.extend(float_to_bytes(self._timestamp))
|
|
parts.extend(self._data.py_bytes)
|
|
parts.extend(self._prev_hash.py_bytes)
|
|
parts.extend(self._nonce.to_bytes(8, "big"))
|
|
|
|
return hashlib.sha256(parts)
|
|
|
|
def compute_hash_bytes(self):
|
|
return self._perform_hash().digest()
|
|
|
|
def compute_hash(self):
|
|
return self._perform_hash().hexdigest()
|
|
|
|
# Python public API
|
|
@property
|
|
def index(self):
|
|
return self._index
|
|
|
|
@property
|
|
def timestamp(self):
|
|
return timestamp_to_datetime(self._timestamp)
|
|
|
|
@property
|
|
def data(self):
|
|
return self._data.py_string
|
|
|
|
@property
|
|
def prev_hash(self):
|
|
return self._prev_hash.py_string
|
|
|
|
@property
|
|
def nonce(self):
|
|
return self._nonce
|
|
|
|
@nonce.setter
|
|
def nonce(self, value: int):
|
|
self._nonce = value
|
|
|
|
@property
|
|
def hash(self):
|
|
return self._hash.py_string
|
|
|
|
@hash.setter
|
|
def hash(self, value: str):
|
|
if not isinstance(value, str):
|
|
raise TypeError("No string")
|
|
self._hash = StringHolder.__new__(StringHolder, value)
|
|
|
|
|
|
@cython.cclass
|
|
class Blockchain:
|
|
_difficulty = cython.declare(cython.int, visibility="private")
|
|
_index = cython.declare(cython.Py_ssize_t, visibility="private")
|
|
chain = cython.declare(list[Block], visibility="public")
|
|
|
|
def __cinit__(self):
|
|
self._index = 0
|
|
|
|
def __init__(
|
|
self,
|
|
difficulty: int = 1,
|
|
) -> None:
|
|
self.chain: list[Block] = []
|
|
# self.difficulty = difficulty.to_bytes(8, "big")
|
|
self._difficulty = difficulty
|
|
|
|
@property
|
|
def index(self):
|
|
return self._index
|
|
|
|
@property
|
|
def difficulty(self):
|
|
return self._difficulty
|
|
|
|
@cython.ccall
|
|
def create_genesis_block(self):
|
|
genesis = Block(
|
|
index=0,
|
|
data="Genesis Block",
|
|
previous_hash="0",
|
|
nonce=0,
|
|
)
|
|
genesis.hash = genesis.compute_hash()
|
|
self.chain.append(genesis)
|
|
|
|
@cython.ccall
|
|
def proof_of_work(
|
|
self,
|
|
block: Block,
|
|
) -> str:
|
|
prefix = "0" * self._difficulty
|
|
while True:
|
|
block_hash = block.compute_hash()
|
|
if block_hash.startswith(prefix):
|
|
return block_hash
|
|
block._nonce += 1
|
|
|
|
@cython.ccall
|
|
def add_block(self, data: str) -> Block:
|
|
prev_hash = self.chain[self._index].hash
|
|
new_block = Block(
|
|
index=(self._index + 1),
|
|
data=data,
|
|
previous_hash=prev_hash,
|
|
nonce=0,
|
|
)
|
|
|
|
# start = time.perf_counter()
|
|
new_block.hash = self.proof_of_work(new_block)
|
|
# elapsed = time.perf_counter() - start
|
|
|
|
self.chain.append(new_block)
|
|
self._index += 1
|
|
print(f"Mined block {new_block.index} with nonce={new_block.nonce}")
|
|
# print(f"Mined block {new_block.index} in {elapsed:.3f}s with nonce={new_block.nonce}")
|
|
|
|
return new_block
|