Skip to content

GSP Network API Reference

The GSP Network module provides network-based rendering capabilities, enabling remote visualization and client-server architectures.

Overview

gsp_network

gsp_network provides functionality for rendering graphics over a network using different remote renderers.

Renderer Module

The renderer module contains the network renderer implementation for remote rendering.

gsp_network.renderer

Network renderer that sends rendering requests to a remote server and displays the results using Matplotlib.

NetworkRenderer

Bases: gsp.types.renderer_base.RendererBase

Note: this requires a running gsp_network server. See the README for instructions.

IMPORTANT: it DOES NOT depend on GSP matplotlib renderer, it only uses pip matplotlib to display the remotely rendered images.

Source code in src/gsp_network/renderer/network_renderer.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
class NetworkRenderer(RendererBase):
    """**Note**: this requires a running gsp_network server. See the README for instructions.

    **IMPORTANT**: it DOES NOT depend on GSP matplotlib renderer, it only uses pip matplotlib to display the remotely rendered images.
    """

    def __init__(self, canvas: Canvas, server_base_url: str, remote_renderer_name: Literal["matplotlib", "datoviz"] = "matplotlib") -> None:
        """Initialize the NetworkRenderer.

        Args:
            canvas (Canvas): _description_
            server_base_url (str): _description_
            remote_renderer_name (Literal["matplotlib", "datoviz"], optional): _description_. Defaults to "matplotlib".
        """
        self._canvas = canvas
        self._server_base_url = server_base_url
        self._remote_renderer_name: Literal["matplotlib", "datoviz"] = remote_renderer_name

        # Create a figure
        figure_width = self._canvas.get_width() / self._canvas.get_dpi()
        figure_height = self._canvas.get_height() / self._canvas.get_dpi()
        self._figure: matplotlib.figure.Figure = matplotlib.pyplot.figure(figsize=(figure_width, figure_height), dpi=self._canvas.get_dpi())
        assert self._figure.canvas.manager is not None, "matplotlib figure canvas manager is None"
        self._figure.canvas.manager.set_window_title(f"Network ({self._remote_renderer_name})")

        # get the only axes in the figure
        self._mpl_axes = self._figure.add_axes((0, 0, 1, 1))
        # hide the borders
        self._mpl_axes.axis("off")

        # create an np.array to hold the image
        image_data_np = np.zeros((self._canvas.get_height(), self._canvas.get_width(), 3), dtype=np.uint8)
        self._axes_image = self._mpl_axes.imshow(image_data_np, aspect="auto")

    def get_canvas(self) -> Canvas:
        """Get the canvas associated with the network renderer.

        Returns:
            Canvas: The canvas associated with the network renderer.
        """
        return self._canvas

    def close(self) -> None:
        """Close the network renderer and release resources."""
        # stop the event loop if any - thus .show(block=True) will return
        self._figure.canvas.stop_event_loop()
        # close the figure
        matplotlib.pyplot.close(self._figure)
        self._figure = None  # type: ignore

    def get_remote_renderer_name(self) -> Literal["matplotlib", "datoviz"]:
        """Get the name of the remote renderer being used.

        Returns:
            Literal["matplotlib", "datoviz"]: The name of the remote renderer.
        """
        return self._remote_renderer_name

    def render(
        self,
        viewports: Sequence[Viewport],
        visuals: Sequence[VisualBase],
        model_matrices: Sequence[TransBuf],
        cameras: Sequence[Camera],
    ) -> bytes:
        """Render the scene remotely and update the matplotlib figure with the rendered image.

        Args:
            viewports (Sequence[Viewport]): The viewports to render.
            visuals (Sequence[VisualBase]): The visuals to render.
            model_matrices (Sequence[TransBuf]): The model matrices for the visuals.
            cameras (Sequence[Camera]): The cameras to use for rendering.

        Returns:
            bytes: The rendered image data in PNG format.

        Raises:
            Exception: If the network request fails.
        """
        # =============================================================================
        # Serialize the scene and create the payload
        # =============================================================================
        pydanticSerializer = PydanticSerializer(self._canvas)
        pydantic_scene_dict = pydanticSerializer.serialize(
            viewports=viewports,
            visuals=visuals,
            model_matrices=model_matrices,
            cameras=cameras,
        )

        payload: NetworkPayload = {
            "renderer_name": self._remote_renderer_name,
            "data": pydantic_scene_dict,
        }

        # =============================================================================
        # do network request to send the payload and get the rendered image
        # =============================================================================
        # Send the POST request with JSON data
        call_url = f"{self._server_base_url}/render"
        headers = {"Content-Type": "application/json"}
        response = requests.post(call_url, data=json.dumps(payload), headers=headers)

        # Check the response status
        if response.status_code != HttpStatus.OK:
            raise Exception(f"Request failed with status code {response.status_code}")
        image_png_data = response.content

        # =============================================================================
        # Render the image in the matplotlib figure
        # =============================================================================
        assert self._axes_image is not None, "PANIC self._axes_image is None"
        # update the image data
        image_data_io = io.BytesIO(image_png_data)
        image_data_np = matplotlib.image.imread(image_data_io, format="png")
        self._axes_image.set_data(image_data_np)

        # return png data as bytes
        return image_png_data

    def show(self) -> None:
        """Show the rendered canvas (blocking call)."""
        # handle non-interactive mode for tests
        in_test = os.environ.get("GSP_TEST") == "True"
        if in_test:
            return

        matplotlib.pyplot.show()

    def get_mpl_figure(self) -> matplotlib.figure.Figure:
        """Get the underlying Matplotlib figure.

        Returns:
            matplotlib.figure.Figure: The Matplotlib figure used by the renderer.
        """
        return self._figure

__init__(canvas: Canvas, server_base_url: str, remote_renderer_name: Literal['matplotlib', 'datoviz'] = 'matplotlib') -> None

Initialize the NetworkRenderer.

Parameters:

Name Type Description Default
canvas gsp.core.canvas.Canvas

description

 required
server_base_url str

description

 required
remote_renderer_name typing.Literal['matplotlib', 'datoviz']

description. Defaults to "matplotlib".

 'matplotlib'
Source code in src/gsp_network/renderer/network_renderer.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(self, canvas: Canvas, server_base_url: str, remote_renderer_name: Literal["matplotlib", "datoviz"] = "matplotlib") -> None:
    """Initialize the NetworkRenderer.

    Args:
        canvas (Canvas): _description_
        server_base_url (str): _description_
        remote_renderer_name (Literal["matplotlib", "datoviz"], optional): _description_. Defaults to "matplotlib".
    """
    self._canvas = canvas
    self._server_base_url = server_base_url
    self._remote_renderer_name: Literal["matplotlib", "datoviz"] = remote_renderer_name

    # Create a figure
    figure_width = self._canvas.get_width() / self._canvas.get_dpi()
    figure_height = self._canvas.get_height() / self._canvas.get_dpi()
    self._figure: matplotlib.figure.Figure = matplotlib.pyplot.figure(figsize=(figure_width, figure_height), dpi=self._canvas.get_dpi())
    assert self._figure.canvas.manager is not None, "matplotlib figure canvas manager is None"
    self._figure.canvas.manager.set_window_title(f"Network ({self._remote_renderer_name})")

    # get the only axes in the figure
    self._mpl_axes = self._figure.add_axes((0, 0, 1, 1))
    # hide the borders
    self._mpl_axes.axis("off")

    # create an np.array to hold the image
    image_data_np = np.zeros((self._canvas.get_height(), self._canvas.get_width(), 3), dtype=np.uint8)
    self._axes_image = self._mpl_axes.imshow(image_data_np, aspect="auto")

get_canvas() -> Canvas

Get the canvas associated with the network renderer.

Returns:

Name Type Description
Canvas gsp.core.canvas.Canvas

