14622423522

Committed 23 Apr 2025 03:46PM UTC coverage: 62.375% (-2.8%) from 65.132%

Build # 14622423522

Build Type

push

github

Committed by

sabvdf

Commit Message

Finishes function slimming; Working on test coverage.

Run Details

108 of 166 new or added lines in 10 files covered. (65.06%)

105 existing lines in 5 files now uncovered.

562 of 901 relevant lines covered (62.38%)

0.62 hits per line

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

86.67

/datasim/plot.py

from abc import ABC
from typing import Any, 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 .queue import Queue
from .resource import Resource
from .types import PlotType
import datasim.simulation as simulation


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.
        """
        self.plot_type = plot_type
        self.title = title
        self.legend_x = legend_x
        self.legend_y = legend_y

        self._buffer_size = max(20000, simulation.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,
                        title=self.title,
                        x=self.legend_x,
                        y=self.legend_y,
                    ),
                )
            case PlotType.pie:
                self.trace = cast(
                    Figure,
                    px.pie(
                        self._data_frame,
                        title=self.title,
                        names=self.legend_x,
                        values=self.legend_y,
                    ),
                )
            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(PlotData):
    """Data source from watching the amount of a :class:`Resource`."""

    source: Resource
    plot_users: bool
    frequency: int

    def __init__(
        self,
        source_id: str,
        plot_users: bool = False,
        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 = simulation.world().resource(source_id)
        self.plot_users = plot_users
        self.frequency = frequency

    def _tick(self):
        if (self.frequency == 0 and self.source.changed_tick == simulation.ticks) or (
            simulation.ticks % self.frequency == 0
        ):
            self._x_buffer[self._buffer_index] = simulation.time
            self._y_buffer[self._buffer_index] = (
                len(self.source.users) if self.plot_users else 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_id: str,
        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 = simulation.world().queue(source_id)
        self.frequency = frequency

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

sabvdf / datasim / 14622423522

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