14550200639

Committed 19 Apr 2025 02:50PM UTC coverage: 46.47% (-1.6%) from 48.084%

Build # 14550200639

Build Type

push

github

Committed by

sabvdf

Commit Message

Fixes CI and typing; Adds functional resource usage, except for active user index deletion.

Run Details

57 of 114 new or added lines in 8 files covered. (50.0%)

1 existing line in 1 file now uncovered.

283 of 609 relevant lines covered (46.47%)

0.46 hits per line

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

44.74

/datasim/plot.py

from abc import ABC
from enum import Enum
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


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(PlotData):
    """Data source from watching the amount of a :class:`Resource`."""

    source: Resource
    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 = "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.
        """
        from .world import World

        super().__init__(plot_type, title, legend_x, legend_y)
        self.source = World.current.resource(source_id)
        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_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.
        """
        from .world import World

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

sabvdf / datasim / 14550200639

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