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

75.38

/datasim/queue.py

from typing import Final, Generic, List, Literal, Optional, Tuple, TypeVar

from .entity import Entity
from .logging import log
from .types import LogLevel, PlotType, Number
from . import simulation

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


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

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

    def __init__(
        self,
        id: str,
        capacity: int = 0,
        auto_plot: PlotType | Literal[False] = PlotType.line,
        plot_frequency: int = 1,
        plot_title: Optional[str] = None,
    ):
        """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.capacity = capacity
        self.queue = []
        self.changed_tick = 0

        simulation.world().add(self)

        if auto_plot:
            self.make_plot(auto_plot, plot_frequency, plot_title)

    @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_type: PlotType = PlotType.line,
        frequency: int = 1,
        plot_title: Optional[str] = None,
    ):
        """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 Plot, QueuePlotData

        simulation.world().add_plot(
            Plot(
                self.id,
                QueuePlotData(
                    self.id,
                    frequency,
                    plot_type,
                    plot_title,
                    legend_y=EntityType.__name__.lower(),
                ),
            )
        )

    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 not self.full:
            log(
                f"{entity} joining {self} at tick {simulation.ticks}",
                LogLevel.verbose,
                45,
            )

            self.queue.insert(0, (entity, amount))
            self.changed_tick = simulation.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 = simulation.ticks

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

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