satella.coding.structures.heaps package¶
Submodules¶
satella.coding.structures.heaps.base module¶
- class satella.coding.structures.heaps.base.Heap(from_list=None)¶
Bases:
UserList
,Generic
[T
]Sane heap as object - not like heapq.
Goes from lowest-to-highest (first popped is smallest). Standard Python comparision rules apply.
Not thread-safe
- Parameters:
from_list (Optional[Iterable[T]]) –
- filter_map(filter_fun=None, map_fun=None)¶
- Get only items that return True when condition(item) is True. Apply a
transform: item’ = item(condition) on
the rest. Maintain heap invariant.
- Parameters:
filter_fun (Optional[Callable[[T], bool]]) –
map_fun (Optional[Callable[[T], Any]]) –
- iter_ascending()¶
Return an iterator returning all elements in this heap sorted ascending. State of the heap is not changed
- Return type:
Iterable[T]
- iter_descending()¶
Return an iterator returning all elements in this heap sorted descending. State of the heap is not changed.
This loads all elements of the heap into memory at once, so be careful.
- Return type:
Iterable[T]
- pop()¶
Return smallest element of the heap.
- Raises:
IndexError – on empty heap
- Return type:
T
- pop_item(item)¶
Pop an item off the heap, maintaining the heap invariant
- Raises:
ValueError – element not found
- Parameters:
item (T) –
- Return type:
T
- push(item, *args)¶
Use it like:
>>> heap.push(3)
or:
>>> heap.push(4, myobject)
However this syntax is
Deprecated since version 2.25.5: It’s not legible
- Parameters:
item (T) –
- Return type:
None
- push_many(items)¶
Put multiple items on the heap. Faster than
>>> for item in items: >>> heap.push(item)
- Param:
an iterable with items to put
- Parameters:
items (Iterable[T]) –
- Return type:
None
- class satella.coding.structures.heaps.base.SetHeap(from_list=None)¶
Bases:
Heap
A heap with additional invariant that no two elements are the same.
Note that elements you insert in this must be eq-able and hashable, ie. you can put them in a dict.
Optimized for fast insertions and fast __contains__
#notthreadsafe
- Parameters:
from_list (Optional[Iterable[T]]) –
- filter_map(filter_fun=None, map_fun=None)¶
- Get only items that return True when condition(item) is True. Apply a
transform: item’ = item(condition) on
the rest. Maintain heap invariant.
- Parameters:
filter_fun (Optional[Callable[[T], bool]]) –
map_fun (Optional[Callable[[T], Any]]) –
- pop()¶
Return smallest element of the heap.
- Raises:
IndexError – on empty heap
- Return type:
T
- push(item)¶
Use it like:
>>> heap.push(3)
or:
>>> heap.push(4, myobject)
However this syntax is
Deprecated since version 2.25.5: It’s not legible
- Parameters:
item (T) –
- push_many(items)¶
Not that much faster than inserting anything by hand
- Parameters:
items (Iterable[T]) –
- Return type:
None
satella.coding.structures.heaps.time module¶
- class satella.coding.structures.heaps.time.TimeBasedHeap(default_clock_source=<built-in function monotonic>)¶
Bases:
Heap
A heap of items sorted by timestamps.
It is easy to ask for items, whose timestamps are LOWER than a value, and easy to remove them.
Can be used to implement a scheduling service, ie. store jobs, and each interval query which of them should be executed. This loses time resolution, but is fast.
Can use current time with put/pop_less_than. Use default_clock_source to pass a callable:
time.time
time.monotonic
Default is time.monotonic
#notthreadsafe
Initialize an empty heap
- Parameters:
default_clock_source (Callable[[], Union[int, float]]) –
- get_timestamp(item)¶
Return the timestamp for given item
- Parameters:
item (T) –
- Return type:
Union[int, float]
- items()¶
Return an iterable, but WITHOUT timestamps (only items), in unspecified order
- Return type:
Iterable[Tuple[Union[int, float], T]]
- peek_closest()¶
Return the closest object to the execution deadline, but not discard it from the heap.
- Raises:
IndexError – the heap is empty
- Return type:
Tuple[Union[int, float], T]
- pop_item(item)¶
Pop an item off the heap, maintaining the heap invariant.
The item will be a second part of the tuple
- Raises:
ValueError – element not found
- Parameters:
item (T) –
- Return type:
Tuple[Union[int, float], T]
- pop_less_than(less=None)¶
Return all elements less (sharp inequality) than particular value.
This changes state of the heap
- Parameters:
less (Optional[Union[int, float]]) – value to compare against. If left at default, it will be the default clock source specified at construction.
- Returns:
an Iterator
- Return type:
Iterator[Tuple[Union[int, float], T]]
- pop_timestamp(timestamp)¶
Get first item with given timestamp, while maintaining the heap invariant
- Raises:
ValueError – element not found
- Parameters:
timestamp (Union[int, float]) –
- Return type:
T
- put(timestamp_or_value, value=None)¶
Put an item on heap.
Pass timestamp, item or just an item for default time
- Parameters:
timestamp_or_value (Union[T, int, float]) –
value (Optional[T]) –
- Return type:
None
- remove(item)¶
Remove all things equal to item
- Parameters:
item (T) –
- Return type:
None
- class satella.coding.structures.heaps.time.TimeBasedSetHeap(default_clock_source=None)¶
Bases:
Heap
A heap of items sorted by timestamps, with such invariant that every item can appear at most once.
Note that elements you insert in this must be eq-able and hashable, ie. you can put them in a dict.
It is easy to ask for items, whose timestamps are LOWER than a value, and easy to remove them.
Can be used to implement a scheduling service, ie. store jobs, and each interval query which of them should be executed. This loses time resolution, but is fast.
Can use current time with put/pop_less_than. Use default_clock_source to pass a callable:
time.time
time.monotonic
Default is time.monotonic
#notthreadsafe
Initialize an empty heap
- Parameters:
default_clock_source (Callable[[], Union[int, float]]) –
- get_timestamp(item)¶
Return the timestamp for given item
- Raises:
ValueError – item not found
- Parameters:
item (T) –
- Return type:
Union[int, float]
- items()¶
Return an iterator, but WITHOUT timestamps (only items), in unspecified order
- Return type:
Iterator[T]
- pop()¶
Return smallest element of the heap.
- Raises:
IndexError – on empty heap
- Return type:
Tuple[Union[int, float], T]
- pop_item(item)¶
Pop an item off the heap, maintaining the heap invariant.
The item will be a second part of the tuple
- Raises:
ValueError – element not found
- Parameters:
item (T) –
- Return type:
Tuple[Union[int, float], T]
- pop_less_than(less=None)¶
Return all elements less (sharp inequality) than particular value.
This changes state of the heap
- Parameters:
less (Optional[Union[int, float]]) – value to compare against
- Returns:
a Generator
- Return type:
Iterator[Tuple[Union[int, float], T]]
- pop_timestamp(timestamp)¶
Pop an arbitary object (in case there’s two) item with given timestamp, while maintaining the heap invariant
- Raises:
ValueError – element not found
- Parameters:
timestamp (Union[int, float]) –
- Return type:
T
- push(item)¶
Use it like:
>>> heap.push(3)
or:
>>> heap.push(4, myobject)
However this syntax is
Deprecated since version 2.25.5: It’s not legible
- Parameters:
item (Tuple[Union[int, float], T]) –
- Return type:
None
- put(timestamp_or_value, value=None)¶
Put an item on heap.
Pass timestamp, item or just an item for default time
- Parameters:
timestamp_or_value (Union[T, int, float]) –
value (Optional[T]) –
- Return type:
None
- remove(item)¶
Remove all things equal to item
- Parameters:
item (T) –
- Return type:
None
Module contents¶
- class satella.coding.structures.heaps.Heap(from_list=None)¶
Bases:
UserList
,Generic
[T
]Sane heap as object - not like heapq.
Goes from lowest-to-highest (first popped is smallest). Standard Python comparision rules apply.
Not thread-safe
- Parameters:
from_list (Optional[Iterable[T]]) –
- filter_map(filter_fun=None, map_fun=None)¶
- Get only items that return True when condition(item) is True. Apply a
transform: item’ = item(condition) on
the rest. Maintain heap invariant.
- Parameters:
filter_fun (Optional[Callable[[T], bool]]) –
map_fun (Optional[Callable[[T], Any]]) –
- iter_ascending()¶
Return an iterator returning all elements in this heap sorted ascending. State of the heap is not changed
- Return type:
Iterable[T]
- iter_descending()¶
Return an iterator returning all elements in this heap sorted descending. State of the heap is not changed.
This loads all elements of the heap into memory at once, so be careful.
- Return type:
Iterable[T]
- pop()¶
Return smallest element of the heap.
- Raises:
IndexError – on empty heap
- Return type:
T
- pop_item(item)¶
Pop an item off the heap, maintaining the heap invariant
- Raises:
ValueError – element not found
- Parameters:
item (T) –
- Return type:
T
- push(item, *args)¶
Use it like:
>>> heap.push(3)
or:
>>> heap.push(4, myobject)
However this syntax is
Deprecated since version 2.25.5: It’s not legible
- Parameters:
item (T) –
- Return type:
None
- push_many(items)¶
Put multiple items on the heap. Faster than
>>> for item in items: >>> heap.push(item)
- Param:
an iterable with items to put
- Parameters:
items (Iterable[T]) –
- Return type:
None
- class satella.coding.structures.heaps.SetHeap(from_list=None)¶
Bases:
Heap
A heap with additional invariant that no two elements are the same.
Note that elements you insert in this must be eq-able and hashable, ie. you can put them in a dict.
Optimized for fast insertions and fast __contains__
#notthreadsafe
- Parameters:
from_list (Optional[Iterable[T]]) –
- filter_map(filter_fun=None, map_fun=None)¶
- Get only items that return True when condition(item) is True. Apply a
transform: item’ = item(condition) on
the rest. Maintain heap invariant.
- Parameters:
filter_fun (Optional[Callable[[T], bool]]) –
map_fun (Optional[Callable[[T], Any]]) –
- pop()¶
Return smallest element of the heap.
- Raises:
IndexError – on empty heap
- Return type:
T
- push(item)¶
Use it like:
>>> heap.push(3)
or:
>>> heap.push(4, myobject)
However this syntax is
Deprecated since version 2.25.5: It’s not legible
- Parameters:
item (T) –
- push_many(items)¶
Not that much faster than inserting anything by hand
- Parameters:
items (Iterable[T]) –
- Return type:
None
- class satella.coding.structures.heaps.TimeBasedHeap(default_clock_source=<built-in function monotonic>)¶
Bases:
Heap
A heap of items sorted by timestamps.
It is easy to ask for items, whose timestamps are LOWER than a value, and easy to remove them.
Can be used to implement a scheduling service, ie. store jobs, and each interval query which of them should be executed. This loses time resolution, but is fast.
Can use current time with put/pop_less_than. Use default_clock_source to pass a callable:
time.time
time.monotonic
Default is time.monotonic
#notthreadsafe
Initialize an empty heap
- Parameters:
default_clock_source (Callable[[], Union[int, float]]) –
- get_timestamp(item)¶
Return the timestamp for given item
- Parameters:
item (T) –
- Return type:
Union[int, float]
- items()¶
Return an iterable, but WITHOUT timestamps (only items), in unspecified order
- Return type:
Iterable[Tuple[Union[int, float], T]]
- peek_closest()¶
Return the closest object to the execution deadline, but not discard it from the heap.
- Raises:
IndexError – the heap is empty
- Return type:
Tuple[Union[int, float], T]
- pop_item(item)¶
Pop an item off the heap, maintaining the heap invariant.
The item will be a second part of the tuple
- Raises:
ValueError – element not found
- Parameters:
item (T) –
- Return type:
Tuple[Union[int, float], T]
- pop_less_than(less=None)¶
Return all elements less (sharp inequality) than particular value.
This changes state of the heap
- Parameters:
less (Optional[Union[int, float]]) – value to compare against. If left at default, it will be the default clock source specified at construction.
- Returns:
an Iterator
- Return type:
Iterator[Tuple[Union[int, float], T]]
- pop_timestamp(timestamp)¶
Get first item with given timestamp, while maintaining the heap invariant
- Raises:
ValueError – element not found
- Parameters:
timestamp (Union[int, float]) –
- Return type:
T
- put(timestamp_or_value, value=None)¶
Put an item on heap.
Pass timestamp, item or just an item for default time
- Parameters:
timestamp_or_value (Union[T, int, float]) –
value (Optional[T]) –
- Return type:
None
- remove(item)¶
Remove all things equal to item
- Parameters:
item (T) –
- Return type:
None
- class satella.coding.structures.heaps.TimeBasedSetHeap(default_clock_source=None)¶
Bases:
Heap
A heap of items sorted by timestamps, with such invariant that every item can appear at most once.
Note that elements you insert in this must be eq-able and hashable, ie. you can put them in a dict.
It is easy to ask for items, whose timestamps are LOWER than a value, and easy to remove them.
Can be used to implement a scheduling service, ie. store jobs, and each interval query which of them should be executed. This loses time resolution, but is fast.
Can use current time with put/pop_less_than. Use default_clock_source to pass a callable:
time.time
time.monotonic
Default is time.monotonic
#notthreadsafe
Initialize an empty heap
- Parameters:
default_clock_source (Callable[[], Union[int, float]]) –
- get_timestamp(item)¶
Return the timestamp for given item
- Raises:
ValueError – item not found
- Parameters:
item (T) –
- Return type:
Union[int, float]
- items()¶
Return an iterator, but WITHOUT timestamps (only items), in unspecified order
- Return type:
Iterator[T]
- pop()¶
Return smallest element of the heap.
- Raises:
IndexError – on empty heap
- Return type:
Tuple[Union[int, float], T]
- pop_item(item)¶
Pop an item off the heap, maintaining the heap invariant.
The item will be a second part of the tuple
- Raises:
ValueError – element not found
- Parameters:
item (T) –
- Return type:
Tuple[Union[int, float], T]
- pop_less_than(less=None)¶
Return all elements less (sharp inequality) than particular value.
This changes state of the heap
- Parameters:
less (Optional[Union[int, float]]) – value to compare against
- Returns:
a Generator
- Return type:
Iterator[Tuple[Union[int, float], T]]
- pop_timestamp(timestamp)¶
Pop an arbitary object (in case there’s two) item with given timestamp, while maintaining the heap invariant
- Raises:
ValueError – element not found
- Parameters:
timestamp (Union[int, float]) –
- Return type:
T
- push(item)¶
Use it like:
>>> heap.push(3)
or:
>>> heap.push(4, myobject)
However this syntax is
Deprecated since version 2.25.5: It’s not legible
- Parameters:
item (Tuple[Union[int, float], T]) –
- Return type:
None
- put(timestamp_or_value, value=None)¶
Put an item on heap.
Pass timestamp, item or just an item for default time
- Parameters:
timestamp_or_value (Union[T, int, float]) –
value (Optional[T]) –
- Return type:
None
- remove(item)¶
Remove all things equal to item
- Parameters:
item (T) –
- Return type:
None