The canvas associated with the network renderer.
Source code in src/gsp_network/renderer/network_renderer.py 
71
72
73
74
75
76
77
def get_canvas(self) -> Canvas:
    """Get the canvas associated with the network renderer.

    Returns:
        Canvas: The canvas associated with the network renderer.
    """
    return self._canvas

close() -> None

Close the network renderer and release resources.

Source code in src/gsp_network/renderer/network_renderer.py 
79
80
81
82
83
84
85
def close(self) -> None:
    """Close the network renderer and release resources."""
    # stop the event loop if any - thus .show(block=True) will return
    self._figure.canvas.stop_event_loop()
    # close the figure
    matplotlib.pyplot.close(self._figure)
    self._figure = None  # type: ignore

get_remote_renderer_name() -> Literal['matplotlib', 'datoviz']

Get the name of the remote renderer being used.

Returns:

Type Description
typing.Literal['matplotlib', 'datoviz']

Literal["matplotlib", "datoviz"]: The name of the remote renderer.
Source code in src/gsp_network/renderer/network_renderer.py 
87
88
89
90
91
92
93
def get_remote_renderer_name(self) -> Literal["matplotlib", "datoviz"]:
    """Get the name of the remote renderer being used.

    Returns:
        Literal["matplotlib", "datoviz"]: The name of the remote renderer.
    """
    return self._remote_renderer_name

render(viewports: Sequence[Viewport], visuals: Sequence[VisualBase], model_matrices: Sequence[TransBuf], cameras: Sequence[Camera]) -> bytes

Render the scene remotely and update the matplotlib figure with the rendered image.

Parameters:

Name Type Description Default
viewports typing.Sequence[gsp.core.viewport.Viewport]

The viewports to render.

 required
visuals typing.Sequence[gsp.types.visual_base.VisualBase]

The visuals to render.

 required
model_matrices typing.Sequence[gsp.types.transbuf.TransBuf]

The model matrices for the visuals.

 required
cameras typing.Sequence[gsp.core.camera.Camera]

The cameras to use for rendering.

 required

Returns:

Name Type Description
bytes bytes

The rendered image data in PNG format.

Raises:

Type Description
Exception

If the network request fails.
Source code in src/gsp_network/renderer/network_renderer.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def render(
    self,
    viewports: Sequence[Viewport],
    visuals: Sequence[VisualBase],
    model_matrices: Sequence[TransBuf],
    cameras: Sequence[Camera],
) -> bytes:
    """Render the scene remotely and update the matplotlib figure with the rendered image.

    Args:
        viewports (Sequence[Viewport]): The viewports to render.
        visuals (Sequence[VisualBase]): The visuals to render.
        model_matrices (Sequence[TransBuf]): The model matrices for the visuals.
        cameras (Sequence[Camera]): The cameras to use for rendering.

    Returns:
        bytes: The rendered image data in PNG format.

    Raises:
        Exception: If the network request fails.
    """
    # =============================================================================
    # Serialize the scene and create the payload
    # =============================================================================
    pydanticSerializer = PydanticSerializer(self._canvas)
    pydantic_scene_dict = pydanticSerializer.serialize(
        viewports=viewports,
        visuals=visuals,
        model_matrices=model_matrices,
        cameras=cameras,
    )

    payload: NetworkPayload = {
        "renderer_name": self._remote_renderer_name,
        "data": pydantic_scene_dict,
    }

    # =============================================================================
    # do network request to send the payload and get the rendered image
    # =============================================================================
    # Send the POST request with JSON data
    call_url = f"{self._server_base_url}/render"
    headers = {"Content-Type": "application/json"}
    response = requests.post(call_url, data=json.dumps(payload), headers=headers)

    # Check the response status
    if response.status_code != HttpStatus.OK:
        raise Exception(f"Request failed with status code {response.status_code}")
    image_png_data = response.content

    # =============================================================================
    # Render the image in the matplotlib figure
    # =============================================================================
    assert self._axes_image is not None, "PANIC self._axes_image is None"
    # update the image data
    image_data_io = io.BytesIO(image_png_data)
    image_data_np = matplotlib.image.imread(image_data_io, format="png")
    self._axes_image.set_data(image_data_np)

    # return png data as bytes
    return image_png_data

show() -> None

Show the rendered canvas (blocking call).

Source code in src/gsp_network/renderer/network_renderer.py 
157
158
159
160
161
162
163
164
def show(self) -> None:
    """Show the rendered canvas (blocking call)."""
    # handle non-interactive mode for tests
    in_test = os.environ.get("GSP_TEST") == "True"
    if in_test:
        return

    matplotlib.pyplot.show()

get_mpl_figure() -> matplotlib.figure.Figure

Get the underlying Matplotlib figure.

Returns:

Type Description
matplotlib.figure.Figure

matplotlib.figure.Figure: The Matplotlib figure used by the renderer.
Source code in src/gsp_network/renderer/network_renderer.py 
166
167
168
169
170
171
172
def get_mpl_figure(self) -> matplotlib.figure.Figure:
    """Get the underlying Matplotlib figure.

    Returns:
        matplotlib.figure.Figure: The Matplotlib figure used by the renderer.
    """
    return self._figure

Network Renderer

gsp_network.renderer.network_renderer

Network renderer that sends rendering requests to a remote server and displays the results using Matplotlib.

NetworkPayload

Bases: typing.TypedDict

Type definition for the network payload sent to the server.

Source code in src/gsp_network/renderer/network_renderer.py 
31
32
33
34
class NetworkPayload(TypedDict):
    """Type definition for the network payload sent to the server."""
    renderer_name: Literal["matplotlib", "datoviz"]
    data: PydanticDict

NetworkRenderer

Bases: gsp.types.renderer_base.RendererBase

Note: this requires a running gsp_network server. See the README for instructions.

IMPORTANT: it DOES NOT depend on GSP matplotlib renderer, it only uses pip matplotlib to display the remotely rendered images.

