SimplyPy API Reference

api

get_context()

Fetch the current context used.

Source code in simplypy/api.py

def get_context():
    """
    Fetch the current context used.
    """
    try:
        return _context.get()
    except LookupError:
        raise RuntimeError("Tracker has not been initialized !")

log_param(key_or_dict=None, value=None, **kwargs)

Logs parameters to the current simulation context. Accepts a key/value pair, a dictionary, or keyword arguments.

Source code in simplypy/api.py

def log_param(
    key_or_dict: Optional[Union[str, Dict[str, Any]]] = None,
    value: Optional[Any] = None,
    **kwargs,
):
    """
    Logs parameters to the current simulation context.
    Accepts a key/value pair, a dictionary, or keyword arguments.
    """
    ctx = get_context()

    # CASE 1: The user passed a string key and a value
    # Example: simply.log_param("velocity", 100)
    if isinstance(key_or_dict, str):
        if value is None:
            raise ValueError(
                f"You provided a parameter name '{key_or_dict}' but no value."
            )
        ctx.log_param(key_or_dict, value)

    # CASE 2: The user passed a dictionary
    # Example: simply.log_param({"velocity": 100, "drag": 0.4})
    elif isinstance(key_or_dict, dict):
        for k, v in key_or_dict.items():
            ctx.log_param(k, v)

    # CASE 3: The user passed kwargs
    # Example: simply.log_param(velocity=100, drag=0.4)
    if kwargs:
        for k, v in kwargs.items():
            ctx.log_param(k, v)

    # ERROR CHECK: Did they pass absolutely nothing?
    # Example: simply.log_param()
    if key_or_dict is None and not kwargs:
        raise ValueError(
            "Invalid Entry: You must provide a key/value pair, a dictionary, or keyword arguments."
        )

log_result(id=None, value=None, path=None, save_result=True, **kwargs)

Unified function to log a result value, a dictionary of results, or a file path. If save_result is True for a path, it will create an artefact. If save_result is True for a dic or a key/value par it will add it to the blob. Both case will save a pointer to the db.

Source code in simplypy/api.py

def log_result(
    id: Optional[Union[str, Dict[str, Any]]] = None,
    value: Optional[Any] = None,
    path: Optional[Union[str, Path]] = None,
    save_result: bool = True,
    **kwargs
):
    """
    Unified function to log a result value, a dictionary of results, or a file path.
    If save_result is True for a path, it will create an artefact.
    If save_result is True for a dic or a key/value par it will add it to the blob.
    Both case will save a pointer to the db.
    """
    ctx = get_context()

    """
    First we check for edgecase of the funtion. The user cannot send a key/value pair and a path at the same time.
    The user cannpt send nothing.
    """

    # 1. ERROR CHECK: Ambiguous input
    if value is not None and path is not None:
        raise ValueError(
            "Ambiguous input: You cannot provide both a 'value' and a 'path' at the same time."
        )

    # 2. ERROR CHECK: Missing data
    if value is None and path is None and not isinstance(id, dict) and not kwargs:
        raise ValueError(
            "Missing data: You must provide a 'value', a 'path', or a dictionary of results."
        )

    # 3. CASE: Dictionary of results
    if isinstance(id, dict):
        for k, v in id.items():
            ctx.log_result(k, {"value": v, "path": None, "save_result": save_result})
        return  # Exit early so we don't trigger the other cases

    # 4. CASE: A file path is provided
    if path is not None:
        file_path = Path(path)
        # If the user didn't provide a string ID, automatically use the file name
        result_id = id if isinstance(id, str) else file_path.name

        ctx.log_result(
            result_id,
            {"value": None, "path": str(file_path), "save_result": save_result},
        )
        return

    # 5. CASE: A standard single value is provided
    if isinstance(id, str) and value is not None:
        ctx.log_result(id, {"value": value, "path": None, "save_result": save_result})
        return

    if kwargs:
        for k, v in kwargs.items():
            ctx.log_result(k, {"value": v, "path": None, "save_result": save_result})

    # Catch-all for weird edge cases (e.g., id was an integer)
    else:
        raise ValueError("Invalid input format. The 'id' must be a string or a dictionary.")

