14549006187

Committed 19 Apr 2025 12:14PM UTC coverage: 48.084% (-3.9%) from 51.953%

Build # 14549006187

Build Type

push

github

Committed by

sabvdf

Commit Message

Adds local CI check; WIP Reworking update logic to fix performance issues when running with Streamlit dashboard.

Run Details

12 of 79 new or added lines in 7 files covered. (15.19%)

64 existing lines in 4 files now uncovered.

276 of 574 relevant lines covered (48.08%)

0.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

45.7

/datasim/plot.py

from abc import ABC
from enum import Enum
from typing import Any, Generic, List, Optional, cast
from pandas import DataFrame
from plotly.graph_objs._figure import Figure

import numpy as np
import plotly.express as px
import streamlit as st

from .dashboard import Dashboard
from .entity import Entity
from .types import Number
from .queue import Queue
from .resource import Resource


class PlotType(Enum):
    """The type of plot to render."""

    bar = "bar"
    line = "line"
    pie = "pie"
    scatter = "scatter"

    def __str__(self) -> str:
        """Get a string representation of the plot type."""
        match self:
            case PlotType.bar:
                return "Bar chart"
            case PlotType.line:
                return "Line graph"
            case PlotType.pie:
                return "Pie chart"
            case PlotType.scatter:
                return "Scatter plot"


class PlotData(ABC):
    """Abstract superclass of different types of data to plot."""

    plot_type: PlotType
    title: Optional[str]
    trace: Optional[Figure] = None
    dashboard: Optional[Dashboard] = None
    plot: Optional[Any] = None
    legend_x: str
    legend_y: str

    def __init__(
        self,
        plot_type: PlotType,
        title: Optional[str],
        legend_x: str = "x",
        legend_y: str = "y",
    ):
        """Create a data source to plot from.

        Args:
            plot_type ("scatter", "line", "bar", "pie"): Type of plot to render.
            title (str, optional): Title to use over the plot. Defaults to None.
        """
        from .world import World

        self.plot_type = plot_type
        self.title = title
        self.legend_x = legend_x
        self.legend_y = legend_y

        self._buffer_size = max(10000, World.end_tick)
        self._buffer_index = 0
        self._x_buffer = np.zeros(self._buffer_size)
        self._y_buffer = np.zeros(self._buffer_size)

    @property
    def _data_frame(self):
        return DataFrame(
            {
                self.legend_x: self._x_buffer[: self._buffer_index],
                self.legend_y: self._y_buffer[: self._buffer_index],
            }
        )

    def _update_traces(self):
        if self.dashboard is None:
            if "dashboard" in st.session_state:
                self.dashboard = cast(Dashboard, st.session_state.dashboard)

        if self.dashboard is None:
            return

        match self.plot_type:
            case PlotType.bar:
                self.trace = cast(
                    Figure,
                    px.bar(
                        self._data_frame,
                        title=self.title,
                        x=self.legend_x,
                        y=self.legend_y,
                    ),
                )
            case PlotType.line:
                self.trace = cast(
                    Figure,
                    px.line(
                        self._data_frame,
                        markers=True,
                        title=self.title,
                        x=self.legend_x,
                        y=self.legend_y,
                    ),
                )
            case PlotType.pie:
                pass
            case PlotType.scatter:
                self.trace = cast(
                    Figure,
                    px.scatter(
                        self._data_frame,
                        x=self.legend_x,
                        y=self.legend_y,
                        title=self.title,
                    ),
                )

        if (
            self.plot
            and self.plot.id not in self.dashboard.plots
            and self.trace is not None
        ):
            self.dashboard.plots[self.plot.id] = self.trace

    def _tick(self):
        pass