Source code in src/gsp_network/renderer/network_renderer.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
class NetworkRenderer(RendererBase):
    """**Note**: this requires a running gsp_network server. See the README for instructions.

    **IMPORTANT**: it DOES NOT depend on GSP matplotlib renderer, it only uses pip matplotlib to display the remotely rendered images.
    """

    def __init__(self, canvas: Canvas, server_base_url: str, remote_renderer_name: Literal["matplotlib", "datoviz"] = "matplotlib") -> None:
        """Initialize the NetworkRenderer.

        Args:
            canvas (Canvas): _description_
            server_base_url (str): _description_
            remote_renderer_name (Literal["matplotlib", "datoviz"], optional): _description_. Defaults to "matplotlib".
        """
        self._canvas = canvas
        self._server_base_url = server_base_url
        self._remote_renderer_name: Literal["matplotlib", "datoviz"] = remote_renderer_name

        # Create a figure
        figure_width = self._canvas.get_width() / self._canvas.get_dpi()
        figure_height = self._canvas.get_height() / self._canvas.get_dpi()
        self._figure: matplotlib.figure.Figure = matplotlib.pyplot.figure(figsize=(figure_width, figure_height), dpi=self._canvas.get_dpi())
        assert self._figure.canvas.manager is not None, "matplotlib figure canvas manager is None"
        self._figure.canvas.manager.set_window_title(f"Network ({self._remote_renderer_name})")

        # get the only axes in the figure
        self._mpl_axes = self._figure.add_axes((0, 0, 1, 1))
        # hide the borders
        self._mpl_axes.axis("off")

        # create an np.array to hold the image
        image_data_np = np.zeros((self._canvas.get_height(), self._canvas.get_width(), 3), dtype=np.uint8)
        self._axes_image = self._mpl_axes.imshow(image_data_np, aspect="auto")

    def get_canvas(self) -> Canvas:
        """Get the canvas associated with the network renderer.

        Returns:
            Canvas: The canvas associated with the network renderer.
        """
        return self._canvas

    def close(self) -> None:
        """Close the network renderer and release resources."""
        # stop the event loop if any - thus .show(block=True) will return
        self._figure.canvas.stop_event_loop()
        # close the figure
        matplotlib.pyplot.close(self._figure)
        self._figure = None  # type: ignore

    def get_remote_renderer_name(self) -> Literal["matplotlib", "datoviz"]:
        """Get the name of the remote renderer being used.

        Returns:
            Literal["matplotlib", "datoviz"]: The name of the remote renderer.
        """
        return self._remote_renderer_name

    def render(
        self,
        viewports: Sequence[Viewport],
        visuals: Sequence[VisualBase],
        model_matrices: Sequence[TransBuf],
        cameras: Sequence[Camera],
    ) -> bytes:
        """Render the scene remotely and update the matplotlib figure with the rendered image.

        Args:
            viewports (Sequence[Viewport]): The viewports to render.
            visuals (Sequence[VisualBase]): The visuals to render.
            model_matrices (Sequence[TransBuf]): The model matrices for the visuals.
            cameras (Sequence[Camera]): The cameras to use for rendering.

        Returns:
            bytes: The rendered image data in PNG format.

        Raises:
            Exception: If the network request fails.
        """
        # =============================================================================
        # Serialize the scene and create the payload
        # =============================================================================
        pydanticSerializer = PydanticSerializer(self._canvas)
        pydantic_scene_dict = pydanticSerializer.serialize(
            viewports=viewports,
            visuals=visuals,
            model_matrices=model_matrices,
            cameras=cameras,
        )

        payload: NetworkPayload = {
            "renderer_name": self._remote_renderer_name,
            "data": pydantic_scene_dict,
        }

        # =============================================================================
        # do network request to send the payload and get the rendered image
        # =============================================================================
        # Send the POST request with JSON data
        call_url = f"{self._server_base_url}/render"
        headers = {"Content-Type": "application/json"}
        response = requests.post(call_url, data=json.dumps(payload), headers=headers)

        # Check the response status
        if response.status_code != HttpStatus.OK:
            raise Exception(f"Request failed with status code {response.status_code}")
        image_png_data = response.content

        # =============================================================================
        # Render the image in the matplotlib figure
        # =============================================================================
        assert self._axes_image is not None, "PANIC self._axes_image is None"
        # update the image data
        image_data_io = io.BytesIO(image_png_data)
        image_data_np = matplotlib.image.imread(image_data_io, format="png")
        self._axes_image.set_data(image_data_np)

        # return png data as bytes
        return image_png_data

    def show(self) -> None:
        """Show the rendered canvas (blocking call)."""
        # handle non-interactive mode for tests
        in_test = os.environ.get("GSP_TEST") == "True"
        if in_test:
            return

        matplotlib.pyplot.show()

    def get_mpl_figure(self) -> matplotlib.figure.Figure:
        """Get the underlying Matplotlib figure.

        Returns:
            matplotlib.figure.Figure: The Matplotlib figure used by the renderer.
        """
        return self._figure

__init__(canvas: Canvas, server_base_url: str, remote_renderer_name: Literal['matplotlib', 'datoviz'] = 'matplotlib') -> None

Initialize the NetworkRenderer.

Parameters:

Name Type Description Default
canvas gsp.core.canvas.Canvas

description

 required
server_base_url str

description

 required
remote_renderer_name typing.Literal['matplotlib', 'datoviz']

description. Defaults to "matplotlib".

 'matplotlib'
Source code in src/gsp_network/renderer/network_renderer.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(self, canvas: Canvas, server_base_url: str, remote_renderer_name: Literal["matplotlib", "datoviz"] = "matplotlib") -> None:
    """Initialize the NetworkRenderer.

    Args:
        canvas (Canvas): _description_
        server_base_url (str): _description_
        remote_renderer_name (Literal["matplotlib", "datoviz"], optional): _description_. Defaults to "matplotlib".
    """
    self._canvas = canvas
    self._server_base_url = server_base_url
    self._remote_renderer_name: Literal["matplotlib", "datoviz"] = remote_renderer_name

    # Create a figure
    figure_width = self._canvas.get_width() / self._canvas.get_dpi()
    figure_height = self._canvas.get_height() / self._canvas.get_dpi()
    self._figure: matplotlib.figure.Figure = matplotlib.pyplot.figure(figsize=(figure_width, figure_height), dpi=self._canvas.get_dpi())
    assert self._figure.canvas.manager is not None, "matplotlib figure canvas manager is None"
    self._figure.canvas.manager.set_window_title(f"Network ({self._remote_renderer_name})")

    # get the only axes in the figure
    self._mpl_axes = self._figure.add_axes((0, 0, 1, 1))
    # hide the borders
    self._mpl_axes.axis("off")

    # create an np.array to hold the image
    image_data_np = np.zeros((self._canvas.get_height(), self._canvas.get_width(), 3), dtype=np.uint8)
    self._axes_image = self._mpl_axes.imshow(image_data_np, aspect="auto")

close() -> None

Close the network renderer and release resources.

Source code in src/gsp_network/renderer/network_renderer.py 
79
80
81
82
83
84
85
def close(self) -> None:
    """Close the network renderer and release resources."""
    # stop the event loop if any - thus .show(block=True) will return
    self._figure.canvas.stop_event_loop()
    # close the figure
    matplotlib.pyplot.close(self._figure)
    self._figure = None  # type: ignore

get_canvas() -> Canvas

Get the canvas associated with the network renderer.

Returns:

Name Type Description
Canvas gsp.core.canvas.Canvas

The canvas associated with the network renderer.
Source code in src/gsp_network/renderer/network_renderer.py 
71
72
73
74
75
76
77
def get_canvas(self) -> Canvas:
    """Get the canvas associated with the network renderer.

    Returns:
        Canvas: The canvas associated with the network renderer.
    """
    return self._canvas

get_mpl_figure() -> matplotlib.figure.Figure

Get the underlying Matplotlib figure.

Returns:

Type Description
matplotlib.figure.Figure

matplotlib.figure.Figure: The Matplotlib figure used by the renderer.
Source code in src/gsp_network/renderer/network_renderer.py 
166
167
168
169
170
171
172
def get_mpl_figure(self) -> matplotlib.figure.Figure:
    """Get the underlying Matplotlib figure.

    Returns:
        matplotlib.figure.Figure: The Matplotlib figure used by the renderer.
    """
    return self._figure

get_remote_renderer_name() -> Literal['matplotlib', 'datoviz']

Get the name of the remote renderer being used.

Returns:

Type Description
typing.Literal['matplotlib', 'datoviz']

Literal["matplotlib", "datoviz"]: The name of the remote renderer.
Source code in src/gsp_network/renderer/network_renderer.py 
87
88
89
90
91
92
93
def get_remote_renderer_name(self) -> Literal["matplotlib", "datoviz"]:
    """Get the name of the remote renderer being used.

    Returns:
        Literal["matplotlib", "datoviz"]: The name of the remote renderer.
    """
    return self._remote_renderer_name

render(viewports: Sequence[Viewport], visuals: Sequence[VisualBase], model_matrices: Sequence[TransBuf], cameras: Sequence[Camera]) -> bytes

Render the scene remotely and update the matplotlib figure with the rendered image.

Parameters:

Name Type Description Default
viewports typing.Sequence[gsp.core.viewport.Viewport]

The viewports to render.

 required
visuals typing.Sequence[gsp.types.visual_base.VisualBase]

The visuals to render.

 required
model_matrices typing.Sequence[gsp.types.transbuf.TransBuf]

The model matrices for the visuals.

 required
cameras typing.Sequence[gsp.core.camera.Camera]

The cameras to use for rendering.

 required

Returns:

Name Type Description
bytes bytes

The rendered image data in PNG format.

Raises:

Type Description
Exception

