15344616636

Committed 30 May 2025 10:21AM UTC coverage: 59.629% (-2.6%) from 62.179%

Build # 15344616636

Build Type

push

github

Committed by

sabvdf

Commit Message

Fixes all CI tests and adds support files for using uv instead of pip.

Run Details

16 of 25 new or added lines in 8 files covered. (64.0%)

337 existing lines in 8 files now uncovered.

836 of 1402 relevant lines covered (59.63%)

0.6 hits per line

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

70.89

/datasim/queue.py

from typing import Dict, Final, Generic, List, Tuple, TypeVar

from .entity import Entity
from .logging import log
from .types import LogLevel, Number, PlotOptions

EntityType = TypeVar("EntityType", bound=Entity)


class Queue(Generic[EntityType]):
    """A queue for entities to wait for resource availability."""

    world: Final
    id: Final[str]
    queue: Final[List[Tuple[EntityType, Number]]]
    capacity: int
    changed_tick: int

    def __init__(
        self,
        world,
        id: str,
        capacity: int = 0,
        auto_plot: bool = True,
        plot_id: str = "",
        plot_frequency: int = 1,
        plot_options: PlotOptions = PlotOptions(),
    ):
        """Create a waiting queue for entities.

        Args:
            id (str): Identifier / name of the queue.
            capacity (int): Maximum queue length. Defaults to 0 for no maximum.
            auto_plot (`PlotType` or `False`, optional): Whether to automatically add a plot to the dashboard
                for this resource, and which type of plot if so. Defaults to `PlotType.line`.
            plot_frequency (int, optional): Whether to add a data point every `frequency` ticks.
                If set to `0`, adds a data point only when the quantity changes. Defaults to `1`.
            plot_title (Optional[str], optional): An optional plot title. Defaults to `None`.
        """
        self.id = id
        self.world = world
        self.capacity = capacity
        self.queue = []
        self.changed_tick = 0

        self.world.add(self)

        if auto_plot:
            self.make_plot(plot_id, plot_frequency, plot_options)

    @staticmethod
    def _from_yaml(world, params: Dict) -> "Queue":
        id = list(params.keys())
        if len(id) > 1:
            raise ValueError(f"Unable to parse yaml: Multiple keys found in {params}")

        id = id[0]
        params = params[id]

        return Queue(
            world,
            id,
            params.get("capacity", 0),
            params.get("auto_plot", True),
            params.get("plot_id", ""),
            params.get("plot_frequency", 1),
            PlotOptions._from_yaml(params.get("plot_options", {})),
        )

    @property
    def full(self) -> bool:
        """Check the queue is full."""
        return 0 < self.capacity <= len(self.queue)

    def __len__(self) -> int:
        """Get the current length of the queue."""
        return len(self.queue)

    def __int__(self) -> int:
        """Get the current length of the queue."""
        return self.__len__()

    def make_plot(
        self,
        plot_id: str = "",
        frequency: int = 1,
        plot_options: PlotOptions = PlotOptions(),
    ):
        """Create a plot for this Queue. Also automatically used when `auto_plot` is True at creation.

        Args:
            plot_type (PlotType, optional): The type of plot to add. Defaults to PlotType.line.
            frequency (int, optional): Plot every x ticks or only on change if set to 0. Defaults to 0.
            plot_title (Optional[str], optional): Optional title for the plot. Defaults to None.
        """
        from .plot import QueuePlotData

        if plot_id == "":
            plot_id = self.id

        self.plot = QueuePlotData(self.world, self.id, frequency, plot_options)
        self.world.add_plot(plot_id, self.plot)

    def enqueue(self, entity: EntityType, amount: Number = None) -> bool:
        """Put an entity at the end of the queue.

        Args:
            entity (T): The entity to enqueue.
                Beware: If this entity is already in the list, it will be added another time.
            amount (int or float, optional): The amount that the entity wants to take from the resource.
                Defaults to None.

        Returns:
            bool:
                If the entity was succesfully added to the queue.
        """
        if hasattr(self, "plot") and self.plot.options.legend_y == "":
            self.plot.options.legend_y = str(entity.plural).lower()

        if not self.full:
            log(
                f"{entity} joining {self}",
                LogLevel.verbose,
                45,
                world=self.world,
            )

            self.queue.insert(0, (entity, amount))
            self.changed_tick = self.world.ticks

            return True

        return False

    def dequeue(self) -> EntityType | Tuple[EntityType, Number] | None:
        """Remove the entity from the front of the queue and returns it.

        Returns:
            Entity: The entity that was at the front of this queue.
        """
        if len(self.queue) == 0:
            return None

        (e, a) = self.queue.pop()
        self.changed_tick = self.world.ticks

        log(
            f"{e} left {self}",
            LogLevel.verbose,
            45,
        )

        if a is None:
            return e
        return (e, a)

    def peek(self) -> EntityType | None:
        """Return the entity at the front of the queue without removing it.

        Returns:
            Entity: The entity at the front of this queue.
        """
        if len(self.queue) == 0:
            return None

        (e, a) = self.queue[-1]
        return e

    def peek_with_amount(self) -> Tuple[EntityType, Number] | None:
        """Return the entity at the front of the queue without removing it.

        Returns:
            Entity: The entity at the front of this queue.
        """
        if len(self.queue) == 0:
            return None

        return self.queue[-1]

    def prioritize(self, entity: EntityType) -> bool:
        """Pushes an entity to the front of the list.

        If the entity was not in the list, it will not be added;
        If the entity is in the list more than once, the copy furthest to the back will be put at the front.

        Args:
            entity (Entity): _description_
        """
        (_, entry) = [
            (i, (e, a)) for i, (e, a) in enumerate(self.queue) if e is entity
        ][0]
        if self.queue.remove(entry):
            self.queue.append(entry)
            self.changed_tick = self.world.ticks
            return True

        return False

    def __str__(self) -> str:
        """Get a string representation of this queue."""
        return self.__repr__()

    def __repr__(self) -> str:
        """Get a string representation of this queue."""
        return f"Queue {self.id} length {len(self.queue)}" + (
            "" if self.capacity == 0 else "/{self.capacity}"
        )