set_context(ctx)

Update the context.

Source code in simplypy/api.py

def set_context(ctx):
    """
    Update the context.
    """
    _context.set(ctx)

track(func=None, *, save_result=False, project_name=None, project_path=None)

This is a multi purpose function used to launch the tracker fo a simulation, log a specific user function or do both. It is either a function or a decorator. If the user use it as a track() at the beggining of the simulation, it will create the tracker instance for the run. If used as a decorator it will either create the tracker if it does not exist and log the function in callstack and its IO and some metadata.

Source code in simplypy/api.py

def track(func=None, *, save_result=False, project_name=None, project_path=None):
    """
    This is a multi purpose function used to launch the tracker fo a simulation, log a specific user function or do both.
    It is either a function or a decorator. If the user use it as a track() at the beggining of the simulation, it will create
    the tracker instance for the run. If used as a decorator it will either create the tracker if it does not exist and log the
    function in callstack and its IO and some metadata.
    """
    if project_path is None:
        # Gets the file path of the code that called this __init__ if the user doesn't provide a project_path
        caller_frame = inspect.stack()[1]
        caller_filename = caller_frame.filename
        project_path = Path(caller_filename).parent.resolve()
    else:
        project_path = Path(project_path).resolve()

    if func is None:  # If we don't use it as a decorator
        simulation_tracker = Tracker(project_name, project_path)
        set_context(simulation_tracker)
        return None

    def decorator(func):
        return _decorate(func, project_name, project_path, save_result=save_result)

    return decorator(func)  # If we use it as a decorator

metadata

get_imports()

This function will get all the imports used in a python script and sort them between user imports and system imports and also will differentiate if an import is a library or a script made by the user.

Source code in simplypy/metadata.py

def get_imports():
    '''
    This function will get all the imports used in a python script and sort them between user imports and system imports and also will differentiate
    if an import is a library or a script made by the user.
    '''
    custom_modules = []
    stdlib = sys.stdlib_module_names # Get all modules in the current script

    for name, mod in sys.modules.items():
        path = getattr(mod, "__file__", None)

        if path == None:
            continue
        path = os.path.abspath(path)
        if name in stdlib or"site-packages" in path or "dist-packages" in path: 
            continue
        if not os.path.isfile(path):
            continue

        custom_modules.append(path) # If the module is a users script add it to the custom module list

    return sorted(set(custom_modules)), sys.modules.keys() # Add also the lis of all imported modules

get_main_script()

Get the location of the main script

Source code in simplypy/metadata.py

def get_main_script():
    '''
    Get the location of the main script
    '''
    main_mod = sys.modules.get('__main__')
    if main_mod and hasattr(main_mod, '__file__'):
        return os.path.abspath(main_mod.__file__)
    return None

load_main_script_source()

Get the source code of the main script where the user created the tracker

Source code in simplypy/metadata.py

def load_main_script_source():
    '''
    Get the source code of the main script where the user created the tracker
    '''
    path = get_main_script()
    if not path:
        return None
    try:
        with open(path, "r", encoding="utf-8") as f:
            return f.read()
    except Exception:
        return None

save_custom_sources(custom_modules)

Get the source code of each user imports

Source code in simplypy/metadata.py

def save_custom_sources(custom_modules):
    '''
    Get the source code of each user imports
    '''
    saved = {}
    for path in custom_modules:
        try:
            with open(path, "r", encoding="utf-8") as f:
                saved[path] = f.read()
        except(OSError, UnicodeDecodeError):
            # Add something to the logging system for the future
            continue
    return saved

serverCom

ServerCom