If the network request fails.
Source code in src/gsp_network/renderer/network_renderer.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def render(
    self,
    viewports: Sequence[Viewport],
    visuals: Sequence[VisualBase],
    model_matrices: Sequence[TransBuf],
    cameras: Sequence[Camera],
) -> bytes:
    """Render the scene remotely and update the matplotlib figure with the rendered image.

    Args:
        viewports (Sequence[Viewport]): The viewports to render.
        visuals (Sequence[VisualBase]): The visuals to render.
        model_matrices (Sequence[TransBuf]): The model matrices for the visuals.
        cameras (Sequence[Camera]): The cameras to use for rendering.

    Returns:
        bytes: The rendered image data in PNG format.

    Raises:
        Exception: If the network request fails.
    """
    # =============================================================================
    # Serialize the scene and create the payload
    # =============================================================================
    pydanticSerializer = PydanticSerializer(self._canvas)
    pydantic_scene_dict = pydanticSerializer.serialize(
        viewports=viewports,
        visuals=visuals,
        model_matrices=model_matrices,
        cameras=cameras,
    )

    payload: NetworkPayload = {
        "renderer_name": self._remote_renderer_name,
        "data": pydantic_scene_dict,
    }

    # =============================================================================
    # do network request to send the payload and get the rendered image
    # =============================================================================
    # Send the POST request with JSON data
    call_url = f"{self._server_base_url}/render"
    headers = {"Content-Type": "application/json"}
    response = requests.post(call_url, data=json.dumps(payload), headers=headers)

    # Check the response status
    if response.status_code != HttpStatus.OK:
        raise Exception(f"Request failed with status code {response.status_code}")
    image_png_data = response.content

    # =============================================================================
    # Render the image in the matplotlib figure
    # =============================================================================
    assert self._axes_image is not None, "PANIC self._axes_image is None"
    # update the image data
    image_data_io = io.BytesIO(image_png_data)
    image_data_np = matplotlib.image.imread(image_data_io, format="png")
    self._axes_image.set_data(image_data_np)

    # return png data as bytes
    return image_png_data

show() -> None

Show the rendered canvas (blocking call).

Source code in src/gsp_network/renderer/network_renderer.py 
157
158
159
160
161
162
163
164
def show(self) -> None:
    """Show the rendered canvas (blocking call)."""
    # handle non-interactive mode for tests
    in_test = os.environ.get("GSP_TEST") == "True"
    if in_test:
        return

    matplotlib.pyplot.show()

Tools Module

The tools module provides server utilities for network-based rendering.

gsp_network.tools

tools package initialization.

Network Server

gsp_network.tools.network_server

Server example using Flask to render a scene from JSON input.

  • use Flask to create a simple web server
  • render with matplotlib or datoviz based on environment variable

debug_save_payload: bool = False module-attribute

Enable saving of received payloads and rendered images for debugging.

debug_save_serial: int = 0 module-attribute

Serial number for debug saved files.

ServerSample

Sample class to demonstrate server functionality.

Source code in src/gsp_network/tools/network_server.py 
123
124
125
126
127
128
129
130
131
132
class ServerSample:
    """Sample class to demonstrate server functionality."""

    def __init__(self):
        """Initialize the server sample."""
        pass

    def run(self):
        """Run the Flask server."""
        flask_app.run(threaded=False, debug=False)  # Enable debug mode if desired

__init__()

Initialize the server sample.

Source code in src/gsp_network/tools/network_server.py 
126
127
128
def __init__(self):
    """Initialize the server sample."""
    pass

run()

Run the Flask server.

Source code in src/gsp_network/tools/network_server.py 
130
131
132
def run(self):
    """Run the Flask server."""
    flask_app.run(threaded=False, debug=False)  # Enable debug mode if desired

render_scene_json() -> Response

Flask route to render a scene from JSON input.

Returns:

Name Type Description
Response flask.Response

Flask response containing the rendered PNG image.
Source code in src/gsp_network/tools/network_server.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@flask_app.route("/render", methods=["POST"])
def render_scene_json() -> Response:
    """Flask route to render a scene from JSON input.

    Returns:
        Response: Flask response containing the rendered PNG image.
    """
    payload: NetworkPayload = request.get_json()

    # Log the received payload for debugging
    print("Received payload")

    ###############################################################################
    # Load the scene from JSON
    #

    pydanticDict: PydanticDict = payload["data"]

    pydanticParser = PydanticParser()
    parsed_canvas, parsed_viewports, parsed_visuals, parsed_model_matrices, parsed_cameras = pydanticParser.parse(pydanticDict)

    ###############################################################################
    # Render the loaded scene with matplotlib or datoviz based on environment variable
    #
    renderer_name = payload["renderer_name"]
    if renderer_name == "matplotlib":
        renderer = MatplotlibRenderer(parsed_canvas)
    else:
        renderer = DatovizRenderer(parsed_canvas, offscreen=True)
    image_png_data = renderer.render(parsed_viewports, parsed_visuals, parsed_model_matrices, parsed_cameras, return_image=True)

    print(f"Rendered image size: {text_cyan(str(len(image_png_data)))} bytes")

    # =============================================================================
    # Save payload+image on debug
    # =============================================================================
    if debug_save_payload:
        global debug_save_serial

        # get path for payload+image
        debug_save_serial += 1
        basename = f"payload_{datetime.now().strftime("%Y%m%d_%H%M%S")}_{debug_save_serial}"
        folder_path = pathlib.Path(__file__).parent / "network_server_debug"
        payload_path = folder_path / f"{basename}.json"
        image_path = folder_path / f"{basename}.png"

        # save payload
        # folder_path.mkdir(parents=True, exist_ok=True)
        with open(payload_path, "w") as payload_file:
            payload_file.write(request.get_data(as_text=True))
        print(f"Saved payload to: {text_cyan(str(payload_path))}")
        # save image
        with open(image_path, "wb") as image_file:
            image_file.write(image_png_data)
        print(f"Saved rendered image to: {text_cyan(str(image_path))}")

    ###############################################################################
    # Return the rendered image as a PNG file
    #
    return send_file(
        io.BytesIO(image_png_data),
        mimetype="image/png",
        as_attachment=True,
        download_name="rendered_scene.png",
    )

text_cyan(text: str) -> str

Return the given text string wrapped in ANSI escape codes for cyan color.

Parameters:

Name Type Description Default
text str

The text to color.

 required

Returns:

Name Type Description
str str

The colored text string.
Source code in src/gsp_network/tools/network_server.py 
36
37
38
39
40
41
42
43
44
45
def text_cyan(text: str) -> str:
    """Return the given text string wrapped in ANSI escape codes for cyan color.

    Args:
        text (str): The text to color.

    Returns:
        str: The colored text string.
    """
    return colorama.Fore.CYAN + text + colorama.Style.RESET_ALL

Network Server Kill

gsp_network.tools.network_server_kill

Kill any process using port 5000 (commonly used for flask server).

in shell: lsof -ti tcp:5000 | xargs kill

main() -> int

Main function to kill processes using port 5000.

Returns:

Name Type Description
int int

Exit code (0 for success, 1 for failure).
Source code in src/gsp_network/tools/network_server_kill.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def main() -> int:
    """Main function to kill processes using port 5000.

    Returns:
        int: Exit code (0 for success, 1 for failure).
    """
    # parse command line arguments
    parser = argparse.ArgumentParser(description="Kill any process using port 5000 (commonly used for flask server).")
    _ = parser.parse_args()

    port = 5000

    try:
        # Get the list of process IDs using the specified port
        result = subprocess.run(["lsof", "-ti", f"tcp:{port}"], capture_output=True, text=True)
        pids = result.stdout.strip().split("\n")

        if pids == [""]:
            print(f"No processes found using port {port}.")
            return 0

        for pid in pids:
            os.kill(int(pid), signal.SIGTERM)
            print(f"Killed process with PID: {pid} using port {port}.")

        return 0

    except Exception as error:
        print(f"An error occurred: {error}")
        return 1

