From febe0556f289be96fccea3c09cf2a88efd84b91a Mon Sep 17 00:00:00 2001
From: gifrerenom <lluis.gifre@cttc.es>
Date: Tue, 24 Jan 2023 12:26:37 +0000
Subject: [PATCH] Common - Context Queries:
- added get_device helper method
---
src/common/tools/context_queries/Device.py | 21 +++++++++++++++++++--
1 file changed, 19 insertions(+), 2 deletions(-)
diff --git a/src/common/tools/context_queries/Device.py b/src/common/tools/context_queries/Device.py
index e5b205d46..ed8772cf6 100644
--- a/src/common/tools/context_queries/Device.py
+++ b/src/common/tools/context_queries/Device.py
@@ -12,11 +12,28 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from typing import List, Set
-from common.proto.context_pb2 import ContextId, Device, Empty, Topology, TopologyId
+import grpc, logging
+from typing import List, Optional, Set
+from common.proto.context_pb2 import ContextId, Device, DeviceId, Empty, Topology, TopologyId
from common.tools.object_factory.Topology import json_topology_id
from context.client.ContextClient import ContextClient
+LOGGER = logging.getLogger(__name__)
+
+def get_device(context_client : ContextClient, device_uuid : str, rw_copy : bool = False) -> Optional[Device]:
+ try:
+ # pylint: disable=no-member
+ device_id = DeviceId()
+ device_id.device_uuid.uuid = device_uuid
+ ro_device = context_client.GetDevice(device_id)
+ if not rw_copy: return ro_device
+ rw_device = Device()
+ rw_device.CopyFrom(ro_device)
+ return rw_device
+ except grpc.RpcError:
+ #LOGGER.exception('Unable to get Device({:s})'.format(str(device_uuid)))
+ return None
+
def get_existing_device_uuids(context_client : ContextClient) -> Set[str]:
existing_device_ids = context_client.ListDeviceIds(Empty())
existing_device_uuids = {device_id.device_uuid.uuid for device_id in existing_device_ids.device_ids}
--
GitLab