Skip to content

contract.harp

HarpRegister 

HarpRegister(
    name: str,
    *,
    description: Optional[str] = None,
    reader_params: Optional[TReaderParams] = None,
    **kwargs,
)

Bases: DataStream[DataFrame, HarpRegisterParams]

Harp device register data stream provider.

A data stream implementation for reading Harp device register data using the Harp Python library.

Parameters:

Name Type Description Default
DataStream

Base class for data stream providers.

 required
Source code in src/contraqctor/contract/base.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(
    self: Self,
    name: str,
    *,
    description: Optional[str] = None,
    reader_params: Optional[_typing.TReaderParams] = None,
    **kwargs,
) -> None:
    if "::" in name:
        raise ValueError("Name cannot contain '::' character.")
    self._name = name

    self._description = description
    self._reader_params = reader_params if reader_params is not None else _typing.UnsetParams
    self._data = _typing.UnsetData
    self._parent: Optional["DataStream"] = None

name property 

name: str

Get the name of the data stream.

Returns:

Name Type Description
str str

Name identifier of the data stream.

resolved_name property 

resolved_name: str

Get the full hierarchical name of the data stream.

Generates a path-like name showing the stream's position in the hierarchy, using '::' as a separator between parent and child names.

Returns:

Name Type Description
str str

The fully resolved name including all parent names.

description property 

description: Optional[str]

Get the description of the data stream.

Returns:

Type Description
Optional[str]

Optional[str]: Description of the data stream, or None if not provided.

parent property 

parent: Optional[DataStream]

Get the parent data stream.

Returns:

Type Description
Optional[DataStream]

Optional[DataStream]: Parent data stream, or None if this is a root stream.

is_collection property 

is_collection: bool

Check if this data stream is a collection of other streams.

Returns:

Name Type Description
bool bool

True if this is a collection stream, False otherwise.

reader_params property 

reader_params: TReaderParams

Get the parameters for the data reader.

Returns:

Name Type Description
TReaderParams TReaderParams

Parameters for the data reader.

has_data property 

has_data: bool

Check if the data stream has loaded data.

Returns:

Name Type Description
bool bool

True if data has been loaded, False otherwise.

data property 

data: TData

Get the loaded data.

Returns:

Name Type Description
TData TData

The loaded data.

Raises:

Type Description
ValueError

If data has not been loaded yet.

read 

read(
    reader_params: Optional[HarpRegisterParams] = None,
) -> DataFrame

Read register data from Harp binary files.

Parameters:

Name Type Description Default
reader_params Optional[HarpRegisterParams]

Parameters for register reading configuration.

 None

Returns:

Type Description
DataFrame

pd.DataFrame: DataFrame containing the register data.

Raises:

Type Description
ValueError

If reader parameters are not set.
Source code in src/contraqctor/contract/harp.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@override
def read(self, reader_params: Optional[HarpRegisterParams] = None) -> pd.DataFrame:
    """Read register data from Harp binary files.

    Args:
        reader_params: Parameters for register reading configuration.

    Returns:
        pd.DataFrame: DataFrame containing the register data.

    Raises:
        ValueError: If reader parameters are not set.
    """
    reader_params = reader_params if reader_params is not None else self._reader_params
    if _typing.is_unset(reader_params):
        raise ValueError("Reader parameters are not set. Cannot read data.")
    return self._reader(reader_params.base_path, epoch=reader_params.epoch, keep_type=reader_params.keep_type)

from_register_reader classmethod 

from_register_reader(
    name: str,
    reg_reader: RegisterReader,
    params: HarpRegisterParams = _DEFAULT_HARP_READER_PARAMS,
) -> Self

Create a HarpRegister data stream from a RegisterReader.

Factory method to create a HarpRegister instance from an existing Harp RegisterReader object.

Parameters:

Name Type Description Default
name str

Name for the register data stream.

 required
reg_reader RegisterReader

Harp RegisterReader object.

 required
params HarpRegisterParams

Parameters for register reading configuration.

 _DEFAULT_HARP_READER_PARAMS

Returns:

Name Type Description
HarpRegister Self

Newly created HarpRegister data stream.
Source code in src/contraqctor/contract/harp.py 
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
@classmethod
def from_register_reader(
    cls,
    name: str,
    reg_reader: harp.reader.RegisterReader,
    params: HarpRegisterParams = _DEFAULT_HARP_READER_PARAMS,
) -> Self:
    """Create a HarpRegister data stream from a RegisterReader.

    Factory method to create a HarpRegister instance from an existing
    Harp RegisterReader object.

    Args:
        name: Name for the register data stream.
        reg_reader: Harp RegisterReader object.
        params: Parameters for register reading configuration.

    Returns:
        HarpRegister: Newly created HarpRegister data stream.
    """
    c = cls(
        name=name,
        description=reg_reader.register.description,
    )
    c.bind_reader_params(params)
    c._reader = reg_reader.read
    c.make_params = cls.make_params
    return c

bind_reader_params 

bind_reader_params(params: TReaderParams) -> Self

Bind reader parameters to the data stream.

Parameters:

Name Type Description Default
params TReaderParams

Parameters to bind to the data stream's reader.

 required

Returns:

Name Type Description
Self Self

The data stream instance for method chaining.

Raises:

Type Description
ValueError

If reader parameters have already been set.
Source code in src/contraqctor/contract/base.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def bind_reader_params(self, params: _typing.TReaderParams) -> Self:
    """Bind reader parameters to the data stream.

    Args:
        params: Parameters to bind to the data stream's reader.

    Returns:
        Self: The data stream instance for method chaining.

    Raises:
        ValueError: If reader parameters have already been set.
    """
    if not _typing.is_unset(self._reader_params):
        raise ValueError("Reader parameters are already set. Cannot bind again.")
    self._reader_params = params
    return self

at 

at(name: str) -> DataStream

Get a child data stream by name.

Parameters:

Name Type Description Default
name str

Name of the child data stream to retrieve.

 required

Returns:

Name Type Description
DataStream DataStream

The child data stream with the given name.

Raises:

Type Description
NotImplementedError

If the data stream does not support child access.

Examples:

# Access stream in a collection
collection = data_collection.load()
temp_stream = collection.at("temperature")

# Or using dictionary-style syntax
humidity_stream = collection["humidity"]
Source code in src/contraqctor/contract/base.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def at(self, name: str) -> "DataStream":
    """Get a child data stream by name.

    Args:
        name: Name of the child data stream to retrieve.

    Returns:
        DataStream: The child data stream with the given name.

    Raises:
        NotImplementedError: If the data stream does not support child access.

    Examples:
        ```python
        # Access stream in a collection
        collection = data_collection.load()
        temp_stream = collection.at("temperature")

        # Or using dictionary-style syntax
        humidity_stream = collection["humidity"]
        ```
    """
    raise NotImplementedError("This method is not implemented for DataStream.")

load 

load() -> Self

Load data into the data stream.

Reads data from the source and stores it in the data stream.

Returns:

Name Type Description
Self Self

The data stream instance for method chaining.

Examples:

from contraqctor.contract import csv

# Create and load a CSV stream
params = csv.CsvParams(path="data/measurements.csv")
csv_stream = csv.Csv("measurements", reader_params=params)
csv_stream.load()

# Access the data
df = csv_stream.data
print(f"Loaded {len(df)} rows")
Source code in src/contraqctor/contract/base.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def load(self) -> Self:
    """Load data into the data stream.

    Reads data from the source and stores it in the data stream.

    Returns:
        Self: The data stream instance for method chaining.

    Examples:
        ```python
        from contraqctor.contract import csv

        # Create and load a CSV stream
        params = csv.CsvParams(path="data/measurements.csv")
        csv_stream = csv.Csv("measurements", reader_params=params)
        csv_stream.load()

        # Access the data
        df = csv_stream.data
        print(f"Loaded {len(df)} rows")
        ```
    """
    self._data = self.read()
    return self

load_all 

load_all(
    strict: bool = False,
) -> list[tuple[DataStream, Exception], None, None]

Recursively load this data stream and all child streams.

Performs depth-first traversal to load all streams in the hierarchy.

Parameters:

Name Type Description Default
strict bool

If True, raises exceptions immediately; otherwise collects and returns them.

 False

Returns:

Name Type Description
list list[tuple[DataStream, Exception], None, None]

List of tuples containing streams and exceptions that occurred during loading.

Raises:

Type Description
Exception

If strict is True and an exception occurs during loading.

Examples:

# Load all streams and handle errors
errors = collection.load_all(strict=False)

if errors:
    for stream, error in errors:
        print(f"Error loading {stream.name}: {error}")
Source code in src/contraqctor/contract/base.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
def load_all(self, strict: bool = False) -> list[tuple["DataStream", Exception], None, None]:
    """Recursively load this data stream and all child streams.

    Performs depth-first traversal to load all streams in the hierarchy.

    Args:
        strict: If True, raises exceptions immediately; otherwise collects and returns them.

    Returns:
        list: List of tuples containing streams and exceptions that occurred during loading.

    Raises:
        Exception: If strict is True and an exception occurs during loading.

    Examples:
        ```python
        # Load all streams and handle errors
        errors = collection.load_all(strict=False)

        if errors:
            for stream, error in errors:
                print(f"Error loading {stream.name}: {error}")
        ```
    """
    self.load()
    exceptions = []
    for stream in self:
        if stream is None:
            continue
        try:
            exceptions += stream.load_all(strict=strict)
        except Exception as e:
            if strict:
                raise e
            exceptions.append((stream, e))
    return exceptions

DeviceYmlByWhoAmI

Bases: _DeviceYmlSource

Device YAML source that finds the file using WhoAmI value.

Specifies that the device YAML should be obtained by looking up a device by its WhoAmI identifier.

Attributes:

Name Type Description
method Literal['whoami']

Fixed as "whoami".
who_am_i Annotated[int, Field(ge=0, le=9999, description='WhoAmI value')]

The WhoAmI value of the device (0-9999).

DeviceYmlByFile

Bases: _DeviceYmlSource

Device YAML source that specifies a file path.

Specifies that the device YAML should be loaded from a local file path.

Attributes:

Name Type Description
method Literal['file']

Fixed as "file".
path Optional[PathLike | str]

Optional path to the device YAML file. If None, assumes "device.yml" in the data directory.

DeviceYmlByUrl

Bases: _DeviceYmlSource

Device YAML source that fetches from a URL.

Specifies that the device YAML should be downloaded from a URL.

Attributes:

Name Type Description
method Literal['http']

Fixed as "http".
url AnyHttpUrl

HTTP URL to download the device YAML file from.

DeviceYmlByRegister0

Bases: _DeviceYmlSource

Device YAML source that infers from register 0 file.

Specifies that the device YAML should be determined by finding and reading the WhoAmI value from register 0 files.

Attributes:

Name Type Description
method Literal['register0']

Fixed as "register0".
register0_glob_pattern List[str]

List of glob patterns to locate register 0 files.

HarpDeviceParams dataclass 

HarpDeviceParams(path: PathLike)

Bases: FilePathBaseParam

Parameters for Harp device data reading.

Defines parameters for locating and reading Harp device data.

Attributes:

Name Type Description
device_yml_hint DeviceYmlSource

Source for the device YAML configuration file.
include_common_registers bool

Whether to include common registers. Defaults to True.
keep_type bool

Whether to preserve type information. Defaults to True.
epoch Optional[datetime]

Reference datetime for timestamp calculations. If provided, timestamps are converted to datetime.

HarpDevice 

HarpDevice(
    name: str,
    *,
    description: Optional[str] = None,
    reader_params: Optional[TReaderParams] = None,
    **kwargs,
)

Bases: DataStreamCollectionBase[HarpRegister, HarpDeviceParams]

Harp device data stream collection provider.

A data stream collection for accessing all registers of a Harp device.

Parameters:

Name Type Description Default
DataStreamCollectionBase

Base class for data stream collection providers.

 required

Examples:

from contraqctor.contract.harp import HarpDevice, HarpDeviceParams, DeviceYmlByWhoAmI

# Create and load a device stream
params = HarpDeviceParams(
    path="behavior.harp",
    device_yml_hint=DeviceYmlByWhoAmI(who_am_i=1216)
)

behavior = HarpDevice("behavior", reader_params=params).load()

# Access registers
digital_input = behavior["DigitalInputState"].data
adc = behavior["AnalogData"].data
Source code in src/contraqctor/contract/base.py 
317
318
319
320
321
322
323
324
325
326
327
def __init__(
    self: Self,
    name: str,
    *,
    description: Optional[str] = None,
    reader_params: Optional[_typing.TReaderParams] = None,
    **kwargs,
) -> None:
    super().__init__(name=name, description=description, reader_params=reader_params, **kwargs)
    self._hashmap: Dict[str, TDataStream] = {}
    self._update_hashmap()

device_reader property 

device_reader: DeviceReader

Get the underlying Harp device reader.

Returns:

Type Description
DeviceReader

harp.reader.DeviceReader: Harp device reader for accessing raw device functionality.

Raises:

Type Description
ValueError

If the device reader has not been initialized.

name property 

name: str

Get the name of the data stream.

Returns:

Name Type Description
str str

Name identifier of the data stream.

resolved_name property 

resolved_name: str

Get the full hierarchical name of the data stream.

Generates a path-like name showing the stream's position in the hierarchy, using '::' as a separator between parent and child names.

Returns:

Name Type Description
str str

The fully resolved name including all parent names.

description property 

description: Optional[str]

Get the description of the data stream.

Returns:

Type Description
Optional[str]

Optional[str]: Description of the data stream, or None if not provided.

parent property 

parent: Optional[DataStream]

Get the parent data stream.

Returns:

Type Description
Optional[DataStream]

Optional[DataStream]: Parent data stream, or None if this is a root stream.

is_collection property 

is_collection: bool

Check if this data stream is a collection of other streams.

Returns:

Name Type Description
bool bool

True if this is a collection stream, False otherwise.

reader_params property 

reader_params: TReaderParams

Get the parameters for the data reader.

Returns:

Name Type Description
TReaderParams TReaderParams

Parameters for the data reader.

has_data property 

has_data: bool

Check if the data stream has loaded data.

Returns:

Name Type Description
bool bool

True if data has been loaded, False otherwise.

data property 

data: TData

Get the loaded data.

Returns:

Name Type Description
TData TData

The loaded data.

Raises:

Type Description
ValueError

If data has not been loaded yet.

read 

read(
    reader_params: Optional[TReaderParams] = None,
) -> TData

Read data using the configured reader.

Parameters:

Name Type Description Default
reader_params Optional[TReaderParams]

Optional parameters to override the default reader parameters.

 None

Returns:

Name Type Description
TData TData

Data read from the source.

Raises:

Type Description
ValueError

If reader parameters are not set.
Source code in src/contraqctor/contract/base.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def read(self, reader_params: Optional[_typing.TReaderParams] = None) -> _typing.TData:
    """Read data using the configured reader.

    Args:
        reader_params: Optional parameters to override the default reader parameters.

    Returns:
        TData: Data read from the source.

    Raises:
        ValueError: If reader parameters are not set.
    """
    reader_params = reader_params if reader_params is not None else self._reader_params
    if _typing.is_unset(reader_params):
        raise ValueError("Reader parameters are not set. Cannot read data.")
    return self._reader(reader_params)

bind_reader_params 

bind_reader_params(params: TReaderParams) -> Self

Bind reader parameters to the data stream.

Parameters:

Name Type Description Default
params TReaderParams

Parameters to bind to the data stream's reader.

 required

Returns:

Name Type Description
Self Self

The data stream instance for method chaining.

Raises:

Type Description
ValueError

If reader parameters have already been set.
Source code in src/contraqctor/contract/base.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def bind_reader_params(self, params: _typing.TReaderParams) -> Self:
    """Bind reader parameters to the data stream.

    Args:
        params: Parameters to bind to the data stream's reader.

    Returns:
        Self: The data stream instance for method chaining.

    Raises:
        ValueError: If reader parameters have already been set.
    """
    if not _typing.is_unset(self._reader_params):
        raise ValueError("Reader parameters are already set. Cannot bind again.")
    self._reader_params = params
    return self

at 

at(name: str) -> TDataStream

Get a child data stream by name.

Parameters:

Name Type Description Default
name str

Name of the child data stream to retrieve.

 required

Returns:

Name Type Description
TDataStream TDataStream

The child data stream with the given name.

Raises:

Type Description
ValueError

If data has not been loaded yet.
KeyError

If no child stream with the given name exists.
Source code in src/contraqctor/contract/base.py 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
@override
def at(self, name: str) -> TDataStream:
    """Get a child data stream by name.

    Args:
        name: Name of the child data stream to retrieve.

    Returns:
        TDataStream: The child data stream with the given name.

    Raises:
        ValueError: If data has not been loaded yet.
        KeyError: If no child stream with the given name exists.
    """
    if not self.has_data:
        raise ValueError("data streams have not been read yet. Cannot access data streams.")
    if name in self._hashmap:
        return self._hashmap[name]
    else:
        raise KeyError(f"Stream with name: '{name}' not found in data streams.")

load 

load()

Load data for this collection.

Overrides the base method to add validation that loaded data is a list of DataStreams.

Returns:

Name Type Description
Self

The collection instance for method chaining.

Raises:

Type Description
ValueError

If loaded data is not a list of DataStreams.
Source code in src/contraqctor/contract/base.py 
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
@override
def load(self):
    """Load data for this collection.

    Overrides the base method to add validation that loaded data is a list of DataStreams.

    Returns:
        Self: The collection instance for method chaining.

    Raises:
        ValueError: If loaded data is not a list of DataStreams.
    """
    super().load()
    if not isinstance(self._data, list):
        self._data = _typing.UnsetData
        raise ValueError("Data must be a list of DataStreams.")
    self._update_hashmap()
    return self

load_all 

load_all(
    strict: bool = False,
) -> list[tuple[DataStream, Exception], None, None]

Recursively load this data stream and all child streams.

Performs depth-first traversal to load all streams in the hierarchy.

Parameters:

Name Type Description Default
strict bool

If True, raises exceptions immediately; otherwise collects and returns them.

 False

Returns:

Name Type Description
list list[tuple[DataStream, Exception], None, None]

List of tuples containing streams and exceptions that occurred during loading.

Raises:

Type Description
Exception

If strict is True and an exception occurs during loading.

Examples:

# Load all streams and handle errors
errors = collection.load_all(strict=False)

if errors:
    for stream, error in errors:
        print(f"Error loading {stream.name}: {error}")
Source code in src/contraqctor/contract/base.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
def load_all(self, strict: bool = False) -> list[tuple["DataStream", Exception], None, None]:
    """Recursively load this data stream and all child streams.

    Performs depth-first traversal to load all streams in the hierarchy.

    Args:
        strict: If True, raises exceptions immediately; otherwise collects and returns them.

    Returns:
        list: List of tuples containing streams and exceptions that occurred during loading.

    Raises:
        Exception: If strict is True and an exception occurs during loading.

    Examples:
        ```python
        # Load all streams and handle errors
        errors = collection.load_all(strict=False)

        if errors:
            for stream, error in errors:
                print(f"Error loading {stream.name}: {error}")
        ```
    """
    self.load()
    exceptions = []
    for stream in self:
        if stream is None:
            continue
        try:
            exceptions += stream.load_all(strict=strict)
        except Exception as e:
            if strict:
                raise e
            exceptions.append((stream, e))
    return exceptions

iter_all 

iter_all() -> Generator[DataStream, None, None]

Iterator for all child data streams, including nested collections.

Implements a depth-first traversal of the stream hierarchy.

Yields:

Name Type Description
DataStream DataStream

All recursively yielded child data streams.
Source code in src/contraqctor/contract/base.py 
436
437
438
439
440
441
442
443
444
445
446
447
448
def iter_all(self) -> Generator[DataStream, None, None]:
    """Iterator for all child data streams, including nested collections.

    Implements a depth-first traversal of the stream hierarchy.

    Yields:
        DataStream: All recursively yielded child data streams.
    """
    for value in self:
        if isinstance(value, DataStream):
            yield value
        if isinstance(value, DataStreamCollectionBase):
            yield from value.iter_all()

fetch_yml_from_who_am_i 

fetch_yml_from_who_am_i(
    who_am_i: int, release: str = "main"
) -> BytesIO

Fetch a device YAML file based on its WhoAmI identifier.

Looks up the device in the WhoAmI registry and downloads its YAML file.

Parameters:

Name Type Description Default
who_am_i int

WhoAmI identifier of the device.

 required
release str

Git branch or tag to use for fetching the YAML file.

 'main'

Returns:

Type Description
BytesIO

io.BytesIO: Memory buffer containing the device YAML content.

Raises:

Type Description
KeyError

If the WhoAmI identifier is not found in the registry.
ValueError

If required repository information is missing or YAML file cannot be found.
Source code in src/contraqctor/contract/harp.py 
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
def fetch_yml_from_who_am_i(who_am_i: int, release: str = "main") -> io.BytesIO:
    """Fetch a device YAML file based on its WhoAmI identifier.

    Looks up the device in the WhoAmI registry and downloads its YAML file.

    Args:
        who_am_i: WhoAmI identifier of the device.
        release: Git branch or tag to use for fetching the YAML file.

    Returns:
        io.BytesIO: Memory buffer containing the device YAML content.

    Raises:
        KeyError: If the WhoAmI identifier is not found in the registry.
        ValueError: If required repository information is missing or YAML file cannot be found.
    """
    try:
        device = fetch_who_am_i_list()[who_am_i]
    except KeyError as e:
        raise KeyError(f"WhoAmI {who_am_i} not found in whoami.yml") from e

    repository_url = device.get("repositoryUrl", None)

    if repository_url is None:
        raise ValueError("Device's repositoryUrl not found in whoami.yml")

    _repo_hint_paths = [
        "{repository_url}/{release}/device.yml",
        "{repository_url}/{release}/software/bonsai/device.yml",
    ]

    yml = None
    for hint in _repo_hint_paths:
        url = hint.format(repository_url=repository_url, release=release)
        if "github.com" in url:
            url = url.replace("github.com", "raw.githubusercontent.com")
        response = requests.get(url, allow_redirects=True, timeout=5)
        if response.status_code == 200:
            yml = io.BytesIO(response.content)
            return yml

    raise ValueError("device.yml not found in any repository")

fetch_who_am_i_list cached 

fetch_who_am_i_list(
    url: str = "https://raw.githubusercontent.com/harp-tech/whoami/main/whoami.yml",
) -> Dict[int, Any]

Fetch and parse the Harp WhoAmI registry.

Downloads and parses the WhoAmI registry YAML file from GitHub. Results are cached for efficiency.

Parameters:

Name Type Description Default
url str

URL to the WhoAmI registry YAML file.

 'https://raw.githubusercontent.com/harp-tech/whoami/main/whoami.yml'

Returns:

Type Description
Dict[int, Any]

Dict[int, Any]: Dictionary mapping WhoAmI identifiers to device information.
Source code in src/contraqctor/contract/harp.py 
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
@cache
def fetch_who_am_i_list(
    url: str = "https://raw.githubusercontent.com/harp-tech/whoami/main/whoami.yml",
) -> Dict[int, Any]:
    """Fetch and parse the Harp WhoAmI registry.

    Downloads and parses the WhoAmI registry YAML file from GitHub.
    Results are cached for efficiency.

    Args:
        url: URL to the WhoAmI registry YAML file.

    Returns:
        Dict[int, Any]: Dictionary mapping WhoAmI identifiers to device information.
    """
    response = requests.get(url, allow_redirects=True, timeout=5)
    content = response.content.decode("utf-8")
    content = yaml.safe_load(content)
    devices = content["devices"]
    return devices