Viewport Events Network

gsp_network.viewport_events.viewport_events_network

NetworkRenderer event handler for viewport events.

ViewportEventsNetwork

Bases: gsp.types.viewport_events_base.ViewportEventsBase

NetworkRenderer event handler for viewport.

Code heavily inspired from ViewportEventsMatplotlib

Source code in src/gsp_network/viewport_events/viewport_events_network.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
class ViewportEventsNetwork(ViewportEventsBase):
    """NetworkRenderer event handler for viewport.

    Code heavily inspired from ViewportEventsMatplotlib
    """

    __slots__ = [
        "_renderer",
        "_viewport",
        "_has_key_focus",
        "_mpl_key_press_cid",
        "_mpl_key_release_cid",
        "_mpl_button_press_cid",
        "_mpl_button_release_cid",
        "_mpl_mouse_move_cid",
        "_mpl_mouse_scroll_cid",
        "_mpl_resize_cid",
    ]

    def __init__(self, renderer: NetworkRenderer, viewport: Viewport) -> None:
        """Initialize the event handler with a renderer and viewport.

        Args:
            renderer: NetworkRenderer associated with this event handler.
            viewport: Viewport associated with this event handler.
        """
        self._renderer = renderer
        """NetworkRenderer associated with this event handler"""
        self._viewport = viewport
        """viewport associated with this event handler"""
        self._has_key_focus = False
        """True if this viewport has the keyboard focus"""

        # Intanciate events
        self.key_press_event = Event[KeyboardEventCallback]()
        self.key_release_event = Event[KeyboardEventCallback]()
        self.button_press_event = Event[MouseEventCallback]()
        self.button_release_event = Event[MouseEventCallback]()
        self.mouse_move_event = Event[MouseEventCallback]()
        self.mouse_scroll_event = Event[MouseEventCallback]()
        self.canvas_resize_event = Event[CanvasResizeEventCallback]()

        # event connections
        mpl_canvas: matplotlib.backend_bases.FigureCanvasBase = self._renderer.get_mpl_figure().canvas
        self._mpl_key_press_cid = mpl_canvas.mpl_connect("key_press_event", typing.cast(Any, self._on_key_press))
        self._mpl_key_release_cid = mpl_canvas.mpl_connect("key_release_event", typing.cast(Any, self._on_key_release))
        self._mpl_button_press_cid = mpl_canvas.mpl_connect("button_press_event", typing.cast(Any, self._on_button_press))
        self._mpl_button_release_cid = mpl_canvas.mpl_connect("button_release_event", typing.cast(Any, self._on_button_release))
        self._mpl_mouse_move_cid = mpl_canvas.mpl_connect("motion_notify_event", typing.cast(Any, self._on_mouse_move))
        self._mpl_mouse_scroll_cid = mpl_canvas.mpl_connect("scroll_event", typing.cast(Any, self._on_mouse_scroll))
        self._mpl_resize_cid = mpl_canvas.mpl_connect("resize_event", typing.cast(Any, self._on_canvas_resize))

    def close(self):
        """Close the event handler and release resources."""
        mpl_canvas: matplotlib.backend_bases.FigureCanvasBase = self._renderer.get_mpl_figure().canvas
        if self._mpl_key_press_cid is not None:
            mpl_canvas.mpl_disconnect(self._mpl_key_press_cid)
            self._mpl_key_press_cid = None
        if self._mpl_key_release_cid is not None:
            mpl_canvas.mpl_disconnect(self._mpl_key_release_cid)
            self._mpl_key_release_cid = None
        if self._mpl_button_press_cid is not None:
            mpl_canvas.mpl_disconnect(self._mpl_button_press_cid)
            self._mpl_button_press_cid = None
        if self._mpl_button_release_cid is not None:
            mpl_canvas.mpl_disconnect(self._mpl_button_release_cid)
            self._mpl_button_release_cid = None
        if self._mpl_mouse_move_cid is not None:
            mpl_canvas.mpl_disconnect(self._mpl_mouse_move_cid)
            self._mpl_mouse_move_cid = None
        if self._mpl_mouse_scroll_cid is not None:
            mpl_canvas.mpl_disconnect(self._mpl_mouse_scroll_cid)
            self._mpl_mouse_scroll_cid = None
        if self._mpl_resize_cid is not None:
            mpl_canvas.mpl_disconnect(self._mpl_resize_cid)
            self._mpl_resize_cid = None

    # =============================================================================
    # Matplotlib event handler
    # =============================================================================

    def _on_key_press(self, mpl_key_event: matplotlib.backend_bases.KeyEvent) -> None:
        # discard events outside the viewport
        if self._has_key_focus is False:
            return
        # convert and dispatch event
        keyboard_event = self._mpl_key_event_to_gsp(mpl_key_event, EventType.KEY_PRESS)
        self.key_press_event.dispatch(keyboard_event)

    def _on_key_release(self, mpl_key_event: matplotlib.backend_bases.KeyEvent) -> None:
        # discard events outside the viewport
        if self._has_key_focus is False:
            return
        # convert and dispatch event
        keyboard_event = self._mpl_key_event_to_gsp(mpl_key_event, EventType.KEY_RELEASE)
        self.key_release_event.dispatch(keyboard_event)

    def _on_button_press(self, mpl_mouse_event: matplotlib.backend_bases.MouseEvent) -> None:
        # Set key focus if the event is inside the viewport, otherwise remove key focus
        if self._viewport_contains_mpl_mouse_event(mpl_mouse_event):
            self._has_key_focus = True
        else:
            self._has_key_focus = False

        # discard events outside the viewport
        if self._viewport_contains_mpl_mouse_event(mpl_mouse_event) is False:
            return

        # convert and dispatch event
        mouse_event = self._mpl_mouse_event_to_gsp(mpl_mouse_event, EventType.BUTTON_PRESS)
        self.button_press_event.dispatch(mouse_event)

    def _on_button_release(self, mpl_mouse_event: matplotlib.backend_bases.MouseEvent) -> None:
        # discard events outside the viewport
        if self._viewport_contains_mpl_mouse_event(mpl_mouse_event) is False:
            return
        # convert and dispatch event
        mouse_event = self._mpl_mouse_event_to_gsp(mpl_mouse_event, EventType.BUTTON_RELEASE)
        self.button_release_event.dispatch(mouse_event)

    def _on_mouse_move(self, mpl_mouse_event: matplotlib.backend_bases.MouseEvent) -> None:
        # discard events outside the viewport
        if self._viewport_contains_mpl_mouse_event(mpl_mouse_event) is False:
            return
        # convert and dispatch event
        mouse_event = self._mpl_mouse_event_to_gsp(mpl_mouse_event, EventType.MOUSE_MOVE)
        self.mouse_move_event.dispatch(mouse_event)

    def _on_mouse_scroll(self, mpl_mouse_event: matplotlib.backend_bases.MouseEvent) -> None:
        # discard events outside the viewport
        if self._viewport_contains_mpl_mouse_event(mpl_mouse_event) is False:
            return
        # convert and dispatch event
        mouse_event = self._mpl_mouse_event_to_gsp(mpl_mouse_event, EventType.MOUSE_SCROLL)
        self.mouse_scroll_event.dispatch(mouse_event)

    def _on_canvas_resize(self, mpl_resize_event: matplotlib.backend_bases.ResizeEvent) -> None:
        # dispatch canvas resize event
        canvas_resize_event = CanvasResizeEvent(
            viewport_uuid=self._viewport.get_uuid(),
            event_type=EventType.CANVAS_RESIZE,
            canvas_width_px=mpl_resize_event.width,
            canvas_height_px=mpl_resize_event.height,
        )
        self.canvas_resize_event.dispatch(canvas_resize_event)

    # =============================================================================
    #
    # =============================================================================

    def _viewport_contains_mpl_mouse_event(self, mpl_mouse_event: matplotlib.backend_bases.MouseEvent) -> bool:
        """Check if the matplotlib mouse event is inside this viewport.

        Args:
            mpl_mouse_event: Matplotlib mouse event.

        Returns:
            True if the mouse event is inside this viewport, False otherwise.
        """
        mouse_x = mpl_mouse_event.x / UnitUtils.device_pixel_ratio()
        mouse_y = mpl_mouse_event.y / UnitUtils.device_pixel_ratio()
        if mouse_x < self._viewport.get_x():
            return False
        if mouse_x >= self._viewport.get_x() + self._viewport.get_width():
            return False
        if mouse_y < self._viewport.get_y():
            return False
        if mouse_y >= self._viewport.get_y() + self._viewport.get_height():
            return False
        return True

    # =============================================================================
    # Conversion matplotlib event to gsp events
    # =============================================================================
    def _mpl_key_event_to_gsp(self, mpl_key_event: matplotlib.backend_bases.KeyEvent, event_type: EventType) -> KeyEvent:
        assert mpl_key_event.key is not None
        keyboard_event = KeyEvent(
            viewport_uuid=self._viewport.get_uuid(),
            event_type=event_type,
            key_name=mpl_key_event.key,
        )
        return keyboard_event

    def _mpl_mouse_event_to_gsp(self, mpl_mouse_event: matplotlib.backend_bases.MouseEvent, event_type: EventType) -> MouseEvent:
        # Sanity check
        assert self._viewport_contains_mpl_mouse_event(mpl_mouse_event), "Mouse event is outside the viewport"

        mouse_x = mpl_mouse_event.x / UnitUtils.device_pixel_ratio()
        mouse_y = mpl_mouse_event.y / UnitUtils.device_pixel_ratio()
        x_viewport = mouse_x - self._viewport.get_x()
        y_viewport = mouse_y - self._viewport.get_y()
        mouse_event = MouseEvent(
            viewport_uuid=self._viewport.get_uuid(),
            event_type=event_type,
            x_ndc=(x_viewport / self._viewport.get_width() - 0.5) * 2.0,
            y_ndc=(y_viewport / self._viewport.get_height() - 0.5) * 2.0,
            left_button=mpl_mouse_event.button == 1,
            middle_button=mpl_mouse_event.button == 2,
            right_button=mpl_mouse_event.button == 3,
            scroll_steps=mpl_mouse_event.step if hasattr(mpl_mouse_event, "step") else 0.0,
        )
        return mouse_event