1	from typing import Dict, Final, Generic, List, Tuple, TypeVar	1✔
2
3	from .entity import Entity	1✔
4	from .logging import log	1✔
5	from .types import LogLevel, Number, PlotOptions	1✔
6
7	EntityType = TypeVar("EntityType", bound=Entity)	1✔
8
9
10	class Queue(Generic[EntityType]):	1✔
11	"""A queue for entities to wait for resource availability."""
12
13	world: Final	1✔
14	id: Final[str]	1✔
15	queue: Final[List[Tuple[EntityType, Number]]]	1✔
16	capacity: int	1✔
17	changed_tick: int	1✔
18
19	def __init__(	1✔
20	self,
21	world,
22	id: str,
23	capacity: int = 0,
24	auto_plot: bool = True,
25	plot_id: str = "",
26	plot_frequency: int = 1,
27	plot_options: PlotOptions = PlotOptions(),
28	):
29	"""Create a waiting queue for entities.
30
31	Args:
32	id (str): Identifier / name of the queue.
33	capacity (int): Maximum queue length. Defaults to 0 for no maximum.
34	auto_plot (`PlotType` or `False`, optional): Whether to automatically add a plot to the dashboard
35	for this resource, and which type of plot if so. Defaults to `PlotType.line`.
36	plot_frequency (int, optional): Whether to add a data point every `frequency` ticks.
37	If set to `0`, adds a data point only when the quantity changes. Defaults to `1`.
38	plot_title (Optional[str], optional): An optional plot title. Defaults to `None`.
39	"""
40	self.id = id	1✔
41	self.world = world	1✔
42	self.capacity = capacity	1✔
43	self.queue = []	1✔
44	self.changed_tick = 0	1✔
45
46	self.world.add(self)	1✔
47
48	if auto_plot:	1✔
49	self.make_plot(plot_id, plot_frequency, plot_options)	1✔
50
51	@staticmethod	1✔
52	def _from_yaml(world, params: Dict) -> "Queue":	1✔
UNCOV 53	id = list(params.keys())	×
UNCOV 54	if len(id) > 1:	×
UNCOV 55	raise ValueError(f"Unable to parse yaml: Multiple keys found in {params}")	×
56
UNCOV 57	id = id[0]	×
UNCOV 58	params = params[id]	×
59
UNCOV 60	return Queue(	×
61	world,
62	id,
63	params.get("capacity", 0),
64	params.get("auto_plot", True),
65	params.get("plot_id", ""),
66	params.get("plot_frequency", 1),
67	PlotOptions._from_yaml(params.get("plot_options", {})),
68	)
69
70	@property	1✔
71	def full(self) -> bool:	1✔
72	"""Check the queue is full."""
73	return 0 < self.capacity <= len(self.queue)	1✔
74
75	def __len__(self) -> int:	1✔
76	"""Get the current length of the queue."""
77	return len(self.queue)	1✔
78
79	def __int__(self) -> int:	1✔
80	"""Get the current length of the queue."""
UNCOV 81	return self.__len__()	×
82
83	def make_plot(	1✔
84	self,
85	plot_id: str = "",
86	frequency: int = 1,
87	plot_options: PlotOptions = PlotOptions(),
88	):
89	"""Create a plot for this Queue. Also automatically used when `auto_plot` is True at creation.
90
91	Args:
92	plot_type (PlotType, optional): The type of plot to add. Defaults to PlotType.line.
93	frequency (int, optional): Plot every x ticks or only on change if set to 0. Defaults to 0.
94	plot_title (Optional[str], optional): Optional title for the plot. Defaults to None.
95	"""
96	from .plot import QueuePlotData	1✔
97
98	if plot_id == "":	1✔
99	plot_id = self.id	1✔
100
101	self.plot = QueuePlotData(self.world, self.id, frequency, plot_options)	1✔
102	self.world.add_plot(plot_id, self.plot)	1✔
103
104	def enqueue(self, entity: EntityType, amount: Number = None) -> bool:	1✔
105	"""Put an entity at the end of the queue.
106
107	Args:
108	entity (T): The entity to enqueue.
109	Beware: If this entity is already in the list, it will be added another time.
110	amount (int or float, optional): The amount that the entity wants to take from the resource.
111	Defaults to None.
112
113	Returns:
114	bool:
115	If the entity was succesfully added to the queue.
116	"""
117	if hasattr(self, "plot") and self.plot.options.legend_y == "":	1✔
118	self.plot.options.legend_y = str(entity.plural).lower()	×
119
120	if not self.full:	1✔
121	log(	1✔
122	f"{entity} joining {self}",
123	LogLevel.verbose,
124	45,
125	world=self.world,
126	)
127
128	self.queue.insert(0, (entity, amount))	1✔
129	self.changed_tick = self.world.ticks	1✔
130
131	return True	1✔
132
133	return False	1✔
134
135	def dequeue(self) -> EntityType \| Tuple[EntityType, Number] \| None:	1✔
136	"""Remove the entity from the front of the queue and returns it.
137
138	Returns:
139	Entity: The entity that was at the front of this queue.
140	"""
141	if len(self.queue) == 0:	1✔
142	return None	×
143
144	(e, a) = self.queue.pop()	1✔
145	self.changed_tick = self.world.ticks	1✔
146
147	log(	1✔
148	f"{e} left {self}",
149	LogLevel.verbose,
150	45,
151	)
152
153	if a is None:	1✔
154	return e	×
155	return (e, a)	1✔
156
157	def peek(self) -> EntityType \| None:	1✔
158	"""Return the entity at the front of the queue without removing it.
159
160	Returns:
161	Entity: The entity at the front of this queue.
162	"""
UNCOV 163	if len(self.queue) == 0:	×
UNCOV 164	return None	×
165
UNCOV 166	(e, a) = self.queue[-1]	×
UNCOV 167	return e	×
168
169	def peek_with_amount(self) -> Tuple[EntityType, Number] \| None:	1✔
170	"""Return the entity at the front of the queue without removing it.
171
172	Returns:
173	Entity: The entity at the front of this queue.
174	"""
UNCOV 175	if len(self.queue) == 0:	×
UNCOV 176	return None	×
177
UNCOV 178	return self.queue[-1]	×
179
180	def prioritize(self, entity: EntityType) -> bool:	1✔
181	"""Pushes an entity to the front of the list.
182
183	If the entity was not in the list, it will not be added;
184	If the entity is in the list more than once, the copy furthest to the back will be put at the front.
185
186	Args:
187	entity (Entity): _description_
188	"""
UNCOV 189	(_, entry) = [	×
190	(i, (e, a)) for i, (e, a) in enumerate(self.queue) if e is entity
191	][0]
UNCOV 192	if self.queue.remove(entry):	×
UNCOV 193	self.queue.append(entry)	×
UNCOV 194	self.changed_tick = self.world.ticks	×
UNCOV 195	return True	×
196
UNCOV 197	return False	×
198
199	def __str__(self) -> str:	1✔
200	"""Get a string representation of this queue."""
201	return self.__repr__()	1✔
202
203	def __repr__(self) -> str:	1✔
204	"""Get a string representation of this queue."""
205	return f"Queue {self.id} length {len(self.queue)}" + (	1✔
206	"" if self.capacity == 0 else "/{self.capacity}"
207	)

sabvdf / datasim / 15344616636

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