Server

rigatoni.Server

Server(port, starting_state, delegate_map=None, json_output=None)

Bases: object

Overarching object for managing the state of a NOODLES session

Handles communication and multiple client connections. The server provides several methods for interacting with and creating delegates. These are especially useful for defining custom methods that the server will expose to clients. Can be instatiated normally, or it can be used as a context manager to automatically start and stop the server while running in a new thread.

Attributes:

Name	Type	Description
port	int	port server is running on
clients	set	client connections
ids	dict	maps object type to slot tracking info (next_slot, on_deck)
state	dict	document's current state, contains all components with component ID as the key
client_state	dict	lagging state to keep track of how up-to-date clients are
references	dict	maps component ID to all the component ID's that reference it
delete_queue	set	components that are referenced but have been requested to be deleted
ready	Event	event to signal when server is ready to accept connections
shutdown_event	Event	event to signal when server is shutting down
thread	Thread	thread server is running on if using context manager
byte_server	ByteServer	slot to store reference to server that serves uri bytes
json_output	str	path to json file to output message logs
custom_delegates	dict	maps component type to delegate class
id_map	dict	maps component type to ID type
id_decoder	dict	maps ID type to base component type, useful for getting base class from ID
message_map	dict	maps action and type to message ID

Parameters:

Name	Type	Description	Default
port	int	port to run server on	required
starting_state	list[StartingComponent]	list of objects containing the info to create components on initialization	required
delegate_map	dict	maps noodles component type to instance of delegate class	None
json_output	str	path to json file to output message logs	None

Raises:

Type	Description
TypeError	invalid arguments to create starting component
ValueError	no method specified for method starting component

Source code in rigatoni/core.py

def __init__(self, port: int, starting_state: list[StartingComponent],
             delegate_map: dict[Type[Delegate], Type[Delegate]] = None, json_output: str = None):
    """Constructor

    Args:
        port (int):
            port to run server on
        starting_state (list[StartingComponent]):
            list of objects containing the info to create components on initialization
        delegate_map (dict):
            maps noodles component type to instance of delegate class
        json_output (str):
            path to json file to output message logs
    Raises:
        TypeError: invalid arguments to create starting component
        ValueError: no method specified for method starting component
    """

    self.port = port
    self.clients = set()
    self.ids = {}
    self.state = {}
    self.client_state = {}
    self.references = {}
    self.delete_queue = set()
    self.ready = threading.Event()
    self.shutdown_event = asyncio.Event()
    self.thread = None
    self.byte_server = None
    self.json_output = json_output
    if json_output:
        with open(json_output, "w") as outfile:  # Clear out old contents
            outfile.write("JSON Log\n")

    # Set up id's and custom delegates
    self.custom_delegates = delegate_map if delegate_map else {}
    self.id_map = id_map.copy()
    for old, new in self.custom_delegates.items():
        self.id_map[new] = self.id_map.pop(old)
    self.id_decoder = {val: key for key, val in id_map.items()}

    self.message_map = {
        ("create", MethodID): 0,
        ("delete", MethodID): 1,
        ("create", SignalID): 2,
        ("delete", SignalID): 3,
        ("create", EntityID): 4,
        ("update", EntityID): 5,
        ("delete", EntityID): 6,
        ("create", PlotID): 7,
        ("update", PlotID): 8,
        ("delete", PlotID): 9,
        ("create", BufferID): 10,
        ("delete", BufferID): 11,
        ("create", BufferViewID): 12,
        ("delete", BufferViewID): 13,
        ("create", MaterialID): 14,
        ("update", MaterialID): 15,
        ("delete", MaterialID): 16,
        ("create", ImageID): 17,
        ("delete", ImageID): 18,
        ("create", TextureID): 19,
        ("delete", TextureID): 20,
        ("create", SamplerID): 21,
        ("delete", SamplerID): 22,
        ("create", LightID): 23,
        ("update", LightID): 24,
        ("delete", LightID): 25,
        ("create", GeometryID): 26,
        ("delete", GeometryID): 27,
        ("create", TableID): 28,
        ("update", TableID): 29,
        ("delete", TableID): 30,
        ("update", None): 31,
        ("reset", None): 32,
        ("invoke", None): 33,
        ("reply", None): 34,
        ("initialized", None): 35
    }

    # Set up starting state
    self.state["document"] = Document(server=self, id=ID(slot=0, gen=0))
    self.client_state["document"] = Document(server=self, id=ID(slot=0, gen=0))
    for starting_component in starting_state:
        comp_type = starting_component.type
        comp_method = starting_component.method

        try:
            comp = self.create_component(comp_type, **starting_component.component_attrs)
        except Exception as e:
            raise TypeError(f"Invalid arguments to create {comp_type}: {e}")

        if comp_type == Method:
            if not comp_method:
                raise ValueError("Method not specified for starting method")

            # Create injected method and set it as attribute
            injected = InjectedMethod(self, comp_method)
            setattr(self, comp.name, injected)

            # Add to document context if flag is set
            if starting_component.document:
                self.state["document"].methods_list.append(comp.id)

        elif comp_type == Signal and starting_component.document:
            self.state["document"].signals_list.append(comp.id)

    logging.debug(f"Server initialized with objects: {self.state}")