__init__(renderer: NetworkRenderer, viewport: Viewport) -> None

Initialize the event handler with a renderer and viewport.

Parameters:

Name Type Description Default
renderer gsp_network.renderer.NetworkRenderer

NetworkRenderer associated with this event handler.

 required
viewport gsp.core.Viewport

Viewport associated with this event handler.

 required
Source code in src/gsp_network/viewport_events/viewport_events_network.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def __init__(self, renderer: NetworkRenderer, viewport: Viewport) -> None:
    """Initialize the event handler with a renderer and viewport.

    Args:
        renderer: NetworkRenderer associated with this event handler.
        viewport: Viewport associated with this event handler.
    """
    self._renderer = renderer
    """NetworkRenderer associated with this event handler"""
    self._viewport = viewport
    """viewport associated with this event handler"""
    self._has_key_focus = False
    """True if this viewport has the keyboard focus"""

    # Intanciate events
    self.key_press_event = Event[KeyboardEventCallback]()
    self.key_release_event = Event[KeyboardEventCallback]()
    self.button_press_event = Event[MouseEventCallback]()
    self.button_release_event = Event[MouseEventCallback]()
    self.mouse_move_event = Event[MouseEventCallback]()
    self.mouse_scroll_event = Event[MouseEventCallback]()
    self.canvas_resize_event = Event[CanvasResizeEventCallback]()

    # event connections
    mpl_canvas: matplotlib.backend_bases.FigureCanvasBase = self._renderer.get_mpl_figure().canvas
    self._mpl_key_press_cid = mpl_canvas.mpl_connect("key_press_event", typing.cast(Any, self._on_key_press))
    self._mpl_key_release_cid = mpl_canvas.mpl_connect("key_release_event", typing.cast(Any, self._on_key_release))
    self._mpl_button_press_cid = mpl_canvas.mpl_connect("button_press_event", typing.cast(Any, self._on_button_press))
    self._mpl_button_release_cid = mpl_canvas.mpl_connect("button_release_event", typing.cast(Any, self._on_button_release))
    self._mpl_mouse_move_cid = mpl_canvas.mpl_connect("motion_notify_event", typing.cast(Any, self._on_mouse_move))
    self._mpl_mouse_scroll_cid = mpl_canvas.mpl_connect("scroll_event", typing.cast(Any, self._on_mouse_scroll))
    self._mpl_resize_cid = mpl_canvas.mpl_connect("resize_event", typing.cast(Any, self._on_canvas_resize))

close()

Close the event handler and release resources.

Source code in src/gsp_network/viewport_events/viewport_events_network.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def close(self):
    """Close the event handler and release resources."""
    mpl_canvas: matplotlib.backend_bases.FigureCanvasBase = self._renderer.get_mpl_figure().canvas
    if self._mpl_key_press_cid is not None:
        mpl_canvas.mpl_disconnect(self._mpl_key_press_cid)
        self._mpl_key_press_cid = None
    if self._mpl_key_release_cid is not None:
        mpl_canvas.mpl_disconnect(self._mpl_key_release_cid)
        self._mpl_key_release_cid = None
    if self._mpl_button_press_cid is not None:
        mpl_canvas.mpl_disconnect(self._mpl_button_press_cid)
        self._mpl_button_press_cid = None
    if self._mpl_button_release_cid is not None:
        mpl_canvas.mpl_disconnect(self._mpl_button_release_cid)
        self._mpl_button_release_cid = None
    if self._mpl_mouse_move_cid is not None:
        mpl_canvas.mpl_disconnect(self._mpl_mouse_move_cid)
        self._mpl_mouse_move_cid = None
    if self._mpl_mouse_scroll_cid is not None:
        mpl_canvas.mpl_disconnect(self._mpl_mouse_scroll_cid)
        self._mpl_mouse_scroll_cid = None
    if self._mpl_resize_cid is not None:
        mpl_canvas.mpl_disconnect(self._mpl_resize_cid)
        self._mpl_resize_cid = None

Animator Network

gsp_network.animator.animator_network

Network-based animator for GSP scenes.

Provides animation capabilities using a NetworkRenderer backend with support for real-time animation display and video export.

AnimatorNetwork

Bases: gsp.types.animator_base.AnimatorBase

Animator for GSP scenes using a network renderer.

Manages animation loops with callback functions that update visuals each frame. Supports real-time display and video export in various formats.