class XYPlotData(PlotData):
    """Data with x and y values as float."""

    def __init__(
        self,
        data_x: List[float] = [],
        data_y: List[float] = [],
        plot_type: PlotType = PlotType.line,
        title: Optional[str] = None,
        legend_x: str = "x",
        legend_y: str = "y",
    ):
        """Create a data source from x and y lists of floats.

        Args:
            data_x (List[float], optional): x values. Defaults to [] to start with an empty data set.
            data_y (List[float], optional): y values. Defaults to [] to start with an empty data set.
            plot_type ("scatter", "line", "bar", "pie", optional): Type of plot to show. Defaults to "line".
            title (Optional[str], optional): Title to use over the plot. Defaults to None.
        """
        super().__init__(plot_type, title, legend_x, legend_y)
        self._x_buffer[: len(data_x)] = data_x
        self._y_buffer[: len(data_y)] = data_y

    def append(self, x: float, y: float):
        """Add a data point to this data set.

        Args:
            x (float): x value of the data point.
            y (float): y value of the data point.
        """
        self._x_buffer[self._buffer_index] = x
        self._y_buffer[self._buffer_index] = y
        self._buffer_index += 1


class CategoryPlotData(PlotData):
    """Data with named categories with float values."""

    labels: List[str]
    values: List[float]

    def __init__(
        self,
        data_x: List[str] = [],
        data_y: List[float] = [],
        plot_type: PlotType = PlotType.line,
        title: Optional[str] = None,
        legend_x: str = "category",
        legend_y: str = "value",
    ):
        """Create a data source from x as categories and y as float values.

        Args:
            data_x (List[str], optional): labels. Defaults to [] to start with an empty data set.
            data_y (List[float], optional): values for each label. Defaults to [] to start with an empty data set.
            plot_type ("scatter", "line", "bar", "pie", optional): Type of plot to show. Defaults to "line".
            title (Optional[str], optional): Title to use over the plot. Defaults to None.
        """
        super().__init__(plot_type, title, legend_x, legend_y)
        self.data_x = data_x
        self.data_y = data_y

    def append(self, label: str, value: float):
        """Add a data point to this data set.

        Args:
            label (str): label of the data point.
            value (float): value of the data point.
        """
        # self.labels.append(label)
        # self.values.append(value)
        # TODO


class NPPlotData(PlotData):
    """Data with Numpy array as source."""

    data: np.ndarray

    def __init__(
        self,
        data: np.ndarray,
        plot_type: PlotType = PlotType.line,
        title: Optional[str] = None,
        legend_x: str = "x",
        legend_y: str = "y",
    ):
        """Create a data source from a Numpy array.

        Args:
            data (:class:`np.ndarray`): Array of data points. Shape should correspond to the dimensions of the plot
                (2 columns for 2D plots, 3 columns for 3D plots).
            plot_type (:class:`PlotType`, optional): Type of plot to show. Defaults to `"line"`.
            title (`str`, optional): Title to use over the plot. Defaults to `None`.

        Raises:
            `TypeError`: When trying to take from a capacity resource without specifying an amount.
        """
        super().__init__(plot_type, title, legend_x, legend_y)
        if data.shape[1] != 2:  # TODO add 3 when adding 3D plots
            raise ValueError("")
        self.data = data

    @property
    def _data_frame(self):
        return DataFrame(
            {
                self.legend_x: self.data[0],
                self.legend_y: self.data[1],
            }
        )


class ResourcePlotData(Generic[Number], PlotData):
    """Data source from watching the amount of a :class:`Resource`."""

    source: Resource[Number]
    frequency: int

    def __init__(
        self,
        source: Resource[Number],
        frequency: int = 1,
        plot_type: PlotType = PlotType.line,
        title: Optional[str] = None,
        legend_x: str = "seconds",
        legend_y: str = "amount",
    ):
        """Create a data source from watching the amount of a :class:`Resource`.

        Args:
            data (:class:`Resource`): Source :class:`Resource`.
            frequency (int, optional): Frequency in ticks to add data points.
                Defaults to 1, meaning a point gets added every tick.
            plot_type ("scatter", "line", "bar", "pie", optional): Type of plot to show. Defaults to "line".
            title (Optional[str], optional): Title to use over the plot. Defaults to None.
        """
        super().__init__(plot_type, title, legend_x, legend_y)
        self.source = source
        self.frequency = frequency

    def _tick(self):
        from .world import World

        if World.ticks % self.frequency == 0:
            self._x_buffer[self._buffer_index] = World.seconds()
            self._y_buffer[self._buffer_index] = self.source.amount
            self._buffer_index += 1


