diff --git a/src/device/service/driver_api/_Driver.py b/src/device/service/driver_api/_Driver.py
index 7dbb9eddb238dcaae9d00b579a1851aacf53225d..371f4cccb4e002e4d232823e47e31f577d1a4285 100644
--- a/src/device/service/driver_api/_Driver.py
+++ b/src/device/service/driver_api/_Driver.py
@@ -15,16 +15,18 @@
import threading
from typing import Any, Iterator, List, Optional, Tuple, Union
-# Special resource names to request to the driver to retrieve the specified configuration/structural resources.
+# Special resource names to request to the driver to retrieve the specified
+# configuration/structural resources.
# These resource names should be used with GetConfig() method.
-RESOURCE_ENDPOINTS = '__endpoints__'
-RESOURCE_INTERFACES = '__interfaces__'
+RESOURCE_ENDPOINTS = '__endpoints__'
+RESOURCE_INTERFACES = '__interfaces__'
RESOURCE_NETWORK_INSTANCES = '__network_instances__'
-RESOURCE_ROUTING_POLICIES = '__routing_policies__'
-RESOURCE_ACL = '__acl__'
+RESOURCE_ROUTING_POLICIES = '__routing_policies__'
+RESOURCE_ACL = '__acl__'
+
class _Driver:
- def __init__(self, address : str, port : int, **settings) -> None:
+ def __init__(self, address: str, port: int, **settings) -> None:
""" Initialize Driver.
Parameters:
address : str
@@ -56,92 +58,122 @@ class _Driver:
""" Retrieve initial configuration of entire device.
Returns:
values : List[Tuple[str, Any]]
- List of tuples (resource key, resource value) for resource keys.
+ List of tuples (resource key, resource value) for
+ resource keys.
"""
raise NotImplementedError()
- def GetConfig(self, resource_keys : List[str] = []) -> List[Tuple[str, Union[Any, None, Exception]]]:
- """ Retrieve running configuration of entire device, or selected resource keys.
+ def GetConfig(self, resource_keys: List[str] = []) -> \
+ List[Tuple[str, Union[Any, None, Exception]]]:
+ """ Retrieve running configuration of entire device or
+ selected resource keys.
Parameters:
resource_keys : List[str]
List of keys pointing to the resources to be retrieved.
Returns:
values : List[Tuple[str, Union[Any, None, Exception]]]
- List of tuples (resource key, resource value) for resource keys requested. If a resource is found,
- the appropriate value type must be retrieved. If a resource is not found, None must be retrieved as
- value for that resource. In case of Exception, the Exception must be retrieved as value.
+ List of tuples (resource key, resource value) for
+ resource keys requested. If a resource is found,
+ the appropriate value type must be retrieved.
+ If a resource is not found, None must be retrieved as
+ value for that resource. In case of Exception,
+ the Exception must be retrieved as value.
"""
raise NotImplementedError()
- def SetConfig(self, resources : List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
+ def SetConfig(self, resources: List[Tuple[str, Any]]) -> \
+ List[Union[bool, Exception]]:
""" Create/Update configuration for a list of resources.
Parameters:
resources : List[Tuple[str, Any]]
- List of tuples, each containing a resource_key pointing the resource to be modified, and a
- resource_value containing the new value to be set.
+ List of tuples, each containing a resource_key pointing the
+ resource to be modified, and a resource_value containing
+ the new value to be set.
Returns:
results : List[Union[bool, Exception]]
- List of results for resource key changes requested. Return values must be in the same order than
- resource keys requested. If a resource is properly set, True must be retrieved; otherwise, the
- Exception that is raised during the processing must be retrieved.
+ List of results for resource key changes requested.
+ Return values must be in the same order as the
+ resource keys requested. If a resource is properly set,
+ True must be retrieved; otherwise, the Exception that is
+ raised during the processing must be retrieved.
"""
raise NotImplementedError()
- def DeleteConfig(self, resources : List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
+ def DeleteConfig(self, resources: List[Tuple[str, Any]]) -> \
+ List[Union[bool, Exception]]:
""" Delete configuration for a list of resources.
Parameters:
resources : List[Tuple[str, Any]]
- List of tuples, each containing a resource_key pointing the resource to be modified, and a
- resource_value containing possible additionally required values to locate the value to be removed.
+ List of tuples, each containing a resource_key pointing the
+ resource to be modified, and a resource_value containing
+ possible additionally required values to locate
+ the value to be removed.
Returns:
- results : List[bool]
- List of results for resource key deletions requested. Return values must be in the same order than
- resource keys requested. If a resource is properly deleted, True must be retrieved; otherwise, the
- Exception that is raised during the processing must be retrieved.
+ results : List[Union[bool, Exception]]
+ List of results for resource key deletions requested.
+ Return values must be in the same order as the resource keys
+ requested. If a resource is properly deleted, True must be
+ retrieved; otherwise, the Exception that is raised during
+ the processing must be retrieved.
"""
raise NotImplementedError()
- def SubscribeState(self, subscriptions : List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]:
- """ Subscribe to state information of entire device, or selected resources. Subscriptions are incremental.
+ def SubscribeState(self, subscriptions: List[Tuple[str, float, float]]) -> \
+ List[Union[bool, Exception]]:
+ """ Subscribe to state information of entire device or
+ selected resources. Subscriptions are incremental.
Driver should keep track of requested resources.
Parameters:
subscriptions : List[Tuple[str, float, float]]
- List of tuples, each containing a resource_key pointing the resource to be subscribed, a
- sampling_duration, and a sampling_interval (both in seconds with float representation) defining,
- respectively, for how long monitoring should last, and the desired monitoring interval for the
- resource specified.
+ List of tuples, each containing a resource_key pointing the
+ resource to be subscribed, a sampling_duration, and a
+ sampling_interval (both in seconds with float
+ representation) defining, respectively, for how long
+ monitoring should last, and the desired monitoring interval
+ for the resource specified.
Returns:
- results : List[bool]
- List of results for resource key subscriptions requested. Return values must be in the same order
- than resource keys requested. If a resource is properly subscribed, True must be retrieved;
- otherwise, the Exception that is raised during the processing must be retrieved.
+ results : List[Union[bool, Exception]]
+ List of results for resource key subscriptions requested.
+ Return values must be in the same order as the resource keys
+ requested. If a resource is properly subscribed,
+ True must be retrieved; otherwise, the Exception that is
+ raised during the processing must be retrieved.
"""
raise NotImplementedError()
- def UnsubscribeState(self, subscriptions : List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]:
- """ Unsubscribe from state information of entire device, or selected resources. Subscriptions are incremental.
+ def UnsubscribeState(self, subscriptions: List[Tuple[str, float, float]]) \
+ -> List[Union[bool, Exception]]:
+ """ Unsubscribe from state information of entire device
+ or selected resources. Subscriptions are incremental.
Driver should keep track of requested resources.
Parameters:
subscriptions : List[str]
- List of tuples, each containing a resource_key pointing the resource to be subscribed, a
- sampling_duration, and a sampling_interval (both in seconds with float representation) defining,
- respectively, for how long monitoring should last, and the desired monitoring interval for the
- resource specified.
+ List of tuples, each containing a resource_key pointing the
+ resource to be subscribed, a sampling_duration, and a
+ sampling_interval (both in seconds with float
+ representation) defining, respectively, for how long
+ monitoring should last, and the desired monitoring interval
+ for the resource specified.
Returns:
results : List[Union[bool, Exception]]
- List of results for resource key unsubscriptions requested. Return values must be in the same order
- than resource keys requested. If a resource is properly unsubscribed, True must be retrieved;
- otherwise, the Exception that is raised during the processing must be retrieved.
+ List of results for resource key un-subscriptions requested.
+ Return values must be in the same order as the resource keys
+ requested. If a resource is properly unsubscribed,
+ True must be retrieved; otherwise, the Exception that is
+ raised during the processing must be retrieved.
"""
raise NotImplementedError()
def GetState(
self, blocking=False, terminate : Optional[threading.Event] = None
) -> Iterator[Tuple[float, str, Any]]:
- """ Retrieve last collected values for subscribed resources. Operates as a generator, so this method should be
- called once and will block until values are available. When values are available, it should yield each of
- them and block again until new values are available. When the driver is destroyed, GetState() can return
- instead of yield to terminate the loop. Terminate enables to request interruption of the generation.
+ """ Retrieve last collected values for subscribed resources.
+ Operates as a generator, so this method should be called once and will
+ block until values are available. When values are available,
+ it should yield each of them and block again until new values are
+ available. When the driver is destroyed, GetState() can return instead
+ of yield to terminate the loop.
+ Terminate enables to request interruption of the generation.
Examples:
# keep looping waiting for extra samples (generator loop)
terminate = threading.Event()
@@ -161,20 +193,27 @@ class _Driver:
if i == 10: terminate.set()
Parameters:
blocking : bool
- Select the driver behaviour. In both cases, the driver will first retrieve the samples accumulated
- and available in the internal queue. Then, if blocking, the driver does not terminate the loop and
- waits for additional samples to come, thus behaving as a generator. If non-blocking, the driver
- terminates the loop and returns. Non-blocking behaviour can be used for periodically polling the
- driver, while blocking can be used when a separate thread is in charge of collecting the samples
- produced by the driver.
+ Select the driver behaviour. In both cases, the driver will
+ first retrieve the samples accumulated and available in the
+ internal queue. Then, if blocking, the driver does not
+ terminate the loop and waits for additional samples to come,
+ thus behaving as a generator. If non-blocking, the driver
+ terminates the loop and returns. Non-blocking behaviour can
+ be used for periodically polling the driver, while blocking
+ can be used when a separate thread is in charge of
+ collecting the samples produced by the driver.
terminate : threading.Event
- Signals the interruption of the GetState method as soon as possible.
+ Signals the interruption of the GetState method as soon as
+ possible.
Returns:
results : Iterator[Tuple[float, str, Any]]
- Sequences of state sample. Each State sample contains a float Unix-like timestamps of the samples in
- seconds with up to microsecond resolution, the resource_key of the sample, and its resource_value.
- Only resources with an active subscription must be retrieved. Interval and duration of the sampling
- process are specified when creating the subscription using method SubscribeState(). Order of values
- yielded is arbitrary.
+ Sequences of state sample. Each State sample contains a
+ float Unix-like timestamps of the samples in seconds with up
+ to microsecond resolution, the resource_key of the sample,
+ and its resource_value.
+ Only resources with an active subscription must be
+ retrieved. Interval and duration of the sampling process are
+ specified when creating the subscription using method
+ SubscribeState(). Order of values yielded is arbitrary.
"""
raise NotImplementedError()