broadcast

broadcast(message)

Broadcast message to all connected clients

Parameters:

Name	Type	Description	Default
message	[tuple]	fully constructed message in form (tag/id, contents)	required

Source code in rigatoni/core.py

def broadcast(self, message: list):
    """Broadcast message to all connected clients

    Args:
        message [tuple]: fully constructed message in form (tag/id, contents)
    """

    # Log message in json file if applicable
    if self.json_output:
        self._log_json(message)

    logging.debug(f"Broadcasting Message: ID's {message[::2]}")
    encoded = dumps(message)
    websockets.broadcast(self.clients, encoded)

create_buffer

create_buffer(name=None, size=None, inline_bytes=None, uri_bytes=None)

Add a Buffer object for the session. Will use a custom delegate if applicable.

Parameters:

Name	Type	Description	Default
name	str	name of the buffer, defaults to "No-Name	None
size	int	size of the buffer in bytes	None
inline_bytes	bytes	bytes for the buffer	None
uri_bytes	str	uri to get the bytes from the web	None

Returns:

Name	Type	Description
Buffer	Buffer	buffer delegate that was created

Source code in rigatoni/core.py

def create_buffer(self, name: Optional[str] = None,
                  size: int = None,
                  inline_bytes: bytes = None,
                  uri_bytes: str = None) -> Buffer:
    """Add a Buffer object for the session. Will use a custom delegate if applicable.

    Args:
        name (str, optional): name of the buffer, defaults to "No-Name
        size (int, optional): size of the buffer in bytes
        inline_bytes (bytes, optional): bytes for the buffer
        uri_bytes (str, optional): uri to get the bytes from the web

    Returns:
        Buffer: buffer delegate that was created
    """
    return self.create_component(Buffer, name=name, size=size, inline_bytes=inline_bytes, uri_bytes=uri_bytes)

create_bufferview

create_bufferview(source_buffer, offset, length, name=None, type='UNK')

Add a BufferView object to the session. Will use a custom delegate if applicable.

Parameters:

Name	Type	Description	Default
source_buffer	BufferID	buffer that the view is based on	required
offset	int	offset in bytes from the start of the buffer	required
length	int	length of the view in bytes	required
name	str	name of the buffer view	None
type	str	type of the buffer view	'UNK'

Returns:

Name	Type	Description
BufferView	BufferView	buffer view delegate that was created

Source code in rigatoni/core.py

def create_bufferview(self,
                      source_buffer: BufferID,
                      offset: int,
                      length: int,
                      name: Optional[str] = None,
                      type: Literal["UNK", "GEOMETRY", "IMAGE"] = "UNK") -> BufferView:
    """Add a BufferView object to the session. Will use a custom delegate if applicable.

    Args:
        source_buffer (BufferID): buffer that the view is based on
        offset (int): offset in bytes from the start of the buffer
        length (int): length of the view in bytes
        name (str, optional): name of the buffer view
        type (str, optional): type of the buffer view

    Returns:
        BufferView: buffer view delegate that was created
    """
    return self.create_component(BufferView, name=name, source_buffer=source_buffer,
                                 offset=offset, length=length, type=type)

create_component

create_component(comp_type, **kwargs)

Officially create new component in state

This method updates state, updates references, and broadcasts msg to clients. It also handles the acquisition of a valid ID. This is a general creator method, but more specific versions exist for each component type. Keyword arguments should be used for specifying the attributes of the component. Any deviation from the spec will raise a validation exception.

Note

Since this method handles the ID, it should not be specified as one of the keyword arguments.

Parameters:

Name	Type	Description	Default
comp_type	Component Type	type of component to be created	required
**kwargs		the user should specify the attributes of the component using keyword arguments. Refer to the noodle objects to see which attributes are required and optional. Any deviation from the spec will raise a validation exception.	{}

Returns:

Name	Type	Description
Delegate	T	delegate for the newly created component

Raises:

Type	Description
ValueError	if the user specifies an invalid attribute for the component