Source code in src/gsp_network/animator/animator_network.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
class AnimatorNetwork(AnimatorBase):
    """Animator for GSP scenes using a network renderer.

    Manages animation loops with callback functions that update visuals each frame.
    Supports real-time display and video export in various formats.
    """

    def __init__(
        self,
        network_renderer: NetworkRenderer,
        fps: int = 50,
        video_duration: float = 10.0,
        video_path: str | None = None,
        video_writer: str | None = None,
    ):
        """Initialize the network animator.

        Args:
            network_renderer: The network renderer to use for rendering frames.
            fps: Target frames per second for the animation.
            video_duration: Total duration of the animation in seconds.
            video_path: Path where the video should be saved. If None, no video is saved.
            video_writer: Video writer to use ("ffmpeg" or "pillow"). If None, auto-detected from extension.

        Raises:
            ValueError: If the video format is not supported.
        """
        self._callbacks: list[AnimatorFunc] = []
        self._network_renderer = network_renderer
        self._fps = fps
        self._video_duration = video_duration
        self._video_path = video_path
        self._video_writer: str | None = video_writer
        self._time_last_update: float | None = None

        self._funcAnimation: matplotlib.animation.FuncAnimation | None = None

        self._canvas: Canvas | None = None
        self._viewports: Sequence[Viewport] | None = None
        self._visuals: Sequence[VisualBase] | None = None
        self._model_matrices: Sequence[TransBuf] | None = None
        self._cameras: Sequence[Camera] | None = None

        self.on_video_saved = Event[VideoSavedCalledback]()
        """Event triggered when the video has been successfully saved to disk."""

        # guess the video writer from the file extension if not provided
        if self._video_path is not None:
            if video_writer is not None:
                self._video_writer = video_writer
            else:
                video_ext = os.path.splitext(self._video_path)[1].lower()
                if video_ext in [".mp4", ".m4v", ".mov"]:
                    self._video_writer = "ffmpeg"
                elif video_ext in [".gif", ".apng", ".webp"]:
                    self._video_writer = "pillow"
                else:
                    raise ValueError(f"Unsupported video format: {video_ext}")

    # =============================================================================
    # .add_callback/.remove_callback/.decorator
    # =============================================================================

    def add_callback(self, func: AnimatorFunc) -> None:
        """Add a callback to the animation loop.

        Args:
            func: The animator function to call on each frame.
        """
        self._callbacks.append(func)

    def remove_callback(self, func: AnimatorFunc) -> None:
        """Remove a callback from the animation loop.

        Args:
            func: The animator function to remove.
        """
        self._callbacks.remove(func)

    def event_listener(self, func: AnimatorFunc) -> AnimatorFunc:
        """Decorator to add a callback to the animation loop.

        Args:
            func: The animator function to register as a callback.

        Returns:
            The wrapped animator function.

        Usage:
            ```python
                @animation_loop.event_listener
                def my_callback(delta_time: float) -> Sequence[VisualBase]:
                    ...

                # later, if needed
                animation_loop.remove_callback(my_callback)
            ```
        """
        self.add_callback(func)

        def wrapper(delta_time: float) -> Sequence[VisualBase]:
            # print("Before the function runs")
            result = func(delta_time)
            # print("After the function runs")
            return result

        return wrapper

    # =============================================================================
    # .start()
    # =============================================================================
    def start(self, viewports: Sequence[Viewport], visuals: Sequence[VisualBase], model_matrices: Sequence[TransBuf], cameras: Sequence[Camera]) -> None:
        """Start the animation loop.

        Begins rendering frames using registered callbacks to update visuals.
        In test mode (GSP_TEST=True), saves a single preview image instead of animating.

        Args:
            viewports: Sequence of viewport regions to render into.
            visuals: Sequence of visual elements to render and animate.
            model_matrices: Sequence of model transformation matrices.
            cameras: Sequence of cameras defining view and projection.
        """
        self._canvas = self._network_renderer.get_canvas()
        self._viewports = viewports
        self._visuals = visuals
        self._model_matrices = model_matrices
        self._cameras = cameras
        self._time_last_update = time.time()

        # =============================================================================
        # Render the image once
        # =============================================================================

        self._network_renderer.render(viewports, visuals, model_matrices, cameras)

        # =============================================================================
        # Handle GSP_TEST=True
        # =============================================================================

        # detect if we are in not interactive mode - used during testing
        in_test = "GSP_TEST" in os.environ and os.environ["GSP_TEST"] == "True"

        # if we are not in interactive mode, save a preview image and return
        if in_test == True:
            # notify all animator callbacks
            changed_visuals: list[VisualBase] = []
            for animator_callback in self._callbacks:
                _changed_visuals = animator_callback(1.0 / self._fps)
                changed_visuals.extend(_changed_visuals)

            # render the scene to get the new image
            image_png_data = self._network_renderer.render(viewports, visuals, model_matrices, cameras)
            # get the main script name
            main_script_name = os.path.basename(__main__.__file__) if hasattr(__main__, "__file__") else "interactive"
            main_script_basename = os.path.splitext(main_script_name)[0]
            # buid the output image path
            image_path = os.path.join(__dirname__, "../../../examples/output", f"{main_script_basename}_animator_network.png")
            image_path = os.path.abspath(image_path)
            # save image_png_data in a image file
            with open(image_path, "wb") as image_file:
                image_file.write(image_png_data)
            # log the event
            print(f"Saved animation preview image to: {image_path}")
            return

        # NOTE: here we are in interactive mode!!

        # =============================================================================
        # Initialize the animation
        # =============================================================================

        figure = self._network_renderer.get_mpl_figure()
        self._funcAnimation = matplotlib.animation.FuncAnimation(
            figure, self._mpl_animate, frames=int(self._video_duration * self._fps), interval=1000.0 / self._fps
        )

        # save the animation if a path is provided
        if self._video_path is not None:
            self._funcAnimation.save(self._video_path, writer=self._video_writer, fps=self._fps)
            # Dispatch the video saved event
            self.on_video_saved.dispatch()

        # =============================================================================
        # Show the animation
        # =============================================================================

        self._network_renderer.show()

    # =============================================================================
    # .stop()
    # =============================================================================
    def stop(self):
        """Stop the animation loop.

        Stops the Matplotlib animation timer and clears internal state.
        """
        self._canvas = None
        self._viewports = None
        self._time_last_update = None

        # stop the animation function timer
        if self._funcAnimation is not None:
            self._funcAnimation.event_source.stop()
            self._funcAnimation = None

    # =============================================================================
    # ._mpl_animate()
    # =============================================================================

    def _mpl_animate(self, frame_index: int) -> list[matplotlib.artist.Artist]:
        """Internal callback for Matplotlib animation.

        Called by Matplotlib's FuncAnimation on each frame to update the display.

        Args:
            frame_index: The current frame number in the animation sequence.

        Returns:
            List of Matplotlib artists that were updated.
        """
        # sanity checks
        assert self._canvas is not None, "Canvas MUST be set during the animation"
        assert self._viewports is not None, "Viewports MUST be set during the animation"
        assert self._visuals is not None, "Visuals MUST be set during the animation"
        assert self._model_matrices is not None, "Model matrices MUST be set during the animation"
        assert self._cameras is not None, "Cameras MUST be set during the animation"

        # compute delta time
        present = time.time()
        delta_time = (present - self._time_last_update) if self._time_last_update is not None else (1 / self._fps)
        self._time_last_update = present

        # notify all animator callbacks
        changed_visuals: list[VisualBase] = []
        for callback in self._callbacks:
            _changed_visuals = callback(delta_time)
            changed_visuals.extend(_changed_visuals)

        # Render the scene to update the visuals
        if len(changed_visuals) > 0:
            self._network_renderer.render(self._viewports, self._visuals, self._model_matrices, self._cameras)

        # return the changed mpl artists
        return [self._network_renderer._axes_image]

on_video_saved = Event[VideoSavedCalledback]() instance-attribute

Event triggered when the video has been successfully saved to disk.

__init__(network_renderer: NetworkRenderer, fps: int = 50, video_duration: float = 10.0, video_path: str | None = None, video_writer: str | None = None)

Initialize the network animator.

Parameters:

Name Type Description Default
network_renderer gsp_network.renderer.NetworkRenderer

The network renderer to use for rendering frames.

 required
fps int

Target frames per second for the animation.

 50
video_duration float

Total duration of the animation in seconds.

 10.0
video_path str | None

Path where the video should be saved. If None, no video is saved.

 None
video_writer str | None

Video writer to use ("ffmpeg" or "pillow"). If None, auto-detected from extension.

 None

Raises:

Type Description
ValueError

