Skip to content

Hub API Reference

xovis.api.hub.client

Xovis SDK - HUB Cloud Client

This module resides within the State & Topology Plane, serving as the primary entry point for fleet orchestration via the Xovis HUB Cloud. It coordinates OAuth2 authentication, dynamic OpenAPI device tunneling, and concurrent bulk operations across distributed edge sensors.

Classes

HubClient

Asynchronous client for interacting with the Xovis HUB Cloud.

Manages the complete lifecycle of the Xovis fleet, including automated OAuth2 token rotation, client-side state caching, and secure Hub-to-Edge tunneling. Acts as the definitive orchestrator for distributed sensor networks.

Source code in src/xovis/api/hub/client.py 
 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
 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
122
123
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
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
class HubClient:
    """
    Asynchronous client for interacting with the Xovis HUB Cloud.

    Manages the complete lifecycle of the Xovis fleet, including automated
    OAuth2 token rotation, client-side state caching, and secure Hub-to-Edge
    tunneling. Acts as the definitive orchestrator for distributed sensor networks.
    """

    def __init__(
        self,
        client_id: Optional[str] = None,
        client_secret: Optional[str] = None,
        token: Optional[str] = None,
        base_url: str = "https://api.xovis.cloud",
        token_url: str = "https://login.xovis.cloud/oauth/token",
        tunnel_base_url: Optional[str] = None,
        timeout: float = 15.0,
        max_retries: int = 5,
        fleet_filter: Optional[dict[str, Any]] = None,
        **kwargs: Any,
    ) -> None:
        """
        Initializes the HubClient and mounts architectural pillars.

        Args:
            client_id (Optional[str], optional): The OAuth2 Client ID provided by Xovis.
                If not provided, resolves from XOVIS_HUB_CLIENT_ID env var.
            client_secret (Optional[str], optional): The OAuth2 Client Secret provided by Xovis.
                If not provided, resolves from XOVIS_HUB_CLIENT_SECRET env var.
            token (Optional[str], optional): Optional static API token.
            base_url (str, optional): The base URL for the Xovis HUB Cloud API.
                Defaults to "https://api.xovis.cloud".
            token_url (str, optional): The Auth0 token endpoint.
                Defaults to "https://login.xovis.cloud/oauth/token".
            tunnel_base_url (Optional[str], optional): Optional override for the
                device tunnel base URL. If not provided, the `base_url` is used.
            timeout (float, optional): Default timeout for HTTP operations.
                Defaults to 15.0.
            max_retries (int, optional): Maximum retry attempts for resilient
                networking. Defaults to 5.
            fleet_filter (Optional[Dict[str, Any]], optional): Client-side filter
                to restrict the visible fleet scope.
            **kwargs (Any): Additional configuration for the HTTP engine.
        """
        import os

        cid = client_id or os.getenv("XOVIS_HUB_CLIENT_ID")
        csec = client_secret or os.getenv("XOVIS_HUB_CLIENT_SECRET")
        static_token = token or os.getenv("XOVIS_HUB_TOKEN")

        if not static_token and (not cid or not csec):
            raise ValueError(
                "Missing Xovis HUB credentials. Provide token, or client_id/client_secret "
                "or set XOVIS_HUB_TOKEN or XOVIS_HUB_CLIENT_ID/XOVIS_HUB_CLIENT_SECRET environment variables."
            )

        self._auth = HubAuth(client_id=cid, client_secret=csec, token_url=token_url, token=static_token)
        self._tunnel_base_url = tunnel_base_url

        # Extract auto_persist_path if provided
        auto_persist_path = kwargs.pop("auto_persist_path", None)

        self._http_client = XovisHTTPClient(base_url=base_url, auth=self._auth, timeout=timeout, max_retries=max_retries, **kwargs)

        self.cache = HubCacheManager(self._http_client, fleet_filter=fleet_filter, auto_persist_path=auto_persist_path)

        self.devices = HubDevicesManager(self._http_client, cache=self.cache)
        self.licenses = HubLicensesManager(self._http_client, cache=self.cache)

    async def connect_device(self, id_or_name: str) -> DeviceClient:
        """
        Spawns a DeviceClient routed through the Hub's dynamic secure tunnel.

        Enables seamless transition from fleet-level management to specific
        edge sensor configuration. Intercepts OAuth2 tokens to authenticate
        the proxied connection.

        The tunnel URL is dynamically constructed per the OpenAPI specification:
        `{base_url}/devices/{mac_address}/tunnel`. The DeviceClient will then
        append its own paths (e.g., `/api/v5/...`).

        CRITICAL: The returned client MUST be used as an asynchronous context
        manager to prevent connection pooling leaks.

        Args:
            id_or_name (str): The MAC address (ID) or human-readable name of
                the target device.

        Returns:
            DeviceClient: A fully hydrated client instance routed via the HUB.
        """
        # Resolves the human-readable name to a MAC address using the local cache
        mac_address = self.devices._resolve_mac_address(id_or_name)

        # Dynamic OpenAPI routing
        base = self._tunnel_base_url or self._http_client.base_url
        tunnel_url = f"{base}/devices/{mac_address}/tunnel"

        client = DeviceClient(
            host=tunnel_url,
            username="hub_tunnel",
            password="hub_tunnel",
            timeout=60.0,  # Explicitly increased for heavy historical aggregations
            max_retries=self._http_client.max_retries,
            limits=httpx.Limits(max_connections=2, max_keepalive_connections=1),
        )

        # Inject Hub authentication for the proxied requests
        client._auth = self._auth
        client._http_client.auth = self._auth
        client._http_client.client.auth = self._auth

        return client

    async def bulk_execute(
        self,
        func: Callable[[DeviceClient], Coroutine[Any, Any, T]],
        fleet_filter: Optional[dict[str, Any]] = None,
    ) -> dict[str, BulkResult[T]]:
        """
        Maps an asynchronous function across the fleet.

        Executes the provided coroutine concurrently across devices in the
        cache using strict fault isolation. Failures on individual devices do
        not interrupt the execution of the remaining fleet.

        Args:
            func (Callable[[DeviceClient], Coroutine[Any, Any, T]]): The async
                configuration function to execute. Receives a connected
                DeviceClient as its primary argument.
            fleet_filter (Optional[Dict[str, Any]]): Dictionary to restrict execution.
                Supported keys: 'macs' (List[str]) to target specific devices.

        Returns:
            Dict[str, BulkResult[T]]: A mapping of MAC addresses to their
                respective execution outcomes.
        """

        async def _execute_single(mac_address: str) -> tuple[str, BulkResult[T]]:
            try:
                async with await self.connect_device(mac_address) as device:
                    result = await func(device)
                    return mac_address, BulkResult[T](success=True, result=result)
            except Exception as e:
                return mac_address, BulkResult[T](success=False, error=str(e))

        # Extract all cached MACs
        macs = [d.id.root if hasattr(d.id, "root") else d.id for d in self.cache._state.devices]

        # Apply specific target filters if provided by an agent
        if fleet_filter and "macs" in fleet_filter:
            target_macs = [m.upper() for m in fleet_filter["macs"]]
            macs = [m for m in macs if m.upper() in target_macs]

        tasks = [_execute_single(mac) for mac in macs]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        return {mac: res for mac, res in results}

    async def __aiter__(self) -> AsyncIterator[DeviceClient]:
        """
        Enables asynchronous iteration over the cached fleet devices.

        Yields:
            DeviceClient: A connected DeviceClient for each device in the fleet.
        """
        macs = [d.id.root if hasattr(d.id, "root") else d.id for d in self.cache._state.devices]
        for mac in macs:
            async with await self.connect_device(mac) as client:
                yield client

    async def __aenter__(self) -> "HubClient":
        """
        Enables asynchronous context management and triggers initial sync.

        Returns:
            HubClient: The initialized and synchronized client.
        """
        await self._http_client.__aenter__()
        await self.cache.load_from_disk()
        await self.cache.sync()
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        """
        Ensures graceful teardown of the HTTP connection pool.
        """
        await self._http_client.__aexit__(exc_type, exc_val, exc_tb)

    async def aclose(self) -> None:
        """
        Manually releases all underlying network resources.
        """
        await self._http_client.aclose()