Source code in rigatoni/core.py

def create_component(self, comp_type: Type[T], **kwargs) -> T:
    """Officially create new component in state

    This method updates state, updates references, and broadcasts msg to clients.
    It also handles the acquisition of a valid ID. This is a general creator method, but
    more specific versions exist for each component type. Keyword arguments should be
    used for specifying the attributes of the component. Any deviation from the spec will
    raise a validation exception.

    !!! note

        Since this method handles the ID, it should not be specified as one of the keyword arguments.

    Args:
        comp_type (Component Type): type of component to be created
        **kwargs: the user should specify the attributes of the component using 
            keyword arguments. Refer to the noodle objects to see which attributes
            are required and optional. Any deviation from the spec will raise a 
            validation exception.

    Returns:
        Delegate: delegate for the newly created component

    Raises:
        ValueError: if the user specifies an invalid attribute for the component
    """

    # Get ID and try to create delegate from args
    comp_type = self.custom_delegates.get(comp_type, comp_type)
    comp_id = self._get_id(comp_type)
    try:
        new_delegate = comp_type(server=self, id=comp_id, **kwargs)
    except Exception as e:
        raise ValueError(f"Args: {kwargs}, invalid for initializing a {comp_type}: {e}")

    # Update state and keep track of initial version for changes / update messages
    self.state[comp_id] = new_delegate
    self.client_state[comp_id] = new_delegate.model_copy()

    # Update references for each component referenced by this one
    self._update_references(new_delegate, new_delegate)

    # Create message and broadcast
    message = self._prepare_message("create", new_delegate)
    self.broadcast(message)

    # Return component or delegate instance if applicable
    return new_delegate

create_entity

create_entity(name, parent=None, transform=None, text_rep=None, web_rep=None, render_rep=None, lights=None, tables=None, plots=None, tags=None, methods_list=None, signals_list=None, influence=None)

Add an Entity object to the session. Will use a custom delegate if applicable.

Parameters:

Name	Type	Description	Default
name	str	name of the entity	required
parent	EntityID	parent entity	None
transform	Mat4	transform for the entity	None
text_rep	TextRepresentation	text representation for the entity	None
web_rep	WebRepresentation	web representation for the entity	None
render_rep	RenderRepresentation	render representation that links to geometry info	None
lights	list[LightID]	list of attached lights	None
tables	list[TableID]	list of attached tables	None
plots	list[PlotID]	list of attached plots	None
tags	list[str]	list of applicable tags	None
methods_list	list[MethodID]	list of methods attached to the entity	None
signals_list	list[SignalID]	list of signals attached to the entity	None
influence	BoundingBox	bounding box for the entity	None

Returns:

Name	Type	Description
Entity	Entity	entity delegate that was created

Source code in rigatoni/core.py

def create_entity(self, name: Optional[str],
                  parent: Optional[EntityID] = None,
                  transform: Optional[Mat4] = None,
                  text_rep: Optional[TextRepresentation] = None,
                  web_rep: Optional[WebRepresentation] = None,
                  render_rep: Optional[RenderRepresentation] = None,
                  lights: Optional[list[LightID]] = None,
                  tables: Optional[list[TableID]] = None,
                  plots: Optional[list[PlotID]] = None,
                  tags: Optional[list[str]] = None,
                  methods_list: Optional[list[MethodID]] = None,
                  signals_list: Optional[list[SignalID]] = None,
                  influence: Optional[BoundingBox] = None) -> Entity:
    """Add an Entity object to the session. Will use a custom delegate if applicable.

    Args:
        name (str): name of the entity
        parent (EntityID, optional): parent entity
        transform (Mat4, optional): transform for the entity
        text_rep (TextRepresentation, optional): text representation for the entity
        web_rep (WebRepresentation, optional): web representation for the entity
        render_rep (RenderRepresentation, optional): render representation that links to geometry info
        lights (list[LightID], optional): list of attached lights
        tables (list[TableID], optional): list of attached tables
        plots (list[PlotID], optional): list of attached plots
        tags (list[str], optional): list of applicable tags
        methods_list (list[MethodID], optional): list of methods attached to the entity
        signals_list (list[SignalID], optional): list of signals attached to the entity
        influence (BoundingBox, optional): bounding box for the entity

    Returns:
        Entity: entity delegate that was created
    """
    return self.create_component(Entity, name=name, parent=parent, transform=transform, text_rep=text_rep,
                                 web_rep=web_rep, render_rep=render_rep, lights=lights, tables=tables, plots=plots,
                                 tags=tags, methods_list=methods_list, signals_list=signals_list,
                                 influence=influence)

create_geometry

create_geometry(patches, name=None)

Add a Geometry object to the session. Will use a custom delegate if applicable.

Parameters:

Name	Type	Description	Default
patches	list[GeometryPatch]	list of geometry patches	required
name	str	name of the geometry	None

Returns:

Name	Type	Description
Geometry	Geometry	geometry delegate that was created

Source code in rigatoni/core.py

def create_geometry(self, patches: list[GeometryPatch], name: Optional[str] = None) -> Geometry:
    """Add a Geometry object to the session. Will use a custom delegate if applicable.

    Args:
        patches (list[GeometryPatch]): list of geometry patches
        name (str, optional): name of the geometry

    Returns:
        Geometry: geometry delegate that was created
    """
    return self.create_component(Geometry, name=name, patches=patches)

create_image

create_image(name=None, buffer_source=None, uri_source=None)

Add an Image object to the session.

Will use a custom delegate if applicable. Must specify either a buffer_source or a uri_source.

Parameters:

Name	Type	Description	Default
name	str	name of the image	None
buffer_source	BufferID	buffer data that for image	None
uri_source	str	uri to get the image bytes from	None

Returns:

Name	Type	Description
Image	Image	image delegate that was created

Source code in rigatoni/core.py

def create_image(self, name: Optional[str] = None,
                 buffer_source: BufferID = None,
                 uri_source: str = None) -> Image:
    """Add an Image object to the session.

    Will use a custom delegate if applicable. Must specify either a buffer_source or a uri_source.

    Args:
        name (str, optional): name of the image
        buffer_source (BufferID, optional): buffer data that for image
        uri_source (str, optional): uri to get the image bytes from

    Returns:
        Image: image delegate that was created
    """
    return self.create_component(Image, name=name, buffer_source=buffer_source, uri_source=uri_source)

create_light

create_light(name=None, color=(1.0, 1.0, 1.0), intensity=1.0, point=None, spot=None, directional=None)

Add a Light object to the session. Will use a custom delegate if applicable.

Parameters:

Name	Type	Description	Default
name	str	name of the light	None
color	RGB	color of the light	(1.0, 1.0, 1.0)
intensity	float	intensity of the light on scale from 0-1	1.0
point	PointLight	point light information	None
spot	SpotLight	spot light information	None
directional	DirectionalLight	directional light information	None

Returns:

Name	Type	Description
Light	Light	light delegate that was created

Source code in rigatoni/core.py

def create_light(self, name: Optional[str] = None,
                 color: Optional[RGB] = (1.0, 1.0, 1.0),
                 intensity: Optional[float] = 1.0,
                 point: PointLight = None,
                 spot: SpotLight = None,
                 directional: DirectionalLight = None) -> Light:
    """Add a Light object to the session. Will use a custom delegate if applicable.

    Args:
        name (str, optional): name of the light
        color (RGB, optional): color of the light
        intensity (float, optional): intensity of the light on scale from 0-1
        point (PointLight, optional): point light information
        spot (SpotLight, optional): spot light information
        directional (DirectionalLight, optional): directional light information

    Returns:
        Light: light delegate that was created
    """
    return self.create_component(Light, name=name, color=color, intensity=intensity,
                                 point=point, spot=spot, directional=directional)

create_material

create_material(name=None, pbr_info=PBRInfo(), normal_texture=None, occlusion_texture=None, occlusion_texture_factor=1.0, emissive_texture=None, emissive_factor=(1.0, 1.0, 1.0), use_alpha=False, alpha_cutoff=0.5, double_sided=False)

Add a Material object to the session. Will use a custom delegate if applicable.

Parameters:

Name	Type	Description	Default
name	str	name of the material	None
pbr_info	PBRInfo	physically based rendering information	PBRInfo()
normal_texture	TextureRef	texture for normal mapping	None
occlusion_texture	TextureRef	texture for occlusion mapping	None
occlusion_texture_factor	float	factor for occlusion mapping	1.0
emissive_texture	TextureRef	texture for emissive mapping	None
emissive_factor	Vec3	factor for emissive mapping	(1.0, 1.0, 1.0)
use_alpha	bool	whether to use alpha	False
alpha_cutoff	float	alpha cutoff value	0.5
double_sided	bool	whether the material is double-sided	False

Returns:

Name	Type	Description
Material	Material	material delegate that was created

Source code in rigatoni/core.py

def create_material(self, name: Optional[str] = None,
                    pbr_info: Optional[PBRInfo] = PBRInfo(),
                    normal_texture: Optional[TextureRef] = None,
                    occlusion_texture: Optional[TextureRef] = None,
                    occlusion_texture_factor: Optional[float] = 1.0,
                    emissive_texture: Optional[TextureRef] = None,
                    emissive_factor: Optional[Vec3] = (1.0, 1.0, 1.0),
                    use_alpha: Optional[bool] = False,
                    alpha_cutoff: Optional[float] = .5,
                    double_sided: Optional[bool] = False) -> Material:
    """Add a Material object to the session. Will use a custom delegate if applicable.

    Args:
        name (str, optional): name of the material
        pbr_info (PBRInfo, optional): physically based rendering information
        normal_texture (TextureRef, optional): texture for normal mapping
        occlusion_texture (TextureRef, optional): texture for occlusion mapping
        occlusion_texture_factor (float, optional): factor for occlusion mapping
        emissive_texture (TextureRef, optional): texture for emissive mapping
        emissive_factor (Vec3, optional): factor for emissive mapping
        use_alpha (bool, optional): whether to use alpha
        alpha_cutoff (float, optional): alpha cutoff value
        double_sided (bool, optional): whether the material is double-sided

    Returns:
        Material: material delegate that was created
    """
    return self.create_component(Material, name=name, pbr_info=pbr_info, normal_texture=normal_texture,
                                 occlusion_texture=occlusion_texture,
                                 occlusion_texture_factor=occlusion_texture_factor,
                                 emissive_texture=emissive_texture, emissive_factor=emissive_factor,
                                 use_alpha=use_alpha, alpha_cutoff=alpha_cutoff, double_sided=double_sided)

create_method

create_method(name, arg_doc, doc=None, return_doc=None)

Add a Method object to the scene and return it. Will use a custom delegate if applicable.

Parameters:

Name	Type	Description	Default
name	str	name of the method	required
arg_doc	list[MethodArg]	list of arguments and documentation for the method	required
doc	str	documentation for the method	None
return_doc	str	documentation for the return value	None

Returns:

Name	Type	Description
Method	Method	method delegate that was created

Source code in rigatoni/core.py

def create_method(self, name: str,
                  arg_doc: list[MethodArg],
                  doc: Optional[str] = None,
                  return_doc: Optional[str] = None) -> Method:
    """Add a Method object to the scene and return it. Will use a custom delegate if applicable.

    Args:
        name (str): name of the method
        arg_doc (list[MethodArg]): list of arguments and documentation for the method
        doc (str, optional): documentation for the method
        return_doc (str, optional): documentation for the return value

    Returns:
        Method: method delegate that was created
    """
    return self.create_component(Method, name=name, doc=doc, return_doc=return_doc, arg_doc=arg_doc)

create_plot

create_plot(name=None, table=None, simple_plot=None, url_plot=None, methods_list=None, signals_list=None)

Add a Plot object to the session. Will use a custom delegate if applicable.

Parameters:

Name	Type	Description	Default
name	str	name of the plot	None
table	TableID	table to be plotted	None
simple_plot	str	simple plot to be plotted	None
url_plot	str	url for the plot	None
methods_list	list[MethodID]	attached methods	None
signals_list	list[SignalID]	attached signals	None

Returns:

Name	Type	Description
Plot	Plot	plot delegate that was created

Source code in rigatoni/core.py

def create_plot(self, name: Optional[str] = None,
                table: Optional[TableID] = None,
                simple_plot: Optional[str] = None,
                url_plot: Optional[str] = None,
                methods_list: Optional[list[MethodID]] = None,
                signals_list: Optional[list[SignalID]] = None) -> Plot:
    """Add a Plot object to the session. Will use a custom delegate if applicable.

    Args:
        name (str, optional): name of the plot
        table (TableID, optional): table to be plotted
        simple_plot (str, optional): simple plot to be plotted
        url_plot (str, optional): url for the plot
        methods_list (list[MethodID], optional): attached methods
        signals_list (list[SignalID], optional): attached signals

    Returns:
        Plot: plot delegate that was created
    """
    return self.create_component(Plot, name=name, table=table, simple_plot=simple_plot, url_plot=url_plot,
                                 methods_list=methods_list, signals_list=signals_list)

create_sampler

create_sampler(name=None, mag_filter='LINEAR', min_filter='LINEAR_MIPMAP_LINEAR', wrap_s='REPEAT', wrap_t='REPEAT')

Add a Sampler object to the session. Will use a custom delegate if applicable.

Parameters:

Name	Type	Description	Default
name	str	name of the sampler	None
mag_filter	str	magnification filter	'LINEAR'
min_filter	str	minification filter	'LINEAR_MIPMAP_LINEAR'
wrap_s	str	wrap mode for s coordinate	'REPEAT'
wrap_t	str	wrap mode for t coordinate	'REPEAT'

Returns:

Name	Type	Description
Sampler	Sampler	sampler delegate that was created

Source code in rigatoni/core.py

def create_sampler(self, name: Optional[str] = None,
                   mag_filter: Optional[Literal["NEAREST", "LINEAR"]] = "LINEAR",
                   min_filter: Optional[
                       Literal["NEAREST", "LINEAR", "LINEAR_MIPMAP_LINEAR"]] = "LINEAR_MIPMAP_LINEAR",
                   wrap_s: Optional[SamplerMode] = "REPEAT",
                   wrap_t: Optional[SamplerMode] = "REPEAT") -> Sampler:
    """Add a Sampler object to the session. Will use a custom delegate if applicable.

    Args:
        name (str, optional): name of the sampler
        mag_filter (str, optional): magnification filter
        min_filter (str, optional): minification filter
        wrap_s (str, optional): wrap mode for s coordinate
        wrap_t (str, optional): wrap mode for t coordinate

    Returns:
        Sampler: sampler delegate that was created
    """
    return self.create_component(Sampler, name=name, mag_filter=mag_filter, min_filter=min_filter,
                                 wrap_s=wrap_s, wrap_t=wrap_t)

create_signal

create_signal(name, doc=None, arg_doc=None)

Add a Signal object to the session. Will use a custom delegate if applicable.

Parameters:

Name	Type	Description	Default
name	str	name of the signal	required
doc	str	documentation for the signal	None
arg_doc	list[MethodArg]	list of arguments and documentation for the signal	None

Returns:

Name	Type	Description
Signal	Signal	signal delegate that was created

Source code in rigatoni/core.py

def create_signal(self, name: str,
                  doc: Optional[str] = None,
                  arg_doc: list[MethodArg] = None) -> Signal:
    """Add a Signal object to the session. Will use a custom delegate if applicable.

    Args:
        name (str): name of the signal
        doc (str, optional): documentation for the signal
        arg_doc (list[MethodArg], optional): list of arguments and documentation for the signal

    Returns:
        Signal: signal delegate that was created
    """
    if arg_doc is None:
        arg_doc = []
    return self.create_component(Signal, name=name, doc=doc, arg_doc=arg_doc)

create_table

create_table(name=None, meta=None, methods_list=None, signals_list=None)

Add a Table object to the session. Will use a custom delegate if applicable.

Parameters:

Name	Type	Description	Default
name	str	name of the table	None
meta	str	meta description for the table	None
methods_list	list[MethodID]	list of methods for the table	None
signals_list	list[SignalID]	list of signals for the table	None

Returns:

Name	Type	Description
Table	Table	table delegate that was created

Source code in rigatoni/core.py

def create_table(self, name: Optional[str] = None,
                 meta: Optional[str] = None,
                 methods_list: Optional[list[MethodID]] = None,
                 signals_list: Optional[list[SignalID]] = None) -> Table:
    """Add a Table object to the session. Will use a custom delegate if applicable.

    Args:
        name (str, optional): name of the table
        meta (str, optional): meta description for the table
        methods_list (list[MethodID], optional): list of methods for the table
        signals_list (list[SignalID], optional): list of signals for the table

    Returns:
        Table: table delegate that was created
    """
    return self.create_component(Table, name=name, meta=meta, methods_list=methods_list, signals_list=signals_list)

create_texture

create_texture(image, name=None, sampler=None)

Add a Texture object to the session. Will use a custom delegate if applicable.

Parameters:

Name	Type	Description	Default
image	ImageID	image to be used for the texture	required
name	str	name of the texture	None
sampler	SamplerID	sampler to be used for the texture	None

Returns:

Name	Type	Description
Texture	Texture	texture delegate that was created

Source code in rigatoni/core.py

def create_texture(self, image: ImageID,
                   name: Optional[str] = None,
                   sampler: Optional[SamplerID] = None) -> Texture:
    """Add a Texture object to the session. Will use a custom delegate if applicable.

    Args:
        image (ImageID): image to be used for the texture
        name (str, optional): name of the texture
        sampler (SamplerID, optional): sampler to be used for the texture

    Returns:
        Texture: texture delegate that was created
    """
    return self.create_component(Texture, name=name, image=image, sampler=sampler)

delete_component

delete_component(delegate, recursive=False)

Delete object in state and update clients