class QueuePlotData(PlotData):
    """Data source from watching the size of a :class:`Queue`."""

    source: Queue
    frequency: int

    def __init__(
        self,
        source: Queue,
        frequency: int = 1,
        plot_type: PlotType = PlotType.line,
        title: Optional[str] = None,
        legend_x: str = "seconds",
        legend_y: str = "length",
    ):
        """Create a data source from watching the size of a :class:`Queue`.

        Args:
            data (:class:`Queue`): Source :class:`Queue`.
            frequency (int, optional): Frequency in ticks to add data points.
                Defaults to 1, meaning a point gets added every tick.
            plot_type ("scatter", "line", "bar", "pie", optional): Type of plot to show. Defaults to "line".
            title (Optional[str], optional): Title to use over the plot. Defaults to None.
        """
        super().__init__(plot_type, title, legend_x, legend_y)
        self.source = source
        self.frequency = frequency

    def _tick(self):
        from .world import World

        if World.ticks % self.frequency == 0:
            self._x_buffer[self._buffer_index] = World.seconds()
            self._y_buffer[self._buffer_index] = len(self.source)
            self._buffer_index += 1


class StatePlotData(PlotData):
    """Data source from watching the state of an :class:`Entity`."""

    data: Entity
    frequency: int

    def __init__(
        self,
        data: Entity,
        frequency: int = 1,
        plot_type: PlotType = PlotType.line,
        title: Optional[str] = None,
    ):
        """Create a data source from watching the state of an :class:`Entity`.

        Args:
            data (:class:`Entity`): Source :class:`Entity`.
            frequency (int, optional): Frequency in ticks to add data points.
                Defaults to 1, meaning a point gets added every tick.
            plot_type ("scatter", "line", "bar", "pie", optional): Type of plot to show. Defaults to "line".
            title (Optional[str], optional): Title to use over the plot. Defaults to None.
        """
        super().__init__(plot_type, title)
        self.data = data
        self.frequency = frequency

        # TODO


class Plot:
    """Base class for easily updating data for plots to be made on the dashboard."""

    id: str
    title: Optional[str]
    figure: Figure
    data: List[PlotData]

    def __init__(self, id: str, *args: PlotData):
        """Create a plot to add to the dashboard using `World.add_plot()`.

        Args:
            id (str): identifier, needs to be unique.
            *args (:class:`PlotData`): Data to start the plot with.
        """
        self.id = id
        self.data = []
        for arg in args:
            self.add_trace(arg)

    def __getitem__(self, key: int) -> PlotData:
        """Get a reference to a data set from this plot.

        Args:
            key (int): Index of the data set.

        Returns:
            tuple(list[float] | list[str], list[float]): Data set at index.
        """
        return self.data[key]

    def _tick(self):
        for data in self.data:
            data._tick()

    def add_trace(self, data: PlotData) -> int:
        """Add a data trace to the plot.

        Args:
            data (PlotData): Data set.

        Returns:
            int: Index of the added data set.
        """
        index: int = len(self.data)
        data.plot = self
        self.data.append(data)
        return index

    def _update(self):
        for data in self.data:
            data._update_traces()

