14553195335

Committed 19 Apr 2025 09:53PM UTC coverage: 65.132% (+18.7%) from 46.47%

Build # 14553195335

Build Type

push

github

Committed by

sabvdf

Commit Message

Finishes first working version of ICU example.

Run Details

11 of 36 new or added lines in 4 files covered. (30.56%)

1 existing line in 1 file now uncovered.

396 of 608 relevant lines covered (65.13%)

0.65 hits per line

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

79.22

/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,
                        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
    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.
        """
        from .world import World

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

sabvdf / datasim / 14553195335

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