This method excepts a delegate, or component ID, and will attempt to delete the component as long as it is not referenced by any other component. If this component is still being used by another, it will be added to a queue so that it can be deleted later once that reference is no longer being used. If recursive flag is set, then all components referenced by this one will also be deleted or at least queued to be deleted.

Parameters:

Name	Type	Description	Default
delegate	Component | Delegate | ID	component / delegate to be deleted	required

Raises:

Type	Description
TypeError	if the user specifies an invalid input type

Source code in rigatoni/core.py

def delete_component(self, delegate: Union[Delegate, ID], recursive: bool = False):
    """Delete object in state and update clients

    This method excepts a delegate, or component ID, and will attempt
    to delete the component as long as it is not referenced by any other component.
    If this component is still being used by another, it will be added to a queue so that
    it can be deleted later once that reference is no longer being used. If recursive flag
    is set, then all components referenced by this one will also be deleted or at least
    queued to be deleted.

    Args:
        delegate (Component | Delegate | ID): component / delegate to be deleted

    Raises:
        TypeError: if the user specifies an invalid input type
    """

    # Handle cases so can except different input types - cast to ID
    if isinstance(delegate, Delegate):
        delegate = delegate
        del_id = delegate.id
    elif isinstance(delegate, ID):
        delegate = self.state[delegate]
        del_id = delegate.id
    else:
        raise TypeError(f"Invalid type for delegate when deleting: {type(delegate)}")

    # Delete all referenced components if recursive flag is set
    if recursive:
        referenced = self._get_referenced(delegate)
        for ref in referenced:
            self.delete_component(ref, recursive=recursive)

    # Delete if no references, or else queue it up for later
    if not self.references.get(del_id):
        self.broadcast(self._prepare_message("delete", delegate))
        del self.state[del_id]
        del self.client_state[del_id]

        # Free up the ID
        self.ids[type(delegate)].on_deck.put(del_id)

        # Clean out references from this object
        for refs in self.references.values():
            while del_id in refs:
                refs.remove(del_id)

        # Check if anything in the queue is now clear to be deleted
        for comp_id in list(self.delete_queue):
            if not self.references.get(comp_id):
                self.delete_queue.remove(comp_id)
                self.delete_component(comp_id)

    else:
        logging.info(f"Couldn't delete {delegate}, referenced by {self.references[del_id]}, added to queue")
        self.delete_queue.add(del_id)

get_delegate

get_delegate(identifier)

Access components in state

Can be called with an ID, name, or context of the delegate

Parameters:

Name	Type	Description	Default
identifier	ID | str| Dict[str, ID]]	identifier for component	required

Returns:

Name	Type	Description
delegate	Delegate	delegate with specified identifier

Raises:

Type	Description
TypeError	if identifier is not of type ID, str, or dict
ValueError	if no component with specified identifier is found or context is invalid

Source code in rigatoni/core.py

def get_delegate(self, identifier: Union[ID, str, Dict[str, ID]]):
    """Access components in state

    Can be called with an ID, name, or context of the delegate

    Args:
        identifier (ID | str| Dict[str, ID]]): identifier for component

    Returns:
        delegate (Delegate): delegate with specified identifier

    Raises:
        TypeError: if identifier is not of type ID, str, or dict
        ValueError: if no component with specified identifier is found or context is invalid
    """
    if isinstance(identifier, ID):
        return self.state[identifier]
    elif isinstance(identifier, str):
        return self.state[self.get_delegate_id(identifier)]
    elif isinstance(identifier, dict):
        return self.get_delegate_by_context(identifier)
    else:
        raise TypeError(f"Invalid type for identifier: {type(identifier)}")

get_delegate_by_context

get_delegate_by_context(context)

Get a component using a context object

This is especially useful in methods that are invoked in a certain context.

Note

Contexts are only used when working with entities, tables, and plots.

Parameters:

Name	Type	Description	Default
context	dict	context of the form {str: ID}	required

Returns:

Name	Type	Description
delegate	Delegate	delegate from specified context

Raises:

Type	Description
ValueError	if context is invalid

Source code in rigatoni/core.py

def get_delegate_by_context(self, context: dict):
    """Get a component using a context object

    This is especially useful in methods that are invoked in a certain context.

    !!! note

        Contexts are only used when working with entities, tables, and plots.

    Args:
        context (dict): context of the form {str: ID}

    Returns:
        delegate (Delegate): delegate from specified context

    Raises:
        ValueError: if context is invalid
    """

    entity = context.get("entity")
    table = context.get("table")
    plot = context.get("plot")
    if entity:
        return self.get_delegate(EntityID(*entity))
    elif table:
        return self.get_delegate(TableID(*table))
    elif plot:
        return self.get_delegate(PlotID(*plot))
    else:
        raise ValueError(f"Invalid context: {context}")