Source code in simplypy/serverCom.py

class ServerCom:
    def __init__(self, project_name, project_path, uuid):
        self.uuid = uuid
        self.command_id = (
            0  # The ID of the current command (increments by 1 every command sent)
        )
        self.project_name = project_name
        self.project_path = project_path
        self.rpc_handler = RPCHandler()

        # Information to connect to the server and communicate with it
        self.IP = "127.0.0.1"
        self.port = 22222
        self.sock = None

    def connect_server(self):
        # Connects to the server
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.connect((self.IP, self.port))

        # Register the run into the server
        register_msg = json.dumps(
            {
                "jsonrpc": self.rpc_handler.JSONRPC_VERSION,
                "method": "REGISTER",
                "params": json.dumps(
                    {
                        "project_name": self.project_name,
                        "project_path": self.project_path,
                        "run_id": self.uuid,
                    }
                ),
                "id": 0,
            }
        )
        self.sock.sendall(register_msg.encode("utf-8"))
        self.command_id += 1  # Increment ID

        # Awaiting for a response from the server
        reply = self.sock.recv(1024)

        # Analysing the decoded response
        decoded = self.rpc_handler.decode_response(reply.decode("utf-8"))
        if decoded["result"] != "ACK":
            raise Exception("Could not register run to the server")

    def dump_run(self):
        """
        This function will tell the server to dump the content of the run to the database.
        """
        command = DumpCmd()
        self.execute_command(command)
        # We then close the connection to the server properly
        self.sock.close()

    def shutdown_server(self):
        command = ShutDownCmd()
        self.execute_command(command)

        # Server should be down, closing connection
        self.sock.close()

    def execute_command(self, p_Command):
        # Connects to the server if not already done
        if self.sock == None:
            self.connect_server()

        # Encode the requested command and sends it to the server
        encoded = self.rpc_handler.encode_request(p_Command, self.command_id, self.uuid)
        self.sock.sendall(encoded.encode("utf-8"))

        self.command_id += 1  # Increment ID

        # Awaiting for a response from the server
        reply = self.sock.recv(1024)

        # Return the decoded response
        try:
            decoded = self.rpc_handler.decode_response(reply.decode("utf-8"))

            return decoded["result"]
        except:
            return None

dump_run()

This function will tell the server to dump the content of the run to the database.

Source code in simplypy/serverCom.py

def dump_run(self):
    """
    This function will tell the server to dump the content of the run to the database.
    """
    command = DumpCmd()
    self.execute_command(command)
    # We then close the connection to the server properly
    self.sock.close()

tracker

Tracker

The tracker is the mirror on the client side of the simulation object on the server. Its tasks during the run of the simulation is to send command to the server from run specific command to user entered command via the api. It will start a connection to the server thanks to the ServerCom object and then will expose its method to the api for the user to use. It will also log some simply specific metadata on the run and send it to the server. (Like what language is used, where is the simulation launch, a profiler)

Source code in simplypy/tracker.py

