mpc_generator.cpython-311.pyc•17.4 kB
�
�5�g�) � �` � d dl mZmZ d dlmZ d dlZd dlZddlmZm Z m
Z
G d� d� � ZdS )� )�Dict�List)�PathN� )�FunctionDefinition�FunctionParameter�FunctionTypec � � e Zd Zdededefd�Zd� Zd� Zd� Z d� Z
d efd
�Zd edefd�Z
d edefd
�Zd edefd�Zdedefd�Zd edefd�Zd edefd�Zd edefd�Zd edefd�Zd edefd�Zd� Zd� Zdefd�Zdefd�Zdefd�ZdS )�MPCGenerator�analysis�
output_dir�
contract_namec �0 � || _ || _ || _ dS )z7Initialize the MPC generator with ABI analysis results.N)r r
r )�selfr r
r s �9/Users/arjun/repos/sparkmango/mpc_server/mpc_generator.py�__init__zMPCGenerator.__init__ s � � ��
�$���*����� c �� � | � � � | � � � | � � � | � � � | � � � dS )z'Generate the MPC server implementation.N)�_create_directory_structure�_generate_server_file�_generate_methods�_generate_state_variables�_generate_documentation)r s r �generatezMPCGenerator.generate sl � �
�(�(�*�*�*�
�"�"�$�$�$�
��� � � �
�&�&�(�(�(�
�$�$�&�&�&�&�&r c � � | j | j dz | j dz | j dz | j dz g}|D ]}|� d�� � �dS )z<Create the necessary directory structure for the MPC server.�methods�state�tests�docsT)�exist_okN)r
�mkdir)r �directories� directorys r r z(MPCGenerator._create_directory_structure sj � �
�O��O�i�'��O�g�%��O�g�%��O�f�$�
�� %� +� +�I��O�O�T�O�*�*�*�*� +� +r c �� � d| j � d| j � d�}t | j | j � � � � d�z d� � 5 }|� |� � ddd� � dS # 1 swxY w Y dS )z"Generate the main MPC server file.z�from mpyc import asyncoro
from mpyc.runtime import mpc
import asyncio
from typing import Dict, Any
from methods import *
from state import State
class a� MPCServer:
def __init__(self):
"""Initialize the MPC server."""
self.mpc = mpc
self.state = State()
async def start(self, port: int = 8000):
"""Start the MPC server."""
await self.mpc.start()
async def stop(self):
"""Stop the MPC server."""
await self.mpc.shutdown()
async def process_call(self, method: str, params: Dict[str, Any]):
"""Process an incoming method call."""
if not hasattr(self, method):
raise ValueError(f"Method {method} not found")
func = getattr(self, method)
result = await func(**params)
return result
if __name__ == '__main__':
server = z,MPCServer()
asyncio.run(server.start())
z
_server.py�wN)r �openr
�lower�write�r �template�fs r r z"MPCGenerator._generate_server_file, s� � �!� ��!� !�>