get_delegate_id

get_delegate_id(name)

Get a component by using its name

Parameters:

Name	Type	Description	Default
name	str	name of component to get	required

Returns:

Name	Type	Description
id	ID	id of component with specified name

Raises:

Type	Description
ValueError	if no component with specified name is found

Source code in rigatoni/core.py

def get_delegate_id(self, name: str):
    """Get a component by using its name

    Args:
        name (str): name of component to get

    Returns:
        id (ID): id of component with specified name

    Raises:
        ValueError: if no component with specified name is found
    """

    for delegate in self.state.values():
        if delegate.name == name:
            return delegate.id
    raise ValueError("No Component Found")

get_ids_by_type

get_ids_by_type(component)

Get all ids for certain component type

Parameters:

Name	Type	Description	Default
component	type	type of component to get ID's for	required

Returns:

Name	Type	Description
ids	list	list of ids for components of specified type

Source code in rigatoni/core.py

def get_ids_by_type(self, component: Type[Delegate]) -> list:
    """Get all ids for certain component type

    Args:
        component (type): type of component to get ID's for

    Returns:
        ids: list of ids for components of specified type
    """

    return [key for key, val in self.state.items() if isinstance(val, component)]

invoke_signal

invoke_signal(signal, on_component, signal_data=None)

Send signal to target component

Parameters:

Name	Type	Description	Default
signal	SignalID | Signal	signal to be invoked	required
on_component	Delegate	component to receive the signal	required
signal_data	dict	data to be sent with the signal	None

Returns:

Name	Type	Description
message	list	message to be broadcast

Raises:

Type	Description
ValueError	if the user specifies an invalid on_component type

Source code in rigatoni/core.py

def invoke_signal(self, signal: Union[SignalID, Signal], on_component: Delegate, signal_data: list = None):
    """Send signal to target component

    Args:
        signal (SignalID | Signal): signal to be invoked
        on_component (Delegate): component to receive the signal
        signal_data (dict): data to be sent with the signal

    Returns:
        message (list): message to be broadcast

    Raises:
        ValueError: if the user specifies an invalid on_component type
    """

    # Cast signal to ID if needed
    if isinstance(signal, Signal):
        signal = signal.id

    # Fill in default signal data if not specified
    if signal_data is None:
        signal_data = []

    # Get context from on_component
    if isinstance(on_component, Entity):
        context = InvokeIDType(entity=on_component.id)
    elif isinstance(on_component, Table):
        context = InvokeIDType(table=on_component.id)
    elif isinstance(on_component, Plot):
        context = InvokeIDType(plot=on_component.id)
    else:
        raise ValueError(f"Invalid on_component type: {type(on_component)}")

    # Create invoke object and broadcast message
    invoke = Invoke(id=signal, context=context, signal_data=signal_data)
    message = self._prepare_message("invoke", invoke)
    self.broadcast(message)
    return message

run

run()

Run the server

This will run indefinitely until the server is shutdown

Source code in rigatoni/core.py

def run(self):
    """Run the server

    This will run indefinitely until the server is shutdown"""
    return asyncio.run(self._start_server())

shutdown

shutdown()

Shuts down the server and closes off communication with clients

Source code in rigatoni/core.py

def shutdown(self):
    """Shuts down the server and closes off communication with clients"""
    logging.info("Shutting down server...")
    self.shutdown_event.set()

update_component

update_component(current)

Update clients with changes to a component

This method broadcasts changes to all clients including only fields specified in the set delta. Local changes to delegates will be saved in the server's state, but this method must be called to update clients.

Parameters:

Name	Type	Description	Default
current	Delegate	component that has been updated, should be a component with an update message	required

Source code in rigatoni/core.py

def update_component(self, current: Delegate):
    """Update clients with changes to a component

    This method broadcasts changes to all clients including only fields
    specified in the set delta. Local changes to delegates will be saved
    in the server's state, but this method must be called to update clients.

    Args:
        current (Delegate): component that has been updated,
            should be a component with an update message
    """

    # Find difference between two states
    outdated = self.client_state[current.id]
    delta = self._find_delta(outdated, current)

    # Update references
    self._update_references(outdated, outdated, removing=True)
    self._update_references(current, current)

    # Update tracking state
    self.client_state[current.id] = current.model_copy()

    # Form message and broadcast
    try:
        message = self._prepare_message("update", current, delta)
        self.broadcast(message)
    except Exception as e:
        raise ValueError(f"This obj can not be updated: {e}")