class Tracker:
    """
    The tracker is the mirror on the client side of the simulation object on the server. Its
    tasks during the run of the simulation is to send command to the server from run specific
    command to user entered command via the api. It will start a connection to the server
    thanks to the ServerCom object and then will expose its method to the api for the user to 
    use. It will also log some simply specific metadata on the run and send it to the server.
    (Like what language is used, where is the simulation launch, a profiler)
    """

    def __init__(self, project_name, project_path=None):

        # Name the project and get its path and assign the uuid
        if project_path is None:
            project_path = os.getcwd()
        self.project_path = Path(project_path).resolve()
        if project_name is None:
            project_name = "Untiteld project"
        self.project_name = project_name
        self.uuid = uuid.uuid4()  # Will need to change to uuid7 in the future

        # Launch the server
        self.server = ServerCom(
            self.project_name, str(self.project_path), str(self.uuid)
        )
        self.server.connect_server()
        self.cwd = os.getcwd()
        self.start_profiler()
        self.callstack = {}
        self.metadata_pysimply()

    def metadata_pysimply(self):
        """
        Send some basic metadata at every run
        """
        custom_modules, sys_modules = get_imports()
        self.start_time = time.time()
        self.metadata = {
            "simply.language": "python",
            "simply.python.cwd": self.cwd,
            "simply.python.main_script_source": load_main_script_source(),
            "simply.python.custom_modules": custom_modules,
            "simply.python.sys_modules": sys_modules,
            "simply.python.source_custom_modules": save_custom_sources(custom_modules),
        }
        # for key, value in self.metadata.items():
        # self.add_metadata(key, value)

    def start_profiler(self):
        """
        Start a python profiler  for the run
        """
        self.profiler = cProfile.Profile()
        self.profiler.enable()
        self.profiler_enabled = True

    def close(self):
        '''
        Method that will be launched when the run is ended. It will save the profiler data and get the runtime. It will also close the connection to the server.
        '''
        self.run_time = time.time() - self.start_time
        self.add_metadata("simply.runtime", self.run_time)

        if self.profiler_enabled:
            self.profiler.disable()
            self.profiler.dump_stats("profiling.txt")
            self.add_metadata("simply.python.cprofiler_dump", "profiling.txt")
        self.add_metadata("simply.python.callstack", self.callstack)
        self.server.dump_run()  # When the simulation ends we ask the server to dump the simulation into the database


    '''
    Here are all the method exposed to the API. They are not redoundant. The goal of the api is to use the contextVar object to call form 
    function call the same tracker object in every part of the code of a run. They thus all point to these method of the tracker object.
    '''

    def add_callstack(self, func_id):
        try:
            self.callstack[func_id] += 1
        except:
            self.callstack[func_id] = 1

    def log_param(self, id, parameter):
        self.server.execute_command(LogParamCmd(id, parameter))

    def log_result(self, id, result):
        self.server.execute_command(LogResultCmd(id, result))

    def add_metadata(self, id, metadata):
        self.server.execute_command(AddMetaDataCmd(id, metadata))

    def add_note(self, note):
        self.server.execute_command(AddNoteCmd("", note))

    def add_tag(self, tag):
        self.server.execute_command(AddTagCmd(tag, True))

close()

Method that will be launched when the run is ended. It will save the profiler data and get the runtime. It will also close the connection to the server.

Source code in simplypy/tracker.py

def close(self):
    '''
    Method that will be launched when the run is ended. It will save the profiler data and get the runtime. It will also close the connection to the server.
    '''
    self.run_time = time.time() - self.start_time
    self.add_metadata("simply.runtime", self.run_time)

    if self.profiler_enabled:
        self.profiler.disable()
        self.profiler.dump_stats("profiling.txt")
        self.add_metadata("simply.python.cprofiler_dump", "profiling.txt")
    self.add_metadata("simply.python.callstack", self.callstack)
    self.server.dump_run()  # When the simulation ends we ask the server to dump the simulation into the database

metadata_pysimply()

Send some basic metadata at every run

Source code in simplypy/tracker.py

def metadata_pysimply(self):
    """
    Send some basic metadata at every run
    """
    custom_modules, sys_modules = get_imports()
    self.start_time = time.time()
    self.metadata = {
        "simply.language": "python",
        "simply.python.cwd": self.cwd,
        "simply.python.main_script_source": load_main_script_source(),
        "simply.python.custom_modules": custom_modules,
        "simply.python.sys_modules": sys_modules,
        "simply.python.source_custom_modules": save_custom_sources(custom_modules),
    }

start_profiler()

Start a python profiler for the run

Source code in simplypy/tracker.py

def start_profiler(self):
    """
    Start a python profiler  for the run
    """
    self.profiler = cProfile.Profile()
    self.profiler.enable()
    self.profiler_enabled = True