-
Notifications
You must be signed in to change notification settings - Fork 9
Expand file tree
/
Copy pathchunksize.py
More file actions
93 lines (76 loc) · 2.62 KB
/
chunksize.py
File metadata and controls
93 lines (76 loc) · 2.62 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
#
# MIT License
#
# Copyright (c) 2024, Mattias Aabmets
#
# The contents of this file are subject to the terms and conditions defined in the License.
# You may not use, modify, or distribute this file except in compliance with the License.
#
# SPDX-License-Identifier: MIT
#
import typing as t
from pydantic import Field, validate_call
from dataclasses import dataclass
from quantcrypt.internal import errors
__all__ = ["KBLiteral", "MBLiteral", "ChunkSizeKB", "ChunkSizeMB", "ChunkSize"]
KBLiteral = t.Literal[1, 2, 4, 8, 16, 32, 64, 128, 256]
MBLiteral = t.Literal[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
@dataclass(frozen=True)
class ChunkSizeKB:
value: int
@validate_call
def __init__(self, size: KBLiteral) -> None:
"""
:param size: The chunk size in kilobytes.
:raises - pydantic.ValidationError:
On invalid size argument value.
"""
object.__setattr__(self, 'value', 1024 * size)
@dataclass(frozen=True)
class ChunkSizeMB:
value: int
@validate_call
def __init__(self, size: MBLiteral) -> None:
"""
:param size: The chunk size in megabytes.
:raises - pydantic.ValidationError:
On invalid size argument value.
"""
object.__setattr__(self, 'value', 1024 ** 2 * size)
class ChunkSize:
atd = t.Annotated[
t.Optional[ChunkSizeKB | ChunkSizeMB],
Field(default=None)
]
def __init__(self):
"""
This class is a collection of classes and is not
intended to be instantiated directly. You can access
the contained **KB** and **MB** classes as attributes
of this class.
"""
raise errors.InvalidUsageError(
"ChunkSize class is a collection of classes and "
"is not intended to be instantiated directly."
)
KB: t.Type[ChunkSizeKB] = ChunkSizeKB
MB: t.Type[ChunkSizeMB] = ChunkSizeMB
@staticmethod
def determine_from_data_size(data_size: int) -> ChunkSizeKB | ChunkSizeMB:
kilo_bytes = 1024
mega_bytes = kilo_bytes * 1024
if data_size <= kilo_bytes * 4:
return ChunkSizeKB(1)
elif data_size <= kilo_bytes * 16:
return ChunkSizeKB(4)
elif data_size <= kilo_bytes * 64:
return ChunkSizeKB(16)
elif data_size <= kilo_bytes * 256:
return ChunkSizeKB(64)
elif data_size <= kilo_bytes * 1024:
return ChunkSizeKB(256)
for x in range(0, 10):
x += 1
if data_size <= mega_bytes * x * 100:
return ChunkSizeMB(t.cast(MBLiteral, x))
return ChunkSizeMB(10)