Compare commits
No commits in common. "dev" and "main" have entirely different histories.
@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "dopt-basics"
|
||||
version = "0.2.5dev0"
|
||||
version = "0.2.4"
|
||||
description = "basic cross-project tools for Python-based d-opt projects"
|
||||
authors = [
|
||||
{name = "Florian Förster", email = "f.foerster@d-opt.com"},
|
||||
@ -69,7 +69,7 @@ directory = "reports/coverage"
|
||||
|
||||
|
||||
[tool.bumpversion]
|
||||
current_version = "0.2.5dev0"
|
||||
current_version = "0.2.4"
|
||||
parse = """(?x)
|
||||
(?P<major>0|[1-9]\\d*)\\.
|
||||
(?P<minor>0|[1-9]\\d*)\\.
|
||||
|
||||
@ -1,127 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Collection
|
||||
from typing import Generic, Never, TypeVar
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
class Chunk(Generic[T]):
|
||||
def __init__(
|
||||
self,
|
||||
capacity: int,
|
||||
overfill_once: bool = True,
|
||||
cap_property: str | None = None,
|
||||
) -> None:
|
||||
self.capacity = capacity
|
||||
self.overfill = overfill_once
|
||||
self.capacity_left: int = capacity
|
||||
self.capacity_held: int = 0
|
||||
self.full: bool = False
|
||||
self._contents: list[T] = []
|
||||
self.cap_property = cap_property
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return self._contents.__repr__()
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self._contents.__str__()
|
||||
|
||||
@property
|
||||
def contents(self) -> list[T]:
|
||||
return self._contents
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._contents)
|
||||
|
||||
def __getitem__(
|
||||
self,
|
||||
idx: int,
|
||||
) -> T:
|
||||
return self._contents[idx]
|
||||
|
||||
def __raise_for_exceeded_capacity(self) -> Never:
|
||||
raise ValueError("Item can not be added due to capacity limit.")
|
||||
|
||||
def _capacity_sufficient(
|
||||
self,
|
||||
size_to_add: int,
|
||||
) -> bool:
|
||||
if self.full:
|
||||
return False
|
||||
|
||||
new_size = len(self) + size_to_add
|
||||
|
||||
if self.overfill and (new_size > self.capacity):
|
||||
return True
|
||||
elif new_size > self.capacity:
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
def _recalc_capacity(
|
||||
self,
|
||||
size_to_add: int,
|
||||
) -> None:
|
||||
self.capacity_held += size_to_add
|
||||
self.capacity_left = max(self.capacity_left - size_to_add, 0)
|
||||
if self.capacity_left == 0:
|
||||
self.full = True
|
||||
else:
|
||||
self.full = False
|
||||
|
||||
def append(
|
||||
self,
|
||||
object_: T,
|
||||
) -> None:
|
||||
size_to_add: int = 1
|
||||
if self.cap_property is not None:
|
||||
if not hasattr(object_, self.cap_property):
|
||||
raise AttributeError("Object does not posses the wanted property")
|
||||
attr = getattr(object_, self.cap_property)
|
||||
if not isinstance(attr, int):
|
||||
raise TypeError("Capacity property must be an integer")
|
||||
size_to_add = attr
|
||||
|
||||
if self._capacity_sufficient(size_to_add):
|
||||
self._contents.append(object_)
|
||||
self._recalc_capacity(size_to_add)
|
||||
else:
|
||||
self.__raise_for_exceeded_capacity()
|
||||
|
||||
def extend(
|
||||
self,
|
||||
collection: Collection[T],
|
||||
) -> None:
|
||||
size_to_add: int = len(collection)
|
||||
if self.cap_property is not None:
|
||||
if not all(hasattr(object_, self.cap_property) for object_ in collection):
|
||||
raise AttributeError("Not all objects posses the wanted property")
|
||||
if not all(
|
||||
isinstance(getattr(object_, self.cap_property), int) for object_ in collection
|
||||
):
|
||||
raise TypeError("Capacity property must be an integer")
|
||||
size_to_add = sum(getattr(object_, self.cap_property) for object_ in collection)
|
||||
|
||||
if self._capacity_sufficient(size_to_add):
|
||||
self._contents.extend(collection)
|
||||
self._recalc_capacity(size_to_add)
|
||||
else:
|
||||
self.__raise_for_exceeded_capacity()
|
||||
|
||||
def remove(
|
||||
self,
|
||||
object_: T,
|
||||
) -> None:
|
||||
size_to_add: int = 1
|
||||
if self.cap_property is not None:
|
||||
if not hasattr(object_, self.cap_property):
|
||||
raise AttributeError("Object does not posses the wanted property")
|
||||
attr = getattr(object_, self.cap_property)
|
||||
if not isinstance(attr, int):
|
||||
raise TypeError("Capacity property must be an integer")
|
||||
size_to_add = attr
|
||||
size_to_remove = (-1) * size_to_add
|
||||
|
||||
self._contents.remove(object_)
|
||||
self._recalc_capacity(size_to_remove)
|
||||
@ -1,19 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import deque
|
||||
from collections.abc import Iterator, Sequence
|
||||
from typing import TypeVar
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
def symmetric_iter(
|
||||
x: Sequence[T],
|
||||
) -> Iterator[T]:
|
||||
d = deque(x)
|
||||
|
||||
for idx in range(len(d)):
|
||||
if idx % 2 == 0:
|
||||
yield d.popleft()
|
||||
else:
|
||||
yield d.pop()
|
||||
@ -1,38 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from dataclasses import fields, is_dataclass
|
||||
from typing import Any, Literal
|
||||
|
||||
|
||||
def obj_size(
|
||||
obj: Any,
|
||||
seen: set[int] | None = None,
|
||||
unit: Literal["b", "kb", "mb", "gb"] = "b",
|
||||
) -> float:
|
||||
size = sys.getsizeof(obj)
|
||||
if seen is None:
|
||||
seen = set()
|
||||
|
||||
obj_id = id(obj)
|
||||
if obj_id in seen:
|
||||
return 0
|
||||
seen.add(obj_id)
|
||||
|
||||
if is_dataclass(obj):
|
||||
for f in fields(obj):
|
||||
size += obj_size(getattr(obj, f.name), seen)
|
||||
elif isinstance(obj, dict):
|
||||
size += sum(obj_size(v, seen) + obj_size(k, seen) for k, v in obj.items())
|
||||
elif isinstance(obj, (list, tuple, set)):
|
||||
size += sum(obj_size(i, seen) for i in obj)
|
||||
|
||||
match unit:
|
||||
case "kb":
|
||||
size /= 1024
|
||||
case "mb":
|
||||
size /= 1024 * 1024
|
||||
case "gb":
|
||||
size /= 1024 * 1024 * 1024
|
||||
|
||||
return size
|
||||
@ -1,142 +0,0 @@
|
||||
import pytest
|
||||
|
||||
from dopt_basics.batching import Chunk
|
||||
|
||||
|
||||
class _Item:
|
||||
def __init__(
|
||||
self,
|
||||
size: int,
|
||||
) -> None:
|
||||
self.size = size
|
||||
|
||||
|
||||
def test_Chunk_base():
|
||||
chunk: Chunk[int] = Chunk(3)
|
||||
|
||||
assert chunk.capacity == 3
|
||||
assert chunk.capacity_held == 0
|
||||
assert chunk.capacity_left == 3
|
||||
assert chunk.full is False
|
||||
assert len(chunk) == 0
|
||||
assert not chunk.contents
|
||||
|
||||
|
||||
# TODO implement behaviour
|
||||
# def test_Chunk_base_overfill_append():
|
||||
# chunk: Chunk[int] = Chunk(3)
|
||||
|
||||
# chunk.append(1)
|
||||
# chunk.append(2)
|
||||
# chunk.append(3)
|
||||
# chunk.append(4) # allow once, then not again
|
||||
|
||||
|
||||
def test_Chunk_base_no_overfill_append():
|
||||
chunk: Chunk[int] = Chunk(3, overfill_once=False)
|
||||
|
||||
chunk.append(1)
|
||||
assert chunk[0] == 1
|
||||
chunk.append(2)
|
||||
chunk.append(3)
|
||||
with pytest.raises(ValueError, match="Item can not be added due to capacity limit"):
|
||||
chunk.append(4)
|
||||
|
||||
|
||||
def test_Chunk_base_overfill_extend():
|
||||
chunk: Chunk[int] = Chunk(3)
|
||||
|
||||
chunk.extend((1, 2))
|
||||
chunk.extend((3, 4))
|
||||
|
||||
with pytest.raises(ValueError, match="Item can not be added due to capacity limit"):
|
||||
chunk.extend((5, 6))
|
||||
|
||||
|
||||
def test_Chunk_base_no_overfill_extend():
|
||||
chunk: Chunk[int] = Chunk(3, overfill_once=False)
|
||||
|
||||
chunk.extend((1, 2))
|
||||
|
||||
with pytest.raises(ValueError, match="Item can not be added due to capacity limit"):
|
||||
chunk.extend((3, 4))
|
||||
|
||||
|
||||
def test_Chunk_base_remove():
|
||||
chunk: Chunk[int] = Chunk(3, overfill_once=False)
|
||||
|
||||
chunk.extend((1, 2))
|
||||
assert chunk.capacity_held == 2
|
||||
assert chunk.capacity_left == 1
|
||||
|
||||
chunk.remove(2)
|
||||
assert chunk.capacity_held == 1
|
||||
assert chunk.capacity_left == 2
|
||||
|
||||
|
||||
def test_Chunk_property_append_success():
|
||||
CAP_PROP = "size"
|
||||
chunk: Chunk[_Item] = Chunk(3, overfill_once=False, cap_property=CAP_PROP)
|
||||
i1 = _Item(2)
|
||||
i2 = _Item(4)
|
||||
|
||||
chunk.append(i1)
|
||||
|
||||
with pytest.raises(ValueError, match="Item can not be added due to capacity limit"):
|
||||
chunk.append(i2)
|
||||
|
||||
|
||||
def test_Chunk_property_append_fail_no_attribute():
|
||||
CAP_PROP = "size"
|
||||
chunk: Chunk = Chunk(3, overfill_once=False, cap_property=CAP_PROP)
|
||||
|
||||
with pytest.raises(AttributeError, match="Object does not posses the wanted property"):
|
||||
chunk.append(1)
|
||||
|
||||
|
||||
def test_Chunk_property_append_fail_wrong_type():
|
||||
CAP_PROP = "size"
|
||||
chunk: Chunk = Chunk(3, overfill_once=False, cap_property=CAP_PROP)
|
||||
i1 = _Item(4)
|
||||
i1.size = "wrong type" # type: ignore
|
||||
|
||||
with pytest.raises(TypeError, match="Capacity property must be an integer"):
|
||||
chunk.append(i1)
|
||||
|
||||
|
||||
def test_Chunk_property_extend_fail_no_attribute():
|
||||
CAP_PROP = "size"
|
||||
chunk: Chunk = Chunk(3, overfill_once=False, cap_property=CAP_PROP)
|
||||
i1 = _Item(2)
|
||||
|
||||
with pytest.raises(AttributeError, match="Not all objects posses the wanted property"):
|
||||
chunk.extend((i1, 2))
|
||||
|
||||
|
||||
def test_Chunk_property_extend_fail_wrong_type():
|
||||
CAP_PROP = "size"
|
||||
chunk: Chunk = Chunk(3, overfill_once=False, cap_property=CAP_PROP)
|
||||
i1 = _Item(2)
|
||||
i2 = _Item(4)
|
||||
i2.size = "wrong type" # type: ignore
|
||||
|
||||
with pytest.raises(TypeError, match="Capacity property must be an integer"):
|
||||
chunk.extend((i1, i2))
|
||||
|
||||
|
||||
def test_Chunk_property_remove_fail_no_attribute():
|
||||
CAP_PROP = "size"
|
||||
chunk: Chunk = Chunk(3, overfill_once=False, cap_property=CAP_PROP)
|
||||
|
||||
with pytest.raises(AttributeError, match="Object does not posses the wanted property"):
|
||||
chunk.remove(1)
|
||||
|
||||
|
||||
def test_Chunk_property_remove_fail_wrong_type():
|
||||
CAP_PROP = "size"
|
||||
chunk: Chunk = Chunk(3, overfill_once=False, cap_property=CAP_PROP)
|
||||
i1 = _Item(4)
|
||||
i1.size = "wrong type" # type: ignore
|
||||
|
||||
with pytest.raises(TypeError, match="Capacity property must be an integer"):
|
||||
chunk.remove(i1)
|
||||
@ -1,25 +0,0 @@
|
||||
from dopt_basics import iteration
|
||||
|
||||
|
||||
def test_symmetric_iter_1():
|
||||
l1_in = [1, 2, 3, 4, 5, 6]
|
||||
l1_out = [1, 6, 2, 5, 3, 4]
|
||||
|
||||
l1_calc = list(iteration.symmetric_iter(l1_in))
|
||||
assert len(l1_in) == len(l1_calc)
|
||||
assert len(l1_out) == len(l1_calc)
|
||||
|
||||
for truth, calc in zip(l1_out, l1_calc):
|
||||
assert truth == calc
|
||||
|
||||
|
||||
def test_symmetric_iter_2():
|
||||
l1_in = [1, 2, 3, 4, 5]
|
||||
l1_out = [1, 5, 2, 4, 3]
|
||||
|
||||
l1_calc = list(iteration.symmetric_iter(l1_in))
|
||||
assert len(l1_in) == len(l1_calc)
|
||||
assert len(l1_out) == len(l1_calc)
|
||||
|
||||
for truth, calc in zip(l1_out, l1_calc):
|
||||
assert truth == calc
|
||||
@ -1,57 +0,0 @@
|
||||
import dataclasses as dc
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
from dopt_basics import system
|
||||
|
||||
|
||||
@dc.dataclass()
|
||||
class _TData:
|
||||
dic: dict[str, str]
|
||||
dic2: dict[str, dict[str, int]]
|
||||
lst: list[int]
|
||||
tup: tuple[str, ...]
|
||||
string: str
|
||||
id_: int
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def TData() -> _TData:
|
||||
return _TData(
|
||||
{"test": "test", "test2": "test"},
|
||||
{"test": {"prop1": 3, "prop2": 500}},
|
||||
[1, 2, 3, 4, 5, 6, 7, 8, 9],
|
||||
("test", "test", "test", "test", "test", "test", "test"),
|
||||
"This is one test string",
|
||||
1234,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def TData_real_size(TData) -> int:
|
||||
return 1435
|
||||
|
||||
|
||||
def test_obj_size(TData, TData_real_size):
|
||||
ret = system.obj_size(TData)
|
||||
|
||||
assert ret == TData_real_size
|
||||
|
||||
|
||||
def test_obj_size_kb(TData, TData_real_size):
|
||||
ret = system.obj_size(TData, unit="kb")
|
||||
|
||||
assert ret == pytest.approx(TData_real_size / 1024)
|
||||
|
||||
|
||||
def test_obj_size_mb(TData, TData_real_size):
|
||||
ret = system.obj_size(TData, unit="mb")
|
||||
|
||||
assert ret == pytest.approx(TData_real_size / 1024**2)
|
||||
|
||||
|
||||
def test_obj_size_gb(TData, TData_real_size):
|
||||
ret = system.obj_size(TData, unit="gb")
|
||||
|
||||
assert ret == pytest.approx(TData_real_size / 1024**3)
|
||||
Loading…
x
Reference in New Issue
Block a user