If the video format is not supported.
Source code in src/gsp_network/animator/animator_network.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def __init__(
    self,
    network_renderer: NetworkRenderer,
    fps: int = 50,
    video_duration: float = 10.0,
    video_path: str | None = None,
    video_writer: str | None = None,
):
    """Initialize the network animator.

    Args:
        network_renderer: The network renderer to use for rendering frames.
        fps: Target frames per second for the animation.
        video_duration: Total duration of the animation in seconds.
        video_path: Path where the video should be saved. If None, no video is saved.
        video_writer: Video writer to use ("ffmpeg" or "pillow"). If None, auto-detected from extension.

    Raises:
        ValueError: If the video format is not supported.
    """
    self._callbacks: list[AnimatorFunc] = []
    self._network_renderer = network_renderer
    self._fps = fps
    self._video_duration = video_duration
    self._video_path = video_path
    self._video_writer: str | None = video_writer
    self._time_last_update: float | None = None

    self._funcAnimation: matplotlib.animation.FuncAnimation | None = None

    self._canvas: Canvas | None = None
    self._viewports: Sequence[Viewport] | None = None
    self._visuals: Sequence[VisualBase] | None = None
    self._model_matrices: Sequence[TransBuf] | None = None
    self._cameras: Sequence[Camera] | None = None

    self.on_video_saved = Event[VideoSavedCalledback]()
    """Event triggered when the video has been successfully saved to disk."""

    # guess the video writer from the file extension if not provided
    if self._video_path is not None:
        if video_writer is not None:
            self._video_writer = video_writer
        else:
            video_ext = os.path.splitext(self._video_path)[1].lower()
            if video_ext in [".mp4", ".m4v", ".mov"]:
                self._video_writer = "ffmpeg"
            elif video_ext in [".gif", ".apng", ".webp"]:
                self._video_writer = "pillow"
            else:
                raise ValueError(f"Unsupported video format: {video_ext}")

add_callback(func: AnimatorFunc) -> None

Add a callback to the animation loop.

Parameters:

Name Type Description Default
func gsp.types.animator_types.AnimatorFunc

The animator function to call on each frame.

 required
Source code in src/gsp_network/animator/animator_network.py 
 94
 95
 96
 97
 98
 99
100
def add_callback(self, func: AnimatorFunc) -> None:
    """Add a callback to the animation loop.

    Args:
        func: The animator function to call on each frame.
    """
    self._callbacks.append(func)

event_listener(func: AnimatorFunc) -> AnimatorFunc

Decorator to add a callback to the animation loop.

Parameters:

Name Type Description Default
func gsp.types.animator_types.AnimatorFunc

The animator function to register as a callback.

 required

Returns:

Type Description
gsp.types.animator_types.AnimatorFunc

The wrapped animator function.
Usage 
    @animation_loop.event_listener
    def my_callback(delta_time: float) -> Sequence[VisualBase]:
        ...

    # later, if needed
    animation_loop.remove_callback(my_callback)
Source code in src/gsp_network/animator/animator_network.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def event_listener(self, func: AnimatorFunc) -> AnimatorFunc:
    """Decorator to add a callback to the animation loop.

    Args:
        func: The animator function to register as a callback.

    Returns:
        The wrapped animator function.

    Usage:
        ```python
            @animation_loop.event_listener
            def my_callback(delta_time: float) -> Sequence[VisualBase]:
                ...

            # later, if needed
            animation_loop.remove_callback(my_callback)
        ```
    """
    self.add_callback(func)

    def wrapper(delta_time: float) -> Sequence[VisualBase]:
        # print("Before the function runs")
        result = func(delta_time)
        # print("After the function runs")
        return result

    return wrapper

remove_callback(func: AnimatorFunc) -> None

Remove a callback from the animation loop.

Parameters:

Name Type Description Default
func gsp.types.animator_types.AnimatorFunc

The animator function to remove.

 required
Source code in src/gsp_network/animator/animator_network.py 
102
103
104
105
106
107
108
def remove_callback(self, func: AnimatorFunc) -> None:
    """Remove a callback from the animation loop.

    Args:
        func: The animator function to remove.
    """
    self._callbacks.remove(func)

start(viewports: Sequence[Viewport], visuals: Sequence[VisualBase], model_matrices: Sequence[TransBuf], cameras: Sequence[Camera]) -> None

Start the animation loop.

Begins rendering frames using registered callbacks to update visuals. In test mode (GSP_TEST=True), saves a single preview image instead of animating.

Parameters:

Name Type Description Default
viewports typing.Sequence[gsp.core.viewport.Viewport]

Sequence of viewport regions to render into.

 required
visuals typing.Sequence[gsp.types.visual_base.VisualBase]

Sequence of visual elements to render and animate.

 required
model_matrices typing.Sequence[gsp.types.transbuf.TransBuf]

Sequence of model transformation matrices.

 required
cameras typing.Sequence[gsp.core.camera.Camera]

Sequence of cameras defining view and projection.

 required
Source code in src/gsp_network/animator/animator_network.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def start(self, viewports: Sequence[Viewport], visuals: Sequence[VisualBase], model_matrices: Sequence[TransBuf], cameras: Sequence[Camera]) -> None:
    """Start the animation loop.

    Begins rendering frames using registered callbacks to update visuals.
    In test mode (GSP_TEST=True), saves a single preview image instead of animating.

    Args:
        viewports: Sequence of viewport regions to render into.
        visuals: Sequence of visual elements to render and animate.
        model_matrices: Sequence of model transformation matrices.
        cameras: Sequence of cameras defining view and projection.
    """
    self._canvas = self._network_renderer.get_canvas()
    self._viewports = viewports
    self._visuals = visuals
    self._model_matrices = model_matrices
    self._cameras = cameras
    self._time_last_update = time.time()

    # =============================================================================
    # Render the image once
    # =============================================================================

    self._network_renderer.render(viewports, visuals, model_matrices, cameras)

    # =============================================================================
    # Handle GSP_TEST=True
    # =============================================================================

    # detect if we are in not interactive mode - used during testing
    in_test = "GSP_TEST" in os.environ and os.environ["GSP_TEST"] == "True"

    # if we are not in interactive mode, save a preview image and return
    if in_test == True:
        # notify all animator callbacks
        changed_visuals: list[VisualBase] = []
        for animator_callback in self._callbacks:
            _changed_visuals = animator_callback(1.0 / self._fps)
            changed_visuals.extend(_changed_visuals)

        # render the scene to get the new image
        image_png_data = self._network_renderer.render(viewports, visuals, model_matrices, cameras)
        # get the main script name
        main_script_name = os.path.basename(__main__.__file__) if hasattr(__main__, "__file__") else "interactive"
        main_script_basename = os.path.splitext(main_script_name)[0]
        # buid the output image path
        image_path = os.path.join(__dirname__, "../../../examples/output", f"{main_script_basename}_animator_network.png")
        image_path = os.path.abspath(image_path)
        # save image_png_data in a image file
        with open(image_path, "wb") as image_file:
            image_file.write(image_png_data)
        # log the event
        print(f"Saved animation preview image to: {image_path}")
        return

    # NOTE: here we are in interactive mode!!

    # =============================================================================
    # Initialize the animation
    # =============================================================================

    figure = self._network_renderer.get_mpl_figure()
    self._funcAnimation = matplotlib.animation.FuncAnimation(
        figure, self._mpl_animate, frames=int(self._video_duration * self._fps), interval=1000.0 / self._fps
    )

    # save the animation if a path is provided
    if self._video_path is not None:
        self._funcAnimation.save(self._video_path, writer=self._video_writer, fps=self._fps)
        # Dispatch the video saved event
        self.on_video_saved.dispatch()

    # =============================================================================
    # Show the animation
    # =============================================================================

    self._network_renderer.show()

stop()

Stop the animation loop.

Stops the Matplotlib animation timer and clears internal state.

Source code in src/gsp_network/animator/animator_network.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
def stop(self):
    """Stop the animation loop.

    Stops the Matplotlib animation timer and clears internal state.
    """
    self._canvas = None
    self._viewports = None
    self._time_last_update = None

    # stop the animation function timer
    if self._funcAnimation is not None:
        self._funcAnimation.event_source.stop()
        self._funcAnimation = None