14549006187

Committed 19 Apr 2025 12:14PM UTC coverage: 48.084% (-3.9%) from 51.953%

Build # 14549006187

Build Type

push

github

Committed by

sabvdf

Commit Message

Adds local CI check; WIP Reworking update logic to fix performance issues when running with Streamlit dashboard.

Run Details

12 of 79 new or added lines in 7 files covered. (15.19%)

64 existing lines in 4 files now uncovered.

276 of 574 relevant lines covered (48.08%)

0.48 hits per line

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

41.46

/datasim/queue.py

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

from .types import Number

EntityType = TypeVar("EntityType")


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

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

    def __init__(self, id: str, capacity: int = 0):
        """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.
        """
        self.id = id
        self.capacity = capacity
        self.queue = []

    @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 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:
            self.queue.insert(0, (entity, amount))
            return True

        return False

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

        Returns:
            Entity: The entity that was at the front of this queue.
        """
        (e, a) = self.queue.pop()
        if a is None:
            return e
        return (e, a)

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

        Returns:
            Entity: The entity at the front of this queue.
        """
        (e, a) = self.queue[-1]
        if a is None:
            return e
        return (e, a)

    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)
            return True

        return False

    def __repr__(self):
        """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, Tuple, TypeVar	1✔
2
3	from .types import Number	1✔
4
5	EntityType = TypeVar("EntityType")	1✔
6
7
8	class Queue(Generic[EntityType, Number]):	1✔
9	"""A queue for entities to wait for resource availability."""
10
11	id: Final[str]	1✔
12	queue: Final[List[Tuple[EntityType, Number]]]	1✔
13	capacity: int	1✔
14
15	def __init__(self, id: str, capacity: int = 0):	1✔
16	"""Create a waiting queue for entities.
17
18	Args:
19	id (str): Identifier / name of the queue.
20	capacity (int): Maximum queue length. Defaults to 0 for no maximum.
21	"""
22	self.id = id	×
23	self.capacity = capacity	×
24	self.queue = []	×
25
26	@property	1✔
27	def full(self) -> bool:	1✔
28	"""Check the queue is full."""
29	return 0 < self.capacity <= len(self.queue)	×
30
31	def __len__(self) -> int:	1✔
32	"""Get the current length of the queue."""
33	return len(self.queue)	×
34
35	def __int__(self) -> int:	1✔
36	"""Get the current length of the queue."""
UNCOV 37	return self.__len__()	×
38
39	def enqueue(self, entity: EntityType, amount: Number = None) -> bool:	1✔
40	"""Put an entity at the end of the queue.
41
42	Args:
43	entity (T): The entity to enqueue.
44	Beware: If this entity is already in the list, it will be added another time.
45	amount (int or float, optional): The amount that the entity wants to take from the resource.
46	Defaults to None.
47
48	Returns:
49	bool:
50	If the entity was succesfully added to the queue.
51	"""
52	if not self.full:	×
UNCOV 53	self.queue.insert(0, (entity, amount))	×
UNCOV 54	return True	×
55
UNCOV 56	return False	×
57
58	def dequeue(self) -> EntityType \| Tuple[EntityType, Number]:	1✔
59	"""Remove the entity from the front of the queue and returns it.
60
61	Returns:
62	Entity: The entity that was at the front of this queue.
63	"""
UNCOV 64	(e, a) = self.queue.pop()	×
UNCOV 65	if a is None:	×
UNCOV 66	return e	×
UNCOV 67	return (e, a)	×
68
69	def peek(self) -> EntityType \| Tuple[EntityType, Number]:	1✔
70	"""Return the entity at the front of the queue without removing it.
71
72	Returns:
73	Entity: The entity at the front of this queue.
74	"""
UNCOV 75	(e, a) = self.queue[-1]	×
UNCOV 76	if a is None:	×
UNCOV 77	return e	×
UNCOV 78	return (e, a)	×
79
80	def prioritize(self, entity: EntityType) -> bool:	1✔
81	"""Pushes an entity to the front of the list.
82
83	If the entity was not in the list, it will not be added;
84	If the entity is in the list more than once, the copy furthest to the back will be put at the front.
85
86	Args:
87	entity (Entity): _description_
88	"""
89	(_, entry) = [	×
90	(i, (e, a)) for i, (e, a) in enumerate(self.queue) if e is entity
91	][0]
92	if self.queue.remove(entry):	×
UNCOV 93	self.queue.append(entry)	×
UNCOV 94	return True	×
95
96	return False	×
97
98	def __repr__(self):	1✔
99	"""Get a string representation of this queue."""
UNCOV 100	return f"Queue {self.id} length {len(self.queue)}" + (	×
101	"" if self.capacity == 0 else "/{self.capacity}"
102	)

sabvdf / datasim / 14549006187

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