Geometry

rigatoni.ByteServer

ByteServer(port=8000)

Bases: object

Server to Host URI Bytes.

This is helpful for large meshes and images that would otherwise be too large to send using CBOR and the websocket connection. The server maps a tag to a buffer, and the client can request the buffer using the url that includes the tag.

Note

The server is start up automatically in a new thread when the class is instantiated, and shutdown must be called to clean up this thread.

Attributes:

Name	Type	Description
host	str	IP address for server
port	int	port server is listening on
socket	socket	socket connection
buffers	dict	mapping tag to buffer
_next_tag	int	next available tag for a buffer
url	str	base url to reach server without tag
thread	Thread	background thread server is running in
running	bool	flag indicating whether server is running

Parameters:

Name	Type	Description	Default
port	int	port to listen and host on	8000

Source code in rigatoni/byte_server.py

def __init__(self, port: int = 8000):
    """Constructor to create the server

    Args:
        port (int): port to listen and host on
    """

    name = socket.gethostname()
    try:  # supposed to work without .local, but had to add to match system preferences - sharing
        self.host = socket.gethostbyname(name)
    except socket.gaierror:
        self.host = socket.gethostbyname(f"{name}.local")

    self.port = port
    self.socket = None
    self.buffers = {}
    self._next_tag = 0
    self.url = f"http://{self.host}:{port}"

    self.thread = threading.Thread(target=self._run, args=())
    self.running = True
    self.ready = threading.Event()
    self.thread.start()
    self.ready.wait()

add_buffer

add_buffer(buffer)

Add buffer to server and return url to reach it

Parameters:

Name	Type	Description	Default
buffer	bytes	bytes to add as buffer	required

Source code in rigatoni/byte_server.py

def add_buffer(self, buffer) -> str:
    """Add buffer to server and return url to reach it

    Args:
        buffer (bytes): bytes to add as buffer
    """

    tag = self._get_tag()
    self.buffers[tag] = buffer
    url = f"{self.url}/{tag}"
    logging.info(f"Adding buffer to byte server: {url}")

    return url

get_buffer

get_buffer(uri)

Helper to get bytes for a URI

Mostly used in geometry creation for exporting as of right now

Parameters:

Name	Type	Description	Default
uri	str	uri for bytes	required

Source code in rigatoni/byte_server.py

def get_buffer(self, uri: str):
    """Helper to get bytes for a URI

    Mostly used in geometry creation for exporting as of right now

    Args:
        uri (str): uri for bytes
    """

    m = re.search(f'(?<={self.port}/).+\Z', uri)
    if m:
        tag = m.group(0)
        buffer_bytes = self.buffers[tag]
        return buffer_bytes
    else:
        raise ValueError("Invalid HTTP Request")

shutdown

shutdown()

Stop running thread

Source code in rigatoni/byte_server.py

def shutdown(self):
    """Stop running thread"""

    self.running = False
    self.thread.join()

---

rigatoni.geometry.methods

Helpful methods for assisting with the creation of geometry objects

---

rigatoni.geometry.objects

Module with additional objects to help with geometry creation

These are based on the noodle_objects.py module and implement validation