1	from abc import ABC	1✔
2	from enum import Enum	1✔
3	from typing import Any, Generic, List, Optional, cast	1✔
4	from pandas import DataFrame	1✔
5	from plotly.graph_objs._figure import Figure	1✔
6
7	import numpy as np	1✔
8	import plotly.express as px	1✔
9	import streamlit as st	1✔
10
11	from .dashboard import Dashboard	1✔
12	from .entity import Entity	1✔
13	from .types import Number	1✔
14	from .queue import Queue	1✔
15	from .resource import Resource	1✔
16
17
18	class PlotType(Enum):	1✔
19	"""The type of plot to render."""
20
21	bar = "bar"	1✔
22	line = "line"	1✔
23	pie = "pie"	1✔
24	scatter = "scatter"	1✔
25
26	def __str__(self) -> str:	1✔
27	"""Get a string representation of the plot type."""
NEW 28	match self:	×
NEW 29	case PlotType.bar:	×
NEW 30	return "Bar chart"	×
NEW 31	case PlotType.line:	×
NEW 32	return "Line graph"	×
NEW 33	case PlotType.pie:	×
NEW 34	return "Pie chart"	×
NEW 35	case PlotType.scatter:	×
NEW 36	return "Scatter plot"	×
37
38
39	class PlotData(ABC):	1✔
40	"""Abstract superclass of different types of data to plot."""
41
42	plot_type: PlotType	1✔
43	title: Optional[str]	1✔
44	trace: Optional[Figure] = None	1✔
45	dashboard: Optional[Dashboard] = None	1✔
46	plot: Optional[Any] = None	1✔
47	legend_x: str	1✔
48	legend_y: str	1✔
49
50	def __init__(	1✔
51	self,
52	plot_type: PlotType,
53	title: Optional[str],
54	legend_x: str = "x",
55	legend_y: str = "y",
56	):
57	"""Create a data source to plot from.
58
59	Args:
60	plot_type ("scatter", "line", "bar", "pie"): Type of plot to render.
61	title (str, optional): Title to use over the plot. Defaults to None.
62	"""
NEW 63	from .world import World	×
64
65	self.plot_type = plot_type	×
66	self.title = title	×
67	self.legend_x = legend_x	×
68	self.legend_y = legend_y	×
69
NEW 70	self._buffer_size = max(10000, World.end_tick)	×
NEW 71	self._buffer_index = 0	×
NEW 72	self._x_buffer = np.zeros(self._buffer_size)	×
NEW 73	self._y_buffer = np.zeros(self._buffer_size)	×
74
75	@property	1✔
76	def _data_frame(self):	1✔
NEW 77	return DataFrame(	×
78	{
79	self.legend_x: self._x_buffer[: self._buffer_index],
80	self.legend_y: self._y_buffer[: self._buffer_index],
81	}
82	)
83
84	def _update_traces(self):	1✔
85	if self.dashboard is None:	×
86	if "dashboard" in st.session_state:	×
87	self.dashboard = cast(Dashboard, st.session_state.dashboard)	×
88
89	if self.dashboard is None:	×
90	return	×
91
92	match self.plot_type:	×
NEW 93	case PlotType.bar:	×
94	self.trace = cast(	×
95	Figure,
96	px.bar(
97	self._data_frame,
98	title=self.title,
99	x=self.legend_x,
100	y=self.legend_y,
101	),
102	)
NEW 103	case PlotType.line:	×
104	self.trace = cast(	×
105	Figure,
106	px.line(
107	self._data_frame,
108	markers=True,
109	title=self.title,
110	x=self.legend_x,
111	y=self.legend_y,
112	),
113	)
NEW 114	case PlotType.pie:	×
115	pass	×
NEW 116	case PlotType.scatter:	×
117	self.trace = cast(	×
118	Figure,
119	px.scatter(
120	self._data_frame,
121	x=self.legend_x,
122	y=self.legend_y,
123	title=self.title,
124	),
125	)
126
127	if (	×
128	self.plot
129	and self.plot.id not in self.dashboard.plots
130	and self.trace is not None
131	):
132	self.dashboard.plots[self.plot.id] = self.trace	×
133
134	def _tick(self):	1✔
135	pass	×
136
137
138	class XYPlotData(PlotData):	1✔
139	"""Data with x and y values as float."""
140
141	def __init__(	1✔
142	self,
143	data_x: List[float] = [],
144	data_y: List[float] = [],
145	plot_type: PlotType = PlotType.line,
146	title: Optional[str] = None,
147	legend_x: str = "x",
148	legend_y: str = "y",
149	):
150	"""Create a data source from x and y lists of floats.
151
152	Args:
153	data_x (List[float], optional): x values. Defaults to [] to start with an empty data set.
154	data_y (List[float], optional): y values. Defaults to [] to start with an empty data set.
155	plot_type ("scatter", "line", "bar", "pie", optional): Type of plot to show. Defaults to "line".
156	title (Optional[str], optional): Title to use over the plot. Defaults to None.
157	"""
158	super().__init__(plot_type, title, legend_x, legend_y)	×
NEW 159	self._x_buffer[: len(data_x)] = data_x	×
NEW 160	self._y_buffer[: len(data_y)] = data_y	×
161
162	def append(self, x: float, y: float):	1✔
163	"""Add a data point to this data set.
164
165	Args:
166	x (float): x value of the data point.
167	y (float): y value of the data point.
168	"""
NEW 169	self._x_buffer[self._buffer_index] = x	×
NEW 170	self._y_buffer[self._buffer_index] = y	×
NEW 171	self._buffer_index += 1	×
172
173
174	class CategoryPlotData(PlotData):	1✔
175	"""Data with named categories with float values."""
176
177	labels: List[str]	1✔
178	values: List[float]	1✔
179
180	def __init__(	1✔
181	self,
182	data_x: List[str] = [],
183	data_y: List[float] = [],
184	plot_type: PlotType = PlotType.line,
185	title: Optional[str] = None,
186	legend_x: str = "category",
187	legend_y: str = "value",
188	):
189	"""Create a data source from x as categories and y as float values.
190
191	Args:
192	data_x (List[str], optional): labels. Defaults to [] to start with an empty data set.
193	data_y (List[float], optional): values for each label. Defaults to [] to start with an empty data set.
194	plot_type ("scatter", "line", "bar", "pie", optional): Type of plot to show. Defaults to "line".
195	title (Optional[str], optional): Title to use over the plot. Defaults to None.
196	"""
197	super().__init__(plot_type, title, legend_x, legend_y)	×
198	self.data_x = data_x	×
199	self.data_y = data_y	×
200
201	def append(self, label: str, value: float):	1✔
202	"""Add a data point to this data set.
203
204	Args:
205	label (str): label of the data point.
206	value (float): value of the data point.
207	"""
208	# self.labels.append(label)
209	# self.values.append(value)
210	# TODO
211
212
213	class NPPlotData(PlotData):	1✔
214	"""Data with Numpy array as source."""
215
216	data: np.ndarray	1✔
217
218	def __init__(	1✔
219	self,
220	data: np.ndarray,
221	plot_type: PlotType = PlotType.line,
222	title: Optional[str] = None,
223	legend_x: str = "x",
224	legend_y: str = "y",
225	):
226	"""Create a data source from a Numpy array.
227
228	Args:
229	data (:class:`np.ndarray`): Array of data points. Shape should correspond to the dimensions of the plot
230	(2 columns for 2D plots, 3 columns for 3D plots).
231	plot_type (:class:`PlotType`, optional): Type of plot to show. Defaults to `"line"`.
232	title (`str`, optional): Title to use over the plot. Defaults to `None`.
233
234	Raises:
235	`TypeError`: When trying to take from a capacity resource without specifying an amount.
236	"""
237	super().__init__(plot_type, title, legend_x, legend_y)	×
NEW 238	if data.shape[1] != 2: # TODO add 3 when adding 3D plots	×
NEW 239	raise ValueError("")	×
UNCOV 240	self.data = data	×
241
242	@property	1✔
243	def _data_frame(self):	1✔
NEW 244	return DataFrame(	×
245	{
246	self.legend_x: self.data[0],
247	self.legend_y: self.data[1],
248	}
249	)
250
251
252	class ResourcePlotData(Generic[Number], PlotData):	1✔
253	"""Data source from watching the amount of a :class:`Resource`."""
254
255	source: Resource[Number]	1✔
256	frequency: int	1✔
257
258	def __init__(	1✔
259	self,
260	source: Resource[Number],
261	frequency: int = 1,
262	plot_type: PlotType = PlotType.line,
263	title: Optional[str] = None,
264	legend_x: str = "seconds",
265	legend_y: str = "amount",
266	):
267	"""Create a data source from watching the amount of a :class:`Resource`.
268
269	Args:
270	data (:class:`Resource`): Source :class:`Resource`.
271	frequency (int, optional): Frequency in ticks to add data points.
272	Defaults to 1, meaning a point gets added every tick.
273	plot_type ("scatter", "line", "bar", "pie", optional): Type of plot to show. Defaults to "line".
274	title (Optional[str], optional): Title to use over the plot. Defaults to None.
275	"""
276	super().__init__(plot_type, title, legend_x, legend_y)	×
277	self.source = source	×
278	self.frequency = frequency	×
279
280	def _tick(self):	1✔
281	from .world import World	×
282
283	if World.ticks % self.frequency == 0:	×
NEW 284	self._x_buffer[self._buffer_index] = World.seconds()	×
NEW 285	self._y_buffer[self._buffer_index] = self.source.amount	×
NEW 286	self._buffer_index += 1	×
287
288
289	class QueuePlotData(PlotData):	1✔
290	"""Data source from watching the size of a :class:`Queue`."""
291
292	source: Queue	1✔
293	frequency: int	1✔
294
295	def __init__(	1✔
296	self,
297	source: Queue,
298	frequency: int = 1,
299	plot_type: PlotType = PlotType.line,
300	title: Optional[str] = None,
301	legend_x: str = "seconds",
302	legend_y: str = "length",
303	):
304	"""Create a data source from watching the size of a :class:`Queue`.
305
306	Args:
307	data (:class:`Queue`): Source :class:`Queue`.
308	frequency (int, optional): Frequency in ticks to add data points.
309	Defaults to 1, meaning a point gets added every tick.
310	plot_type ("scatter", "line", "bar", "pie", optional): Type of plot to show. Defaults to "line".
311	title (Optional[str], optional): Title to use over the plot. Defaults to None.
312	"""
313	super().__init__(plot_type, title, legend_x, legend_y)	×
314	self.source = source	×
315	self.frequency = frequency	×
316
317	def _tick(self):	1✔
318	from .world import World	×
319
320	if World.ticks % self.frequency == 0:	×
NEW 321	self._x_buffer[self._buffer_index] = World.seconds()	×
NEW 322	self._y_buffer[self._buffer_index] = len(self.source)	×
NEW 323	self._buffer_index += 1	×
324
325
326	class StatePlotData(PlotData):	1✔
327	"""Data source from watching the state of an :class:`Entity`."""
328
329	data: Entity	1✔
330	frequency: int	1✔
331
332	def __init__(	1✔
333	self,
334	data: Entity,
335	frequency: int = 1,
336	plot_type: PlotType = PlotType.line,
337	title: Optional[str] = None,
338	):
339	"""Create a data source from watching the state of an :class:`Entity`.
340
341	Args:
342	data (:class:`Entity`): Source :class:`Entity`.
343	frequency (int, optional): Frequency in ticks to add data points.
344	Defaults to 1, meaning a point gets added every tick.
345	plot_type ("scatter", "line", "bar", "pie", optional): Type of plot to show. Defaults to "line".
346	title (Optional[str], optional): Title to use over the plot. Defaults to None.
347	"""
348	super().__init__(plot_type, title)	×
349	self.data = data	×
350	self.frequency = frequency	×
351
352	# TODO
353
354
355	class Plot:	1✔
356	"""Base class for easily updating data for plots to be made on the dashboard."""
357
358	id: str	1✔
359	title: Optional[str]	1✔
360	figure: Figure	1✔
361	data: List[PlotData]	1✔
362
363	def __init__(self, id: str, *args: PlotData):	1✔
364	"""Create a plot to add to the dashboard using `World.add_plot()`.
365
366	Args:
367	id (str): identifier, needs to be unique.
368	*args (:class:`PlotData`): Data to start the plot with.
369	"""
370	self.id = id	×
371	self.data = []	×
372	for arg in args:	×
373	self.add_trace(arg)	×
374
375	def __getitem__(self, key: int) -> PlotData:	1✔
376	"""Get a reference to a data set from this plot.
377
378	Args:
379	key (int): Index of the data set.
380
381	Returns:
382	tuple(list[float] \| list[str], list[float]): Data set at index.
383	"""
384	return self.data[key]	×
385
386	def _tick(self):	1✔
387	for data in self.data:	×
388	data._tick()	×
389
390	def add_trace(self, data: PlotData) -> int:	1✔
391	"""Add a data trace to the plot.
392
393	Args:
394	data (PlotData): Data set.
395
396	Returns:
397	int: Index of the added data set.
398	"""
399	index: int = len(self.data)	×
400	data.plot = self	×
401	self.data.append(data)	×
402	return index	×
403
404	def _update(self):	1✔
405	for data in self.data:	×
406	data._update_traces()	×

sabvdf / datasim / 14549006187

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous