Skip to content

proxystore.utils

proxystore/utils.py

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)]

get_class_path() 

get_class_path(cls: type[Any]) -> str

Get the fully qualified path of a type.

Example 
>>> from proxystore.connectors.connector import Connector
>>> get_class_path(Connector)
'proxystore.connectors.connector.Connector'

Parameters:

  • cls (type[Any]) –

    Class type to get fully qualified path of.

Returns:

  • str –

    Fully qualified path of cls.

Source code in proxystore/utils.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def get_class_path(cls: type[Any]) -> str:
    """Get the fully qualified path of a type.

    Example:
        ```python
        >>> from proxystore.connectors.connector import Connector
        >>> get_class_path(Connector)
        'proxystore.connectors.connector.Connector'
        ```

    Args:
        cls: Class type to get fully qualified path of.

    Returns:
        Fully qualified path of `cls`.
    """
    return f'{cls.__module__}.{cls.__qualname__}'

import_class() 

import_class(path: str) -> type[Any]

Import class via its fully qualified path.

Example 
>>> import_class('proxystore.connectors.connector.Connector')
<class 'proxystore.connectors.connector.Connector'>

Parameters:

  • path (str) –

    Fully qualified path of class to import.

Returns:

Raises:

Source code in proxystore/utils.py 
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
def import_class(path: str) -> type[Any]:
    """Import class via its fully qualified path.

    Example:
        ```python
        >>> import_class('proxystore.connectors.connector.Connector')
        <class 'proxystore.connectors.connector.Connector'>
        ```

    Args:
        path: Fully qualified path of class to import.

    Returns:
        Imported class.

    Raises:
        ImportError: If a class at the `path` is not found.
    """
    module_path, _, name = path.rpartition('.')
    if len(module_path) == 0:
        raise ImportError(
            f'Class path must contain at least one module. Got {path}',
        )
    module = importlib.import_module(module_path)
    return getattr(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 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
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 
93
94
95
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 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
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 
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
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
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)