Methods:
__aenter__() async

Enables asynchronous context management and triggers initial sync.

Returns:

Name Type Description
HubClient HubClient

The initialized and synchronized client.
Source code in src/xovis/api/hub/client.py 
200
201
202
203
204
205
206
207
208
209
210
async def __aenter__(self) -> "HubClient":
    """
    Enables asynchronous context management and triggers initial sync.

    Returns:
        HubClient: The initialized and synchronized client.
    """
    await self._http_client.__aenter__()
    await self.cache.load_from_disk()
    await self.cache.sync()
    return self
__aexit__(exc_type, exc_val, exc_tb) async

Ensures graceful teardown of the HTTP connection pool.

Source code in src/xovis/api/hub/client.py 
212
213
214
215
216
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
    """
    Ensures graceful teardown of the HTTP connection pool.
    """
    await self._http_client.__aexit__(exc_type, exc_val, exc_tb)
__aiter__() async

Enables asynchronous iteration over the cached fleet devices.

Yields:

Name Type Description
DeviceClient AsyncIterator[DeviceClient]

A connected DeviceClient for each device in the fleet.
Source code in src/xovis/api/hub/client.py 
188
189
190
191
192
193
194
195
196
197
198
async def __aiter__(self) -> AsyncIterator[DeviceClient]:
    """
    Enables asynchronous iteration over the cached fleet devices.

    Yields:
        DeviceClient: A connected DeviceClient for each device in the fleet.
    """
    macs = [d.id.root if hasattr(d.id, "root") else d.id for d in self.cache._state.devices]
    for mac in macs:
        async with await self.connect_device(mac) as client:
            yield client
__init__(client_id=None, client_secret=None, token=None, base_url='https://api.xovis.cloud', token_url='https://login.xovis.cloud/oauth/token', tunnel_base_url=None, timeout=15.0, max_retries=5, fleet_filter=None, **kwargs)

Initializes the HubClient and mounts architectural pillars.

Parameters:

Name Type Description Default
client_id Optional[str]

The OAuth2 Client ID provided by Xovis. If not provided, resolves from XOVIS_HUB_CLIENT_ID env var.

 None
client_secret Optional[str]

The OAuth2 Client Secret provided by Xovis. If not provided, resolves from XOVIS_HUB_CLIENT_SECRET env var.

 None
token Optional[str]

Optional static API token.

 None
base_url str

The base URL for the Xovis HUB Cloud API. Defaults to "https://api.xovis.cloud".

 'https://api.xovis.cloud'
token_url str

The Auth0 token endpoint. Defaults to "https://login.xovis.cloud/oauth/token".

 'https://login.xovis.cloud/oauth/token'
tunnel_base_url Optional[str]

Optional override for the device tunnel base URL. If not provided, the base_url is used.

 None
timeout float

Default timeout for HTTP operations. Defaults to 15.0.

 15.0
max_retries int

Maximum retry attempts for resilient networking. Defaults to 5.

 5
fleet_filter Optional[Dict[str, Any]]

Client-side filter to restrict the visible fleet scope.

 None
**kwargs Any

Additional configuration for the HTTP engine.

 {}
Source code in src/xovis/api/hub/client.py 
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
94
95
def __init__(
    self,
    client_id: Optional[str] = None,
    client_secret: Optional[str] = None,
    token: Optional[str] = None,
    base_url: str = "https://api.xovis.cloud",
    token_url: str = "https://login.xovis.cloud/oauth/token",
    tunnel_base_url: Optional[str] = None,
    timeout: float = 15.0,
    max_retries: int = 5,
    fleet_filter: Optional[dict[str, Any]] = None,
    **kwargs: Any,
) -> None:
    """
    Initializes the HubClient and mounts architectural pillars.

    Args:
        client_id (Optional[str], optional): The OAuth2 Client ID provided by Xovis.
            If not provided, resolves from XOVIS_HUB_CLIENT_ID env var.
        client_secret (Optional[str], optional): The OAuth2 Client Secret provided by Xovis.
            If not provided, resolves from XOVIS_HUB_CLIENT_SECRET env var.
        token (Optional[str], optional): Optional static API token.
        base_url (str, optional): The base URL for the Xovis HUB Cloud API.
            Defaults to "https://api.xovis.cloud".
        token_url (str, optional): The Auth0 token endpoint.
            Defaults to "https://login.xovis.cloud/oauth/token".
        tunnel_base_url (Optional[str], optional): Optional override for the
            device tunnel base URL. If not provided, the `base_url` is used.
        timeout (float, optional): Default timeout for HTTP operations.
            Defaults to 15.0.
        max_retries (int, optional): Maximum retry attempts for resilient
            networking. Defaults to 5.
        fleet_filter (Optional[Dict[str, Any]], optional): Client-side filter
            to restrict the visible fleet scope.
        **kwargs (Any): Additional configuration for the HTTP engine.
    """
    import os

    cid = client_id or os.getenv("XOVIS_HUB_CLIENT_ID")
    csec = client_secret or os.getenv("XOVIS_HUB_CLIENT_SECRET")
    static_token = token or os.getenv("XOVIS_HUB_TOKEN")

    if not static_token and (not cid or not csec):
        raise ValueError(
            "Missing Xovis HUB credentials. Provide token, or client_id/client_secret "
            "or set XOVIS_HUB_TOKEN or XOVIS_HUB_CLIENT_ID/XOVIS_HUB_CLIENT_SECRET environment variables."
        )

    self._auth = HubAuth(client_id=cid, client_secret=csec, token_url=token_url, token=static_token)
    self._tunnel_base_url = tunnel_base_url

    # Extract auto_persist_path if provided
    auto_persist_path = kwargs.pop("auto_persist_path", None)

    self._http_client = XovisHTTPClient(base_url=base_url, auth=self._auth, timeout=timeout, max_retries=max_retries, **kwargs)

    self.cache = HubCacheManager(self._http_client, fleet_filter=fleet_filter, auto_persist_path=auto_persist_path)

    self.devices = HubDevicesManager(self._http_client, cache=self.cache)
    self.licenses = HubLicensesManager(self._http_client, cache=self.cache)
aclose() async

Manually releases all underlying network resources.

Source code in src/xovis/api/hub/client.py 
218
219
220
221
222
async def aclose(self) -> None:
    """
    Manually releases all underlying network resources.
    """
    await self._http_client.aclose()
bulk_execute(func, fleet_filter=None) async

Maps an asynchronous function across the fleet.

Executes the provided coroutine concurrently across devices in the cache using strict fault isolation. Failures on individual devices do not interrupt the execution of the remaining fleet.

Parameters:

Name Type Description Default
func Callable[[DeviceClient], Coroutine[Any, Any, T]]

The async configuration function to execute. Receives a connected DeviceClient as its primary argument.

 required
fleet_filter Optional[Dict[str, Any]]

Dictionary to restrict execution. Supported keys: 'macs' (List[str]) to target specific devices.

 None

Returns:

Type Description
dict[str, BulkResult[T]]

Dict[str, BulkResult[T]]: A mapping of MAC addresses to their respective execution outcomes.
Source code in src/xovis/api/hub/client.py 
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
async def bulk_execute(
    self,
    func: Callable[[DeviceClient], Coroutine[Any, Any, T]],
    fleet_filter: Optional[dict[str, Any]] = None,
) -> dict[str, BulkResult[T]]:
    """
    Maps an asynchronous function across the fleet.

    Executes the provided coroutine concurrently across devices in the
    cache using strict fault isolation. Failures on individual devices do
    not interrupt the execution of the remaining fleet.

    Args:
        func (Callable[[DeviceClient], Coroutine[Any, Any, T]]): The async
            configuration function to execute. Receives a connected
            DeviceClient as its primary argument.
        fleet_filter (Optional[Dict[str, Any]]): Dictionary to restrict execution.
            Supported keys: 'macs' (List[str]) to target specific devices.

    Returns:
        Dict[str, BulkResult[T]]: A mapping of MAC addresses to their
            respective execution outcomes.
    """

    async def _execute_single(mac_address: str) -> tuple[str, BulkResult[T]]:
        try:
            async with await self.connect_device(mac_address) as device:
                result = await func(device)
                return mac_address, BulkResult[T](success=True, result=result)
        except Exception as e:
            return mac_address, BulkResult[T](success=False, error=str(e))

    # Extract all cached MACs
    macs = [d.id.root if hasattr(d.id, "root") else d.id for d in self.cache._state.devices]

    # Apply specific target filters if provided by an agent
    if fleet_filter and "macs" in fleet_filter:
        target_macs = [m.upper() for m in fleet_filter["macs"]]
        macs = [m for m in macs if m.upper() in target_macs]

    tasks = [_execute_single(mac) for mac in macs]

    results = await asyncio.gather(*tasks, return_exceptions=True)

    return {mac: res for mac, res in results}
connect_device(id_or_name) async

Spawns a DeviceClient routed through the Hub's dynamic secure tunnel.

Enables seamless transition from fleet-level management to specific edge sensor configuration. Intercepts OAuth2 tokens to authenticate the proxied connection.

The tunnel URL is dynamically constructed per the OpenAPI specification: {base_url}/devices/{mac_address}/tunnel. The DeviceClient will then append its own paths (e.g., /api/v5/...).

CRITICAL: The returned client MUST be used as an asynchronous context manager to prevent connection pooling leaks.

Parameters:

Name Type Description Default
id_or_name str

The MAC address (ID) or human-readable name of the target device.

 required

Returns:

Name Type Description
DeviceClient DeviceClient

A fully hydrated client instance routed via the HUB.
Source code in src/xovis/api/hub/client.py 
 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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
async def connect_device(self, id_or_name: str) -> DeviceClient:
    """
    Spawns a DeviceClient routed through the Hub's dynamic secure tunnel.

    Enables seamless transition from fleet-level management to specific
    edge sensor configuration. Intercepts OAuth2 tokens to authenticate
    the proxied connection.

    The tunnel URL is dynamically constructed per the OpenAPI specification:
    `{base_url}/devices/{mac_address}/tunnel`. The DeviceClient will then
    append its own paths (e.g., `/api/v5/...`).

    CRITICAL: The returned client MUST be used as an asynchronous context
    manager to prevent connection pooling leaks.

    Args:
        id_or_name (str): The MAC address (ID) or human-readable name of
            the target device.

    Returns:
        DeviceClient: A fully hydrated client instance routed via the HUB.
    """
    # Resolves the human-readable name to a MAC address using the local cache
    mac_address = self.devices._resolve_mac_address(id_or_name)

    # Dynamic OpenAPI routing
    base = self._tunnel_base_url or self._http_client.base_url
    tunnel_url = f"{base}/devices/{mac_address}/tunnel"

    client = DeviceClient(
        host=tunnel_url,
        username="hub_tunnel",
        password="hub_tunnel",
        timeout=60.0,  # Explicitly increased for heavy historical aggregations
        max_retries=self._http_client.max_retries,
        limits=httpx.Limits(max_connections=2, max_keepalive_connections=1),
    )

    # Inject Hub authentication for the proxied requests
    client._auth = self._auth
    client._http_client.auth = self._auth
    client._http_client.client.auth = self._auth

    return client

xovis.api.hub.cache

Xovis SDK - Hub Configuration Cache Manager

This module resides within the State & Topology Plane, providing a client-side cache for the Xovis HUB Cloud fleet. It manages device and license state synchronization, implementing robust client-side filtering and dot-notation accessors for interactive environments.

Classes

HubCacheManager

Client-side cache manager for the Xovis HUB Cloud fleet.

Coordinates the synchronization of fleet-wide metadata (devices, licenses) from the HUB Cloud API. Supports high-performance client-side filtering via fleet_filter and provides dot-notation accessors for rapid discovery in REPL environments.

Source code in src/xovis/api/hub/cache.py 
 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
122
123
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
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
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
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
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
class HubCacheManager:
    """
    Client-side cache manager for the Xovis HUB Cloud fleet.

    Coordinates the synchronization of fleet-wide metadata (devices, licenses)
    from the HUB Cloud API. Supports high-performance client-side filtering
    via `fleet_filter` and provides dot-notation accessors for rapid
    discovery in REPL environments.
    """

    def __init__(self, http_client: XovisHTTPClient, fleet_filter: Optional[dict[str, Any]] = None, auto_persist_path: Optional[str] = None) -> None:
        """
        Initializes the HubCacheManager.

        Args:
            http_client (XovisHTTPClient): The resilient Hub API engine.
            fleet_filter (Optional[Dict[str, Any]], optional): A dictionary of
                attributes and values to filter the fleet by (e.g., {"group": "EMEA"}).
                Supports list-based "any-of" matching.
            auto_persist_path (Optional[str], optional): Custom path for cache persistence.
        """
        self._http = http_client
        self.fleet_filter = fleet_filter or {}
        self.auto_persist_path = auto_persist_path
        self._memory_only = False
        self._state = HubStateBucket()

    @property
    def devices(self) -> REPLAccessor[Device]:
        """
        Accessor for devices mapped by their human-readable name.

        Returns:
            REPLAccessor[Device]: A dynamic accessor for name-based device discovery.
        """
        return REPLAccessor(self._state.devices, key_attr="device_name")

    @property
    def devices_by_mac(self) -> REPLAccessor[Device]:
        """
        Accessor for devices mapped by their unique MAC address.

        Returns:
            REPLAccessor[Device]: A dynamic accessor for MAC-based device discovery.
        """

        class DeviceMacWrapper:
            """Internal wrapper to flatten device IDs for REPL access."""

            def __init__(self, device: Device):
                """
                Initializes the MAC wrapper.

                Args:
                    device (Device): The Hub device model to wrap.
                """
                self._device = device
                self.id = device.id.root if hasattr(device.id, "root") else device.id
                self.device_name = device.device_name

            def __getattr__(self, name):
                return getattr(self._device, name)

        wrappers = [DeviceMacWrapper(d) for d in self._state.devices if d.id]
        return REPLAccessor(wrappers, key_attr="id")

    @property
    def licenses(self) -> REPLAccessor[LicenseStatus]:
        """
        Accessor for licenses mapped by device ID.

        Returns:
            REPLAccessor[LicenseStatus]: A dynamic accessor for license status.
        """
        return REPLAccessor(self._state.licenses, key_attr="device_id")

    async def sync(self, preserve_topology: bool = True) -> None:
        """
        Synchronizes the Hub fleet state with the Cloud API.

        Fetches all devices and licenses, applies client-side filtering based on
        the `fleet_filter` configuration, and populates the internal state bucket.
        Warns if the filter is overly restrictive (dropping >90% of devices).

        Args:
            preserve_topology (bool, optional): If True, retains any topological
                roles and parents discovered during the current session's Deep Dive.
                Defaults to True.

        Raises:
            httpx.HTTPError: If the Hub API is unreachable or returns an error.
        """

        # Save volatile state if requested
        saved_roles = self._state.topology_roles.copy() if preserve_topology else {}
        saved_parents = self._state.topology_parents.copy() if preserve_topology else {}
        try:
            # We try the standard path from documentation first
            # The OpenAPI spec indicates that /devices requires a 'state' parameter or 'customer'
            devices_res = await self._http.get("/devices", params={"state": "MANAGED"})
            if devices_res.status_code == 400:
                # If MANAGED fails, try UNMANAGED as fallback
                devices_res = await self._http.get("/devices", params={"state": "UNMANAGED"})

            devices_res.raise_for_status()
            devices_data = DevicesResponse.model_validate(devices_res.json())
        except Exception as e:
            logging.debug(f"Failed to fetch devices from primary /devices path: {e}")
            # Fallback to the device-management prefixed path
            try:
                devices_res = await self._http.get("/device-management/api/public/v1/devices")
                logging.debug(f"Fallback 1 response: {devices_res.status_code}")
                devices_res.raise_for_status()
                devices_data = DevicesResponse.model_validate(devices_res.json())
            except Exception as e1:
                logging.debug(f"Failed fallback 1: {e1}")
                # Try with the ONLINE status filter
                try:
                    devices_res = await self._http.get(
                        "/device-management/api/public/v1/devices",
                        params={"deviceStatus": "ONLINE"},
                    )
                    logging.debug(f"Fallback 2 response: {devices_res.status_code}")
                    devices_res.raise_for_status()
                    devices_data = DevicesResponse.model_validate(devices_res.json())
                except Exception as e2:
                    logging.debug(f"Failed all device fetch paths: {e2}")
                    devices_data = DevicesResponse(items=[])

        raw_count = len(devices_data.items or [])
        logging.info(f"Hub Sync: Fetched {raw_count} devices from cloud.")
        filtered_devices = []
        if devices_data.items:
            for device in devices_data.items:
                # DEBUG: Print device structure if needed
                # logging.debug(f"Syncing device: {device}")
                if self._matches_filter(device):
                    filtered_devices.append(device)

        logging.info(f"Hub Sync: {len(filtered_devices)} devices remaining after fleet_filter.")
        self._state.devices = filtered_devices

        # ARCHITECTURAL FIX: Enforce uppercase MAC normalization for all Hub devices
        for d in self._state.devices:
            if d.id:
                mac = d.id.root if hasattr(d.id, "root") else str(d.id)
                if hasattr(d.id, "root"):
                    d.id.root = mac.upper()
                else:
                    d.id = mac.upper()

        if raw_count > 0:
            dropped_ratio = (raw_count - len(filtered_devices)) / raw_count
            if dropped_ratio > 0.9 and len(filtered_devices) == 0:
                logging.error(f"Hub fleet_filter dropped ALL fetched devices ({raw_count}). Filter criteria: {self.fleet_filter}")
            elif dropped_ratio > 0.9:
                logging.warning(
                    f"Hub fleet_filter dropped {dropped_ratio:.1%} of fetched devices ({len(filtered_devices)}/{raw_count}). "
                    "Consider refining the filter to improve efficiency."
                )

        try:
            licenses_res = await self._http.get("/license/api/public/v1/licenses/status")
            licenses_res.raise_for_status()
            licenses_data = LicenseStatusResponse.model_validate(licenses_res.json())
        except Exception:
            try:
                licenses_res = await self._http.get("/api/public/v1/licenses/status")
                licenses_res.raise_for_status()
                licenses_data = LicenseStatusResponse.model_validate(licenses_res.json())
            except Exception:
                try:
                    licenses_res = await self._http.get("/licenses/status")
                    licenses_res.raise_for_status()
                    licenses_data = LicenseStatusResponse.model_validate(licenses_res.json())
                except Exception:
                    # If licenses fail, we continue with empty licenses but valid devices
                    licenses_data = LicenseStatusResponse(license_status_list=[])

        # Filter devices for selected customer and group
        device_macs = set()
        for d in filtered_devices:
            if not d.id:
                continue
            mac = d.id.root if hasattr(d.id, "root") else str(d.id)
            device_macs.add(mac)

        filtered_licenses = []
        if hasattr(licenses_data, "license_status_list") and licenses_data.license_status_list:
            for lic in licenses_data.license_status_list:
                mac = lic.device_id.root if hasattr(lic.device_id, "root") else str(lic.device_id)
                if mac in device_macs:
                    filtered_licenses.append(lic)

        self._state.licenses = filtered_licenses

        # Restore volatile topological state
        if preserve_topology:
            self._state.topology_roles.update(saved_roles)
            self._state.topology_parents.update(saved_parents)

        await self.save_to_disk()

    def export_to_file(self, file_path: str, custom_bucket: Optional[HubStateBucket] = None) -> None:
        """
        Writes the current hub state or a custom bucket to an offline JSON file.

        Args:
            file_path (str): The target file path for the JSON workspace.
            custom_bucket (Optional[HubStateBucket], optional): A filtered bucket to
                export instead of the full internal state. Defaults to None.
        """
        bucket = custom_bucket or self._state
        with open(file_path, "w", encoding="utf-8") as f:
            f.write(bucket.model_dump_json(by_alias=True, exclude_none=True))

    def load_from_file(self, file_path: str, merge: bool = False) -> None:
        """
        Reads a JSON workspace file and replaces or merges the internal state.

        Args:
            file_path (str): The source file path to load from.
            merge (bool, optional): If True, merges the loaded state into the
                current state instead of replacing it. Defaults to False.
        """
        with open(file_path, encoding="utf-8") as f:
            data = f.read()
            new_state = HubStateBucket.model_validate_json(data)

            if not merge:
                self._state = new_state
            else:
                # Merge devices (deduplicate by ID)
                existing_macs = {(d.id.root if hasattr(d.id, "root") else d.id).upper() for d in self._state.devices if d.id}
                for d in new_state.devices:
                    mac = (d.id.root if hasattr(d.id, "root") else d.id).upper() if d.id else None
                    if mac and mac not in existing_macs:
                        self._state.devices.append(d)
                        existing_macs.add(mac)

                # Merge topology
                self._state.topology_roles.update(new_state.topology_roles)
                for k, v in new_state.topology_parents.items():
                    if k in self._state.topology_parents:
                        for p in v:
                            if p not in self._state.topology_parents[k]:
                                self._state.topology_parents[k].append(p)
                    else:
                        self._state.topology_parents[k] = v

                # Merge licenses (if any)
                existing_lic_macs = {(l.device_id.root if hasattr(l.device_id, "root") else l.device_id).upper() for l in self._state.licenses}
                for l in new_state.licenses:
                    mac = (l.device_id.root if hasattr(l.device_id, "root") else l.device_id).upper()
                    if mac not in existing_lic_macs:
                        self._state.licenses.append(l)
                        existing_lic_macs.add(mac)

    def _ensure_directory_or_fallback(self, path: Path) -> Optional[Path]:
        """Ensures the directory for the given path is writeable or falls back.

        Implements the 3-Tier cache folder creation and fallback strategy.

        Args:
            path (Path): The desired target file path.

        Returns:
            Optional[Path]: The writeable target path, or None if falling back
                to memory-only.
        """
        if getattr(self, "_memory_only", False):
            return None

        parent = path.parent
        try:
            parent.mkdir(parents=True, exist_ok=True)
            return path
        except (PermissionError, OSError) as exc:
            logging.warning(f"Local directory creation failed at {parent} ({exc}). Attempting global system cache fallback.")

        try:
            from xovis.api.device.cache import CachePaths

            try:
                rel_parts = path.relative_to(CachePaths.BASE_DIR)
                sys_target = CachePaths.get_system_cache_dir() / rel_parts
            except ValueError:
                sys_target = CachePaths.get_system_cache_dir() / path.name

            sys_parent = sys_target.parent
            sys_parent.mkdir(parents=True, exist_ok=True)
            return sys_target
        except (PermissionError, OSError) as exc:
            logging.warning(
                f"Unable to write to system-level cache workspace ({exc}). "
                f"Falling back to temporary memory-only caching. "
                f"To persist cache, ensure write permissions exist."
            )

        self._memory_only = True
        return None

    def _resolve_persist_path(self) -> Optional[str]:
        """Resolves the final persistence path for the Hub fleet.

        Returns:
            Optional[str]: The absolute path to the state file, or None.
        """
        if getattr(self, "_memory_only", False):
            return None

        from pathlib import Path

        from xovis.api.device.cache import CachePaths

        resolved_path: Optional[Path] = None

        if self.auto_persist_path:
            resolved_path = Path(self.auto_persist_path)
        else:
            resolved_path = CachePaths.FLEET_STATE

        if resolved_path:
            final_path = self._ensure_directory_or_fallback(resolved_path)
            return str(final_path) if final_path else None

        return None

    async def save_to_disk(self) -> None:
        """Safely serializes the HubStateBucket to disk."""
        path = self._resolve_persist_path()
        if not path:
            return

        import asyncio

        def _save():
            with open(path, "w", encoding="utf-8") as f:
                f.write(self._state.model_dump_json(by_alias=True, exclude_none=True, indent=2))

        await asyncio.to_thread(_save)

    async def load_from_disk(self) -> None:
        """Deserializes the HubStateBucket from disk."""
        path = self._resolve_persist_path()
        if not path:
            return

        import asyncio
        import os

        if not await asyncio.to_thread(os.path.exists, path):
            return

        def _load():
            with open(path, encoding="utf-8") as f:
                return f.read()

        data = await asyncio.to_thread(_load)
        try:
            self._state = HubStateBucket.model_validate_json(data)
        except Exception as e:
            logging.warning(f"Could not load Hub cache from disk: {e}")

    def _matches_filter(self, device: Device) -> bool:
        """
        Evaluates if a device matches the configured fleet filter.

        Args:
            device (Device): The device model to evaluate.

        Returns:
            bool: True if the device matches all filter criteria, False otherwise.
        """
        for key, value in self.fleet_filter.items():
            device_val = getattr(device, key, None)

            if isinstance(value, list):
                if isinstance(device_val, list):
                    if not any(item in value for item in device_val):
                        return False
                elif device_val not in value:
                    return False
            elif device_val != value:
                return False
        return True
Attributes
devices property

Accessor for devices mapped by their human-readable name.

Returns:

Type Description
REPLAccessor[Device]

REPLAccessor[Device]: A dynamic accessor for name-based device discovery.
devices_by_mac property

Accessor for devices mapped by their unique MAC address.

Returns:

Type Description
REPLAccessor[Device]

REPLAccessor[Device]: A dynamic accessor for MAC-based device discovery.
licenses property

Accessor for licenses mapped by device ID.

Returns:

Type Description
REPLAccessor[LicenseStatus]

REPLAccessor[LicenseStatus]: A dynamic accessor for license status.
Methods:
__init__(http_client, fleet_filter=None, auto_persist_path=None)

Initializes the HubCacheManager.

Parameters:

Name Type Description Default
http_client XovisHTTPClient

The resilient Hub API engine.

 required
fleet_filter Optional[Dict[str, Any]]

A dictionary of attributes and values to filter the fleet by (e.g., {"group": "EMEA"}). Supports list-based "any-of" matching.

 None
auto_persist_path Optional[str]

Custom path for cache persistence.

 None
Source code in src/xovis/api/hub/cache.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(self, http_client: XovisHTTPClient, fleet_filter: Optional[dict[str, Any]] = None, auto_persist_path: Optional[str] = None) -> None:
    """
    Initializes the HubCacheManager.

    Args:
        http_client (XovisHTTPClient): The resilient Hub API engine.
        fleet_filter (Optional[Dict[str, Any]], optional): A dictionary of
            attributes and values to filter the fleet by (e.g., {"group": "EMEA"}).
            Supports list-based "any-of" matching.
        auto_persist_path (Optional[str], optional): Custom path for cache persistence.
    """
    self._http = http_client
    self.fleet_filter = fleet_filter or {}
    self.auto_persist_path = auto_persist_path
    self._memory_only = False
    self._state = HubStateBucket()
_ensure_directory_or_fallback(path)

Ensures the directory for the given path is writeable or falls back.

Implements the 3-Tier cache folder creation and fallback strategy.

Parameters:

Name Type Description Default
path Path

The desired target file path.

 required

Returns:

Type Description
Optional[Path]

Optional[Path]: The writeable target path, or None if falling back to memory-only.
Source code in src/xovis/api/hub/cache.py 
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
def _ensure_directory_or_fallback(self, path: Path) -> Optional[Path]:
    """Ensures the directory for the given path is writeable or falls back.

    Implements the 3-Tier cache folder creation and fallback strategy.

    Args:
        path (Path): The desired target file path.

    Returns:
        Optional[Path]: The writeable target path, or None if falling back
            to memory-only.
    """
    if getattr(self, "_memory_only", False):
        return None

    parent = path.parent
    try:
        parent.mkdir(parents=True, exist_ok=True)
        return path
    except (PermissionError, OSError) as exc:
        logging.warning(f"Local directory creation failed at {parent} ({exc}). Attempting global system cache fallback.")

    try:
        from xovis.api.device.cache import CachePaths

        try:
            rel_parts = path.relative_to(CachePaths.BASE_DIR)
            sys_target = CachePaths.get_system_cache_dir() / rel_parts
        except ValueError:
            sys_target = CachePaths.get_system_cache_dir() / path.name

        sys_parent = sys_target.parent
        sys_parent.mkdir(parents=True, exist_ok=True)
        return sys_target
    except (PermissionError, OSError) as exc:
        logging.warning(
            f"Unable to write to system-level cache workspace ({exc}). "
            f"Falling back to temporary memory-only caching. "
            f"To persist cache, ensure write permissions exist."
        )

    self._memory_only = True
    return None
_matches_filter(device)

Evaluates if a device matches the configured fleet filter.

Parameters:

Name Type Description Default
device Device

The device model to evaluate.

 required

Returns:

Name Type Description
bool bool

True if the device matches all filter criteria, False otherwise.
Source code in src/xovis/api/hub/cache.py 
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
def _matches_filter(self, device: Device) -> bool:
    """
    Evaluates if a device matches the configured fleet filter.

    Args:
        device (Device): The device model to evaluate.

    Returns:
        bool: True if the device matches all filter criteria, False otherwise.
    """
    for key, value in self.fleet_filter.items():
        device_val = getattr(device, key, None)

        if isinstance(value, list):
            if isinstance(device_val, list):
                if not any(item in value for item in device_val):
                    return False
            elif device_val not in value:
                return False
        elif device_val != value:
            return False
    return True
_resolve_persist_path()

Resolves the final persistence path for the Hub fleet.

Returns:

Type Description
Optional[str]

Optional[str]: The absolute path to the state file, or None.
Source code in src/xovis/api/hub/cache.py 
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
def _resolve_persist_path(self) -> Optional[str]:
    """Resolves the final persistence path for the Hub fleet.

    Returns:
        Optional[str]: The absolute path to the state file, or None.
    """
    if getattr(self, "_memory_only", False):
        return None

    from pathlib import Path

    from xovis.api.device.cache import CachePaths

    resolved_path: Optional[Path] = None

    if self.auto_persist_path:
        resolved_path = Path(self.auto_persist_path)
    else:
        resolved_path = CachePaths.FLEET_STATE

    if resolved_path:
        final_path = self._ensure_directory_or_fallback(resolved_path)
        return str(final_path) if final_path else None

    return None
export_to_file(file_path, custom_bucket=None)

Writes the current hub state or a custom bucket to an offline JSON file.

Parameters:

Name Type Description Default
file_path str

The target file path for the JSON workspace.

 required
custom_bucket Optional[HubStateBucket]

A filtered bucket to export instead of the full internal state. Defaults to None.

 None
Source code in src/xovis/api/hub/cache.py 
285
286
287
288
289
290
291
292
293
294
295
296
def export_to_file(self, file_path: str, custom_bucket: Optional[HubStateBucket] = None) -> None:
    """
    Writes the current hub state or a custom bucket to an offline JSON file.

    Args:
        file_path (str): The target file path for the JSON workspace.
        custom_bucket (Optional[HubStateBucket], optional): A filtered bucket to
            export instead of the full internal state. Defaults to None.
    """
    bucket = custom_bucket or self._state
    with open(file_path, "w", encoding="utf-8") as f:
        f.write(bucket.model_dump_json(by_alias=True, exclude_none=True))
load_from_disk() async

Deserializes the HubStateBucket from disk.

Source code in src/xovis/api/hub/cache.py 
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
async def load_from_disk(self) -> None:
    """Deserializes the HubStateBucket from disk."""
    path = self._resolve_persist_path()
    if not path:
        return

    import asyncio
    import os

    if not await asyncio.to_thread(os.path.exists, path):
        return

    def _load():
        with open(path, encoding="utf-8") as f:
            return f.read()

    data = await asyncio.to_thread(_load)
    try:
        self._state = HubStateBucket.model_validate_json(data)
    except Exception as e:
        logging.warning(f"Could not load Hub cache from disk: {e}")
load_from_file(file_path, merge=False)

Reads a JSON workspace file and replaces or merges the internal state.

Parameters:

Name Type Description Default
file_path str

The source file path to load from.

 required
merge bool

If True, merges the loaded state into the current state instead of replacing it. Defaults to False.

 False
Source code in src/xovis/api/hub/cache.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
def load_from_file(self, file_path: str, merge: bool = False) -> None:
    """
    Reads a JSON workspace file and replaces or merges the internal state.

    Args:
        file_path (str): The source file path to load from.
        merge (bool, optional): If True, merges the loaded state into the
            current state instead of replacing it. Defaults to False.
    """
    with open(file_path, encoding="utf-8") as f:
        data = f.read()
        new_state = HubStateBucket.model_validate_json(data)

        if not merge:
            self._state = new_state
        else:
            # Merge devices (deduplicate by ID)
            existing_macs = {(d.id.root if hasattr(d.id, "root") else d.id).upper() for d in self._state.devices if d.id}
            for d in new_state.devices:
                mac = (d.id.root if hasattr(d.id, "root") else d.id).upper() if d.id else None
                if mac and mac not in existing_macs:
                    self._state.devices.append(d)
                    existing_macs.add(mac)

            # Merge topology
            self._state.topology_roles.update(new_state.topology_roles)
            for k, v in new_state.topology_parents.items():
                if k in self._state.topology_parents:
                    for p in v:
                        if p not in self._state.topology_parents[k]:
                            self._state.topology_parents[k].append(p)
                else:
                    self._state.topology_parents[k] = v

            # Merge licenses (if any)
            existing_lic_macs = {(l.device_id.root if hasattr(l.device_id, "root") else l.device_id).upper() for l in self._state.licenses}
            for l in new_state.licenses:
                mac = (l.device_id.root if hasattr(l.device_id, "root") else l.device_id).upper()
                if mac not in existing_lic_macs:
                    self._state.licenses.append(l)
                    existing_lic_macs.add(mac)
save_to_disk() async

Safely serializes the HubStateBucket to disk.

Source code in src/xovis/api/hub/cache.py 
410
411
412
413
414
415
416
417
418
419
420
421
422
async def save_to_disk(self) -> None:
    """Safely serializes the HubStateBucket to disk."""
    path = self._resolve_persist_path()
    if not path:
        return

    import asyncio

    def _save():
        with open(path, "w", encoding="utf-8") as f:
            f.write(self._state.model_dump_json(by_alias=True, exclude_none=True, indent=2))

    await asyncio.to_thread(_save)
sync(preserve_topology=True) async

Synchronizes the Hub fleet state with the Cloud API.

Fetches all devices and licenses, applies client-side filtering based on the fleet_filter configuration, and populates the internal state bucket. Warns if the filter is overly restrictive (dropping >90% of devices).

Parameters:

Name Type Description Default
preserve_topology bool

If True, retains any topological roles and parents discovered during the current session's Deep Dive. Defaults to True.

 True

Raises:

Type Description
HTTPError

If the Hub API is unreachable or returns an error.
Source code in src/xovis/api/hub/cache.py 
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
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
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
async def sync(self, preserve_topology: bool = True) -> None:
    """
    Synchronizes the Hub fleet state with the Cloud API.

    Fetches all devices and licenses, applies client-side filtering based on
    the `fleet_filter` configuration, and populates the internal state bucket.
    Warns if the filter is overly restrictive (dropping >90% of devices).

    Args:
        preserve_topology (bool, optional): If True, retains any topological
            roles and parents discovered during the current session's Deep Dive.
            Defaults to True.

    Raises:
        httpx.HTTPError: If the Hub API is unreachable or returns an error.
    """

    # Save volatile state if requested
    saved_roles = self._state.topology_roles.copy() if preserve_topology else {}
    saved_parents = self._state.topology_parents.copy() if preserve_topology else {}
    try:
        # We try the standard path from documentation first
        # The OpenAPI spec indicates that /devices requires a 'state' parameter or 'customer'
        devices_res = await self._http.get("/devices", params={"state": "MANAGED"})
        if devices_res.status_code == 400:
            # If MANAGED fails, try UNMANAGED as fallback
            devices_res = await self._http.get("/devices", params={"state": "UNMANAGED"})

        devices_res.raise_for_status()
        devices_data = DevicesResponse.model_validate(devices_res.json())
    except Exception as e:
        logging.debug(f"Failed to fetch devices from primary /devices path: {e}")
        # Fallback to the device-management prefixed path
        try:
            devices_res = await self._http.get("/device-management/api/public/v1/devices")
            logging.debug(f"Fallback 1 response: {devices_res.status_code}")
            devices_res.raise_for_status()
            devices_data = DevicesResponse.model_validate(devices_res.json())
        except Exception as e1:
            logging.debug(f"Failed fallback 1: {e1}")
            # Try with the ONLINE status filter
            try:
                devices_res = await self._http.get(
                    "/device-management/api/public/v1/devices",
                    params={"deviceStatus": "ONLINE"},
                )
                logging.debug(f"Fallback 2 response: {devices_res.status_code}")
                devices_res.raise_for_status()
                devices_data = DevicesResponse.model_validate(devices_res.json())
            except Exception as e2:
                logging.debug(f"Failed all device fetch paths: {e2}")
                devices_data = DevicesResponse(items=[])

    raw_count = len(devices_data.items or [])
    logging.info(f"Hub Sync: Fetched {raw_count} devices from cloud.")
    filtered_devices = []
    if devices_data.items:
        for device in devices_data.items:
            # DEBUG: Print device structure if needed
            # logging.debug(f"Syncing device: {device}")
            if self._matches_filter(device):
                filtered_devices.append(device)

    logging.info(f"Hub Sync: {len(filtered_devices)} devices remaining after fleet_filter.")
    self._state.devices = filtered_devices

    # ARCHITECTURAL FIX: Enforce uppercase MAC normalization for all Hub devices
    for d in self._state.devices:
        if d.id:
            mac = d.id.root if hasattr(d.id, "root") else str(d.id)
            if hasattr(d.id, "root"):
                d.id.root = mac.upper()
            else:
                d.id = mac.upper()

    if raw_count > 0:
        dropped_ratio = (raw_count - len(filtered_devices)) / raw_count
        if dropped_ratio > 0.9 and len(filtered_devices) == 0:
            logging.error(f"Hub fleet_filter dropped ALL fetched devices ({raw_count}). Filter criteria: {self.fleet_filter}")
        elif dropped_ratio > 0.9:
            logging.warning(
                f"Hub fleet_filter dropped {dropped_ratio:.1%} of fetched devices ({len(filtered_devices)}/{raw_count}). "
                "Consider refining the filter to improve efficiency."
            )

    try:
        licenses_res = await self._http.get("/license/api/public/v1/licenses/status")
        licenses_res.raise_for_status()
        licenses_data = LicenseStatusResponse.model_validate(licenses_res.json())
    except Exception:
        try:
            licenses_res = await self._http.get("/api/public/v1/licenses/status")
            licenses_res.raise_for_status()
            licenses_data = LicenseStatusResponse.model_validate(licenses_res.json())
        except Exception:
            try:
                licenses_res = await self._http.get("/licenses/status")
                licenses_res.raise_for_status()
                licenses_data = LicenseStatusResponse.model_validate(licenses_res.json())
            except Exception:
                # If licenses fail, we continue with empty licenses but valid devices
                licenses_data = LicenseStatusResponse(license_status_list=[])

    # Filter devices for selected customer and group
    device_macs = set()
    for d in filtered_devices:
        if not d.id:
            continue
        mac = d.id.root if hasattr(d.id, "root") else str(d.id)
        device_macs.add(mac)

    filtered_licenses = []
    if hasattr(licenses_data, "license_status_list") and licenses_data.license_status_list:
        for lic in licenses_data.license_status_list:
            mac = lic.device_id.root if hasattr(lic.device_id, "root") else str(lic.device_id)
            if mac in device_macs:
                filtered_licenses.append(lic)

    self._state.licenses = filtered_licenses

    # Restore volatile topological state
    if preserve_topology:
        self._state.topology_roles.update(saved_roles)
        self._state.topology_parents.update(saved_parents)

    await self.save_to_disk()

HubStateBucket

Bases: BaseModel

Root state container for the Xovis HUB Cloud fleet.

Aggregates collections of devices and their corresponding license statuses fetched from the Hub API, enabling efficient client-side filtering and lookup.

Source code in src/xovis/api/hub/cache.py 
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
class HubStateBucket(BaseModel):
    """
    Root state container for the Xovis HUB Cloud fleet.

    Aggregates collections of devices and their corresponding license statuses
    fetched from the Hub API, enabling efficient client-side filtering and lookup.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True, extra="ignore")
    devices: list[Device] = Field(default_factory=list)
    licenses: list[LicenseStatus] = Field(default_factory=list)
    # Volatile topological state (Master/Child/Standalone mapping) discovered during Deep Dive.
    # Keyed by MAC address.
    topology_roles: dict[str, str] = Field(default_factory=dict)
    topology_parents: dict[str, list[str]] = Field(default_factory=dict)

    @field_validator("topology_roles", mode="before")
    @classmethod
    def normalize_roles_keys(cls, v: Any) -> Any:
        """
        Ensures all topology role keys are normalized to uppercase MAC addresses.

        Args:
            v (Any): Raw role mapping.

        Returns:
            Any: Normalized role dictionary.
        """
        if isinstance(v, dict):
            return {str(k).upper(): v2 for k, v2 in v.items()}
        if v is None:
            return {}
        return v

    @field_validator("topology_parents", mode="before")
    @classmethod
    def normalize_parents(cls, v: Any) -> Any:
        """
        Normalizes parent MAC address lists to consistent uppercase formats.

        Args:
            v (Any): Raw parent mapping.

        Returns:
            Any: Normalized parent dictionary.
        """
        if v is None:
            return {}
        if not isinstance(v, dict):
            return v
        normalized = {}
        for k, val in v.items():
            key = str(k).upper()
            if isinstance(val, list):
                normalized[key] = [str(i).upper() for i in val]
            else:
                normalized[key] = [str(val).upper()]
        return normalized
Methods:
normalize_parents(v) classmethod

Normalizes parent MAC address lists to consistent uppercase formats.

Parameters:

Name Type Description Default
v Any

Raw parent mapping.

 required

Returns:

Name Type Description
Any Any

Normalized parent dictionary.
Source code in src/xovis/api/hub/cache.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@field_validator("topology_parents", mode="before")
@classmethod
def normalize_parents(cls, v: Any) -> Any:
    """
    Normalizes parent MAC address lists to consistent uppercase formats.

    Args:
        v (Any): Raw parent mapping.

    Returns:
        Any: Normalized parent dictionary.
    """
    if v is None:
        return {}
    if not isinstance(v, dict):
        return v
    normalized = {}
    for k, val in v.items():
        key = str(k).upper()
        if isinstance(val, list):
            normalized[key] = [str(i).upper() for i in val]
        else:
            normalized[key] = [str(val).upper()]
    return normalized
normalize_roles_keys(v) classmethod

Ensures all topology role keys are normalized to uppercase MAC addresses.

Parameters:

Name Type Description Default
v Any

Raw role mapping.

 required

Returns:

Name Type Description
Any Any

Normalized role dictionary.
Source code in src/xovis/api/hub/cache.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@field_validator("topology_roles", mode="before")
@classmethod
def normalize_roles_keys(cls, v: Any) -> Any:
    """
    Ensures all topology role keys are normalized to uppercase MAC addresses.

    Args:
        v (Any): Raw role mapping.

    Returns:
        Any: Normalized role dictionary.
    """
    if isinstance(v, dict):
        return {str(k).upper(): v2 for k, v2 in v.items()}
    if v is None:
        return {}
    return v

Resources

xovis.api.hub.resources.hub_device

Xovis SDK - HUB Cloud Device Management Resource

Provides the implementation for managing fleet-wide device metadata, UI access, and categorical assignments on the Xovis HUB Cloud. Operates within the Control Plane and State & Topology Plane.

Classes

HubDevicesManager

Bases: HubResourceManager

Manages device operations on the Xovis HUB Cloud.

This manager orchestrates fleet-wide operations, including metadata synchronization and secure UI tunneling. It utilizes the HubCacheManager to resolve human-readable names to MAC addresses.

Source code in src/xovis/api/hub/resources/hub_device.py 
 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
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class HubDevicesManager(HubResourceManager):
    """
    Manages device operations on the Xovis HUB Cloud.

    This manager orchestrates fleet-wide operations, including metadata
    synchronization and secure UI tunneling. It utilizes the HubCacheManager
    to resolve human-readable names to MAC addresses.
    """

    def __init__(self, http_client: XovisHTTPClient, cache: Optional["HubCacheManager"] = None):
        """
        Initializes the HubDevicesManager.

        Args:
            http_client (XovisHTTPClient): The resilient HTTP client.
            cache (Optional[HubCacheManager]): The HUB-level cache manager.
        """
        super().__init__(http_client, cache)
        self._base_path = "/device-management/api/public/v1/devices"

    async def get_devices(self) -> DevicesResponse:
        """
        Retrieves all devices associated with the current HUB tenant.

        Returns:
            DevicesResponse: A collection of device metadata.
        """
        response = await self._http.get(self._base_path)
        return DevicesResponse.model_validate(response.json())

    async def get_device_ui_access(self, id_or_name: Union[str, list[str]]) -> DeviceUiAccess:
        """
        Generates secure, temporary UI access links for specific devices.

        Args:
            id_or_name (Union[str, List[str]]): Target device MAC(s) or name(s).

        Returns:
            DeviceUiAccess: Secure access URLs for the requested devices.
        """
        device_id = self._resolve_mac_address(id_or_name)
        response = await self._http.get(f"{self._base_path}/{device_id}/webui_link")
        return DeviceUiAccess.model_validate(response.json())

    async def assign_customer(self, id_or_name: Union[str, list[str]], customer_name: str) -> dict:
        """
        Assigns a customer identifier to a list of devices.

        Args:
            id_or_name (Union[str, List[str]]): Target device MAC(s) or name(s).
            customer_name (str): The customer name to assign.

        Returns:
            dict: The response from the HUB Cloud.
        """
        device_ids = self._resolve_multiple(id_or_name)
        payload = DevicesCustomerAssignment(device_ids=device_ids, customer_name=customer_name)
        response = await self._http.post(f"{self._base_path}/assign_customer", json=payload.model_dump(mode="json"))
        return response.json() if response.text else {}

    async def update_categories(
        self,
        id_or_name: Union[str, list[str]],
        categories_to_add: Optional[list[str]] = None,
        categories_to_remove: Optional[list[str]] = None,
    ) -> dict:
        """
        Modifies categorical tags for a list of devices.

        Args:
            id_or_name (Union[str, List[str]]): Target device MAC(s) or name(s).
            categories_to_add (Optional[List[str]]): Tags to append.
            categories_to_remove (Optional[List[str]]): Tags to strip.

        Returns:
            dict: The response from the HUB Cloud.
        """
        device_ids = self._resolve_multiple(id_or_name)
        payload = DevicesCategoriesAssignment(
            device_ids=device_ids,
            categories_to_add=categories_to_add,
            categories_to_remove=categories_to_remove,
        )
        response = await self._http.post(
            f"{self._base_path}/manage_categories",
            json=payload.model_dump(mode="json", exclude_unset=True),
        )
        return response.json() if response.text else {}
Methods:
__init__(http_client, cache=None)

Initializes the HubDevicesManager.

Parameters:

Name Type Description Default
http_client XovisHTTPClient

The resilient HTTP client.

 required
cache Optional[HubCacheManager]

The HUB-level cache manager.

 None
Source code in src/xovis/api/hub/resources/hub_device.py 
33
34
35
36
37
38
39
40
41
42
def __init__(self, http_client: XovisHTTPClient, cache: Optional["HubCacheManager"] = None):
    """
    Initializes the HubDevicesManager.

    Args:
        http_client (XovisHTTPClient): The resilient HTTP client.
        cache (Optional[HubCacheManager]): The HUB-level cache manager.
    """
    super().__init__(http_client, cache)
    self._base_path = "/device-management/api/public/v1/devices"
assign_customer(id_or_name, customer_name) async

Assigns a customer identifier to a list of devices.

Parameters:

Name Type Description Default
id_or_name Union[str, List[str]]

Target device MAC(s) or name(s).

 required
customer_name str

The customer name to assign.

 required

Returns:

Name Type Description
dict dict

The response from the HUB Cloud.
Source code in src/xovis/api/hub/resources/hub_device.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
async def assign_customer(self, id_or_name: Union[str, list[str]], customer_name: str) -> dict:
    """
    Assigns a customer identifier to a list of devices.

    Args:
        id_or_name (Union[str, List[str]]): Target device MAC(s) or name(s).
        customer_name (str): The customer name to assign.

    Returns:
        dict: The response from the HUB Cloud.
    """
    device_ids = self._resolve_multiple(id_or_name)
    payload = DevicesCustomerAssignment(device_ids=device_ids, customer_name=customer_name)
    response = await self._http.post(f"{self._base_path}/assign_customer", json=payload.model_dump(mode="json"))
    return response.json() if response.text else {}
get_device_ui_access(id_or_name) async

Generates secure, temporary UI access links for specific devices.

Parameters:

Name Type Description Default
id_or_name Union[str, List[str]]

Target device MAC(s) or name(s).

 required

Returns:

Name Type Description
DeviceUiAccess DeviceUiAccess

Secure access URLs for the requested devices.
Source code in src/xovis/api/hub/resources/hub_device.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
async def get_device_ui_access(self, id_or_name: Union[str, list[str]]) -> DeviceUiAccess:
    """
    Generates secure, temporary UI access links for specific devices.

    Args:
        id_or_name (Union[str, List[str]]): Target device MAC(s) or name(s).

    Returns:
        DeviceUiAccess: Secure access URLs for the requested devices.
    """
    device_id = self._resolve_mac_address(id_or_name)
    response = await self._http.get(f"{self._base_path}/{device_id}/webui_link")
    return DeviceUiAccess.model_validate(response.json())
get_devices() async

Retrieves all devices associated with the current HUB tenant.

Returns:

Name Type Description
DevicesResponse DevicesResponse

A collection of device metadata.
Source code in src/xovis/api/hub/resources/hub_device.py 
44
45
46
47
48
49
50
51
52
async def get_devices(self) -> DevicesResponse:
    """
    Retrieves all devices associated with the current HUB tenant.

    Returns:
        DevicesResponse: A collection of device metadata.
    """
    response = await self._http.get(self._base_path)
    return DevicesResponse.model_validate(response.json())
update_categories(id_or_name, categories_to_add=None, categories_to_remove=None) async

Modifies categorical tags for a list of devices.

Parameters:

Name Type Description Default
id_or_name Union[str, List[str]]

Target device MAC(s) or name(s).

 required
categories_to_add Optional[List[str]]

Tags to append.

 None
categories_to_remove Optional[List[str]]

Tags to strip.

 None

Returns:

Name Type Description
dict dict

The response from the HUB Cloud.
Source code in src/xovis/api/hub/resources/hub_device.py 
 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
async def update_categories(
    self,
    id_or_name: Union[str, list[str]],
    categories_to_add: Optional[list[str]] = None,
    categories_to_remove: Optional[list[str]] = None,
) -> dict:
    """
    Modifies categorical tags for a list of devices.

    Args:
        id_or_name (Union[str, List[str]]): Target device MAC(s) or name(s).
        categories_to_add (Optional[List[str]]): Tags to append.
        categories_to_remove (Optional[List[str]]): Tags to strip.

    Returns:
        dict: The response from the HUB Cloud.
    """
    device_ids = self._resolve_multiple(id_or_name)
    payload = DevicesCategoriesAssignment(
        device_ids=device_ids,
        categories_to_add=categories_to_add,
        categories_to_remove=categories_to_remove,
    )
    response = await self._http.post(
        f"{self._base_path}/manage_categories",
        json=payload.model_dump(mode="json", exclude_unset=True),
    )
    return response.json() if response.text else {}

xovis.api.hub.resources.hub_license

Xovis SDK - HUB Cloud License Management Resource

Provides the implementation for managing fleet-wide license status and provisioning on the Xovis HUB Cloud. Operates within the Control Plane and State & Topology Plane.

Classes

HubLicensesManager

Bases: HubResourceManager

Manages license provisioning and status on the Xovis HUB Cloud.

This manager orchestrates the distribution and verification of software licenses across the device fleet. It utilizes the HubCacheManager to resolve human-readable names to MAC addresses.

Source code in src/xovis/api/hub/resources/hub_license.py 
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
class HubLicensesManager(HubResourceManager):
    """
    Manages license provisioning and status on the Xovis HUB Cloud.

    This manager orchestrates the distribution and verification of software
    licenses across the device fleet. It utilizes the HubCacheManager to
    resolve human-readable names to MAC addresses.
    """

    def __init__(self, http_client: XovisHTTPClient, cache: Optional["HubCacheManager"] = None):
        """
        Initializes the HubLicensesManager.

        Args:
            http_client (XovisHTTPClient): The resilient HTTP client.
            cache (Optional[HubCacheManager]): The HUB-level cache manager.
        """
        super().__init__(http_client, cache)
        self._base_path = "/license/api/public/v1/licenses"

    async def get_status(self) -> LicenseStatusResponse:
        """
        Retrieves the license status for all devices in the current HUB tenant.

        Returns:
            LicenseStatusResponse: A collection of license status metadata.
        """
        response = await self._http.get(f"{self._base_path}/status")
        return LicenseStatusResponse.model_validate(response.json())

    async def create(self, id_or_name: Union[str, list[str]], bundle_types: list[BundleType]) -> dict:
        """
        Provisions new license bundles to a set of devices.

        Args:
            id_or_name (Union[str, List[str]]): Target device MAC(s) or name(s).
            bundle_types (List[BundleType]): The license bundles to provision.

        Returns:
            dict: The response from the HUB Cloud.
        """
        device_ids = self._resolve_multiple(id_or_name)
        payload = LicenseCreate(device_ids=device_ids, bundle_types=bundle_types)
        response = await self._http.post(self._base_path, json=payload.model_dump(mode="json", exclude_unset=True))
        return response.json() if response.text else {}
Methods:
__init__(http_client, cache=None)

Initializes the HubLicensesManager.

Parameters:

Name Type Description Default
http_client XovisHTTPClient

The resilient HTTP client.

 required
cache Optional[HubCacheManager]

The HUB-level cache manager.

 None
Source code in src/xovis/api/hub/resources/hub_license.py 
28
29
30
31
32
33
34
35
36
37
def __init__(self, http_client: XovisHTTPClient, cache: Optional["HubCacheManager"] = None):
    """
    Initializes the HubLicensesManager.

    Args:
        http_client (XovisHTTPClient): The resilient HTTP client.
        cache (Optional[HubCacheManager]): The HUB-level cache manager.
    """
    super().__init__(http_client, cache)
    self._base_path = "/license/api/public/v1/licenses"
create(id_or_name, bundle_types) async

Provisions new license bundles to a set of devices.

Parameters:

Name Type Description Default
id_or_name Union[str, List[str]]

Target device MAC(s) or name(s).

 required
bundle_types List[BundleType]

The license bundles to provision.

 required

Returns:

Name Type Description
dict dict

The response from the HUB Cloud.
Source code in src/xovis/api/hub/resources/hub_license.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
async def create(self, id_or_name: Union[str, list[str]], bundle_types: list[BundleType]) -> dict:
    """
    Provisions new license bundles to a set of devices.

    Args:
        id_or_name (Union[str, List[str]]): Target device MAC(s) or name(s).
        bundle_types (List[BundleType]): The license bundles to provision.

    Returns:
        dict: The response from the HUB Cloud.
    """
    device_ids = self._resolve_multiple(id_or_name)
    payload = LicenseCreate(device_ids=device_ids, bundle_types=bundle_types)
    response = await self._http.post(self._base_path, json=payload.model_dump(mode="json", exclude_unset=True))
    return response.json() if response.text else {}
get_status() async

Retrieves the license status for all devices in the current HUB tenant.

Returns:

Name Type Description
LicenseStatusResponse LicenseStatusResponse

A collection of license status metadata.
Source code in src/xovis/api/hub/resources/hub_license.py 
39
40
41
42
43
44
45
46
47
async def get_status(self) -> LicenseStatusResponse:
    """
    Retrieves the license status for all devices in the current HUB tenant.

    Returns:
        LicenseStatusResponse: A collection of license status metadata.
    """
    response = await self._http.get(f"{self._base_path}/status")
    return LicenseStatusResponse.model_validate(response.json())