Spaces:
Sleeping
Sleeping
File size: 2,001 Bytes
e2097a5 d2a752d 555b5c6 d2a752d 555b5c6 b9f0f1e d2a752d 555b5c6 d2a752d 555b5c6 d2a752d 555b5c6 d2a752d b9f0f1e |
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 |
from components.functions import NetworkResolver , HeliusAPI ,InfuraRPC , EthereumSigner ,SolanaSigner,EthereumPayment,SolanaPayment
from components.ucg_g1 import UGC_GRAPH_1
def PayloadGen(address):
payload=f"PAYLOAD:{address}"
return payload
TOOLS = {
"NetworkResolver": lambda state: NetworkResolver(state),
"InfuraRPC": lambda state: InfuraRPC(state),
"HeliusAPI": lambda state: HeliusAPI(state),
"PayloadGenerator": lambda state: PayloadGen(state),
"EthereumSigner": lambda state: EthereumSigner(state),
"SolanaSigner": lambda state: SolanaSigner(state),
"EthereumPayment": lambda state: EthereumPayment(state),
"SolanaPayment": lambda state: SolanaPayment(state)
}
class UCGEngine:
def __init__(self, ugc_graph, tools):
self.ops = {op["operation"]: op for op in ugc_graph}
self.tools = tools
self.executed = set()
def run(self, operation,state):
self.state = state
#initial facts
self.executed = set()
self._execute(operation)
return self.state
def _execute(self, op_name):
print(op_name)
if op_name in self.executed:
return
op = self.ops[op_name]
if "depends" in op:
self._execute(op["depends"])
if "when" in op:
if not eval(op["when"], {}, self.state):
return
if op_name in self.tools:
new_facts=self.tools[op_name](self.state)
new_facts =self.tools[op_name](self.state)
print(new_facts)
self.state.update(new_facts)
self.executed.add(op_name)
#runner=UCGEngine(ugc_graph=UGC_GRAPH_1,tools=TOOLS)
#out=runner.run("EthereumPayment", {"address": "0xA34A13e95CE831953e598689e864a97B7DE949eb","to":"0xA34A13e95CE831953e598689e864a97B7DE949eb","amount":"0.5"})
#print(out)
|