Spaces:
Sleeping
Sleeping
| 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) | |