Skip to content

proxystore.utils

General purpose utility functions.

chunk_bytes 

chunk_bytes(
    data: bytes, chunk_size: int
) -> Generator[bytes, None, None]

Yield chunks of binary data.

Parameters:

  • data (bytes) –

    Data to be chunked.

  • chunk_size (int) –

    Chunk size in bytes.

Returns:

  • Generator[bytes, None, None] –

    Generator that yields chunks of bytes.

Source code in proxystore/utils.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def chunk_bytes(
    data: bytes,
    chunk_size: int,
) -> Generator[bytes, None, None]:
    """Yield chunks of binary data.

    Args:
        data: Data to be chunked.
        chunk_size: Chunk size in bytes.

    Returns:
        Generator that yields chunks of bytes.
    """
    length = len(data)
    for index in range(0, length, chunk_size):
        yield data[index : min(index + chunk_size, length)]

create_key 

create_key(obj: Any) -> str

Generate key for the object.

Parameters:

  • obj (Any) –

    Object to create key for.

Returns:

  • str –

    A random 128 bit string.

Source code in proxystore/utils.py 
31
32
33
34
35
36
37
38
39
40
def create_key(obj: Any) -> str:
    """Generate key for the object.

    Args:
        obj: Object to create key for.

    Returns:
        A random 128 bit string.
    """
    return str(random.getrandbits(128))

fullname 

fullname(obj: Any) -> str

Return full name of object.

Source code in proxystore/utils.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
def fullname(obj: Any) -> str:
    """Return full name of object."""
    if hasattr(obj, '__module__'):
        module = obj.__module__
    else:
        module = obj.__class__.__module__
    if hasattr(obj, '__name__'):
        name = obj.__name__
    else:
        name = obj.__class__.__name__
    if module is None or module == str.__module__:
        return name
    return f'{module}.{name}'

home_dir 

home_dir() -> str

Return the absolute path to the proxystore home directory.

If set, $PROXYSTORE_HOME is preferred. Otherwise, $XDG_DATA_HOME/proxystore is returned where $XDG_DATA_HOME defaults to $HOME/.local/share if unset.

Source code in proxystore/utils.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def home_dir() -> str:
    """Return the absolute path to the proxystore home directory.

    If set, ``$PROXYSTORE_HOME`` is preferred. Otherwise,
    ``$XDG_DATA_HOME/proxystore`` is returned where ``$XDG_DATA_HOME`` defaults
    to ``$HOME/.local/share`` if unset.
    """
    path = os.environ.get('PROXYSTORE_HOME')
    if path is None:
        prefix = os.environ.get('XDG_DATA_HOME') or os.path.expanduser(
            '~/.local/share',
        )
        path = os.path.join(prefix, 'proxystore')
    return os.path.abspath(path)

hostname 

hostname() -> str

Return current hostname.

Source code in proxystore/utils.py 
74
75
76
def hostname() -> str:
    """Return current hostname."""
    return socket.gethostname()

bytes_to_readable 

bytes_to_readable(size: int, precision: int = 3) -> str

Convert bytes to human readable value.

Note

This method uses base-10 values for KB, MB, GB, etc. instead of base-2 values (i.e., KiB, MiB, GiB, etc.).

Parameters:

  • size (int) –

    Byte value to make readable.

  • precision (int) –

    Number of decimal places.

Returns:

  • str –

    String with human readable number of bytes.

Raises:

Source code in proxystore/utils.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def bytes_to_readable(size: int, precision: int = 3) -> str:
    """Convert bytes to human readable value.

    Note:
        This method uses base-10 values for KB, MB, GB, etc. instead of
        base-2 values (i.e., KiB, MiB, GiB, etc.).

    Args:
        size: Byte value to make readable.
        precision: Number of decimal places.

    Returns:
        String with human readable number of bytes.

    Raises:
        ValueError: If size is negative.
    """
    kb = int(1e3)
    mb = int(1e6)
    gb = int(1e9)
    tb = int(1e12)

    size_ = float(size)
    if 0 <= size < kb:
        suffix = 'B'
    elif kb <= size < mb:
        suffix = 'KB'
        size_ /= kb
    elif mb <= size < gb:
        suffix = 'MB'
        size_ /= mb
    elif gb <= size < tb:
        suffix = 'GB'
        size_ /= gb
    elif tb <= size:
        suffix = 'TB'
        size_ /= tb
    else:
        raise ValueError(f'Size ({size}) cannot be negative.')

    value = str(round(size_, precision))
    value = value.rstrip('0').rstrip('.')
    return f'{value} {suffix}'

readable_to_bytes 

readable_to_bytes(size: str) -> int

Convert string with bytes units to the integer value of bytes.

Example 
>>> readable_to_bytes('1.2 KB')
1200
>>> readable_to_bytes('0.6 MiB')
629146

Parameters:

  • size (str) –

    String to parse for bytes size.

Returns:

  • int –

    Integer number of bytes parsed from the string.

Raises:

  • ValueError –

    If the input string contains more than two parts (i.e., a value and a unit).

  • ValueError –

    If the unit is not one of KB, MB, GB, TB, KiB, MiB, GiB, or TiB.

  • ValueError –

    If the value cannot be cast to a float.

Source code in proxystore/utils.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def readable_to_bytes(size: str) -> int:
    """Convert string with bytes units to the integer value of bytes.

    Example:
        ```python
        >>> readable_to_bytes('1.2 KB')
        1200
        >>> readable_to_bytes('0.6 MiB')
        629146
        ```

    Args:
        size: String to parse for bytes size.

    Returns:
        Integer number of bytes parsed from the string.

    Raises:
        ValueError: If the input string contains more than two parts (i.e., a
            value and a unit).
        ValueError: If the unit is not one of KB, MB, GB, TB, KiB, MiB, GiB,
            or TiB.
        ValueError: If the value cannot be cast to a float.
    """
    units_to_bytes = {
        'b': 1,
        'kb': int(1e3),
        'mb': int(1e6),
        'gb': int(1e9),
        'tb': int(1e12),
        'kib': int(2**10),
        'mib': int(2**20),
        'gib': int(2**30),
        'tib': int(2**40),
    }

    # Try casting size to value (will only work if no units)
    try:
        return int(float(size))
    except ValueError:
        pass

    # Ensure space between value and unit
    size = re.sub(r'([a-zA-Z]+)', r' \1', size.strip())

    parts = [s.strip() for s in size.split()]
    if len(parts) != 2:
        raise ValueError(
            'Input string "{size}" must contain only a value and a unit.',
        )

    value, unit = parts

    try:
        value_size = decimal.Decimal(value)
    except decimal.InvalidOperation as e:
        raise ValueError(f'Unable to interpret "{value}" as a float.') from e
    try:
        unit_size = units_to_bytes[unit.lower()]
    except KeyError as e:
        raise ValueError(f'Unknown unit type {unit}.') from e

    return int(value_size * unit_size)