Add provisioning filters

azext_iot/central/commands_device.py

@@ -68,10 +68,16 @@ def registration_info(
     device_id=None,
     token=None,
     central_dns_suffix="azureiotcentral.com",
+    device_status=None,
 ):
-    provider = CentralDeviceProvider(cmd=cmd, app_id=app_id, token=token)
+    provider = CentralDeviceProvider(cmd=cmd, app_id=app_id, token=token,)
     if not device_id:
-        return provider.get_all_registration_info(central_dns_suffix=central_dns_suffix)
+        return provider.get_all_registration_info(
+            central_dns_suffix=central_dns_suffix, device_status=device_status
+        )
+
     return provider.get_device_registration_info(
-        device_id=device_id, central_dns_suffix=central_dns_suffix
+        device_id=device_id,
+        central_dns_suffix=central_dns_suffix,
+        device_status=device_status,
     )

azext_iot/central/models/__init__.py

@@ -0,0 +1,5 @@
+# coding=utf-8
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------

azext_iot/central/models/enum.py

@@ -0,0 +1,23 @@
+# coding=utf-8
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+"""
+Enum definitions for central
+
+"""
+
+from enum import Enum
+
+
+class DeviceStatus(Enum):
+    """
+    Type of Device status.
+    """
+
+    provisioned = "provisioned"
+    registered = "registered"
+    blocked = "blocked"
+    unassociated = "unassociated"

azext_iot/central/params.py

@@ -8,7 +8,8 @@
 CLI parameter definitions.
 """
 
-from azure.cli.core.commands.parameters import get_three_state_flag
+from azure.cli.core.commands.parameters import get_three_state_flag, get_enum_type
+from azext_iot.central.models.enum import DeviceStatus
 
 
 def load_central_arguments(self, _):
@@ -60,3 +61,9 @@ def load_central_arguments(self, _):
             help="Central dns suffix. "
             "This enables running cli commands against non public/prod environments",
         )
+        context.argument(
+            "device_status",
+            options_list=["--devicestatus", "-s"],
+            arg_type=get_enum_type(DeviceStatus),
+            help="Indicates filter option for device status",
+        )

azext_iot/central/providers/device_provider.py

@@ -8,6 +8,7 @@
 from knack.log import get_logger
 from azext_iot.central import services as central_services
 from azext_iot.dps.services import global_service as dps_global_service
+from azext_iot.central.models.enum import DeviceStatus
 
 logger = get_logger(__name__)
 
@@ -170,44 +171,91 @@ def get_device_credentials(
         return credentials
 
     def get_device_registration_info(
-        self, device_id, central_dns_suffix="azureiotcentral.com",
+        self,
+        device_id,
+        device_status: DeviceStatus,
+        central_dns_suffix="azureiotcentral.com",
     ):
+        dps_state = {}
         info = self._device_registration_info.get(device_id)
 
         if not info:
-            credentials = self.get_device_credentials(
-                device_id=device_id, central_dns_suffix=central_dns_suffix
+            device_info = self.get_device(device_id)
+            device_essential_info = central_services.device.device_populate_essential_info(
+                device_info, device_status
             )
-            id_scope = credentials["idScope"]
-            key = credentials["symmetricKey"]["primaryKey"]
-            dps_state = dps_global_service.get_registration_state(
-                id_scope=id_scope, key=key, device_id=device_id
+            if (
+                device_essential_info.get("deviceStatus")
+                is DeviceStatus.provisioned.value
+            ):
+                credentials = self.get_device_credentials(
+                    device_id=device_id, central_dns_suffix=central_dns_suffix
+                )
+                id_scope = credentials["idScope"]
+                key = credentials["symmetricKey"]["primaryKey"]
+                dps_state = dps_global_service.get_registration_state(
+                    id_scope=id_scope, key=key, device_id=device_id
+                )
+            dps_state = self.dps_populate_essential_info(
+                dps_state, device_essential_info.get("deviceStatus")
             )
-            central_info = self.get_device(device_id)
             info = {
                 "@device_id": device_id,
                 "dps_state": dps_state,
-                "central_info": central_info,
+                "device_info": device_essential_info,
             }
 
         self._device_registration_info[device_id] = info
 
         return info
 
-    def get_all_registration_info(self, central_dns_suffix="azureiotcentral.com"):
+    def dps_populate_essential_info(self, dps_info, device_status):
+        error = {
+            "provisioned": "None",
+            "registered": "Device it not yet provisioned.",
+            "blocked": "Device is blocked by admin",
+            "unassociated": "Device does not have a valid template associated with it",
+        }
+
+        filtered_dps_info = {
+            "status": dps_info.get("status"),
+            "error": error.get(device_status),
+        }
+        return filtered_dps_info
+
+    def get_all_registration_info(
+        self, device_status, central_dns_suffix="azureiotcentral.com"
+    ):
+
         logger.warning("This command may take a long time to complete execution.")
         devices = self.list_devices(central_dns_suffix=central_dns_suffix)
+
         real_devices = [
             device for device in devices.values() if not device["simulated"]
         ]
-        if len(devices) != len(real_devices):
+
+        real_devices_with_status = [
+            central_services.device.update_device_status(device)
+            for device in real_devices
+        ]
+
+        filtered_devices = real_devices_with_status
+
+        if device_status:
+            filtered_devices = [
+                device
+                for device in real_devices_with_status
+                if device.get("deviceStatus") is device_status
+            ]
+
+        if len(devices) != len(filtered_devices):
             logger.warning(
-                "Getting registration info for following devices. "
-                "All other devices are simulated. "
-                "{}".format([device["id"] for device in real_devices])
+                "Getting registration info for real devices. "
+                "{}".format([device["id"] for device in filtered_devices])
             )
         result = [
-            self.get_device_registration_info(device["id"]) for device in real_devices
+            self.get_device_registration_info(device["id"], device["deviceStatus"])
+            for device in filtered_devices
         ]
 
         return result

azext_iot/central/services/device.py

@@ -9,6 +9,7 @@
 
 from knack.util import CLIError
 from azext_iot.central.services import _utility
+from azext_iot.central.models.enum import DeviceStatus
 
 BASE_PATH = "api/preview/devices"
 
@@ -175,3 +176,38 @@ def get_device_credentials(
 
     response = requests.get(url, headers=headers)
     return _utility.try_extract_result(response)
+
+
+def determine_device_status(device):
+    if device["approved"] is False:
+        return DeviceStatus.blocked.value
+    else:
+        if not device.get("instanceOf"):
+            return DeviceStatus.unassociated.value
+
+        else:
+            if device["provisioned"] is False:
+                return DeviceStatus.registered.value
+
+            else:
+                return DeviceStatus.provisioned.value
+
+
+def update_device_status(device):
+    updated_device = device_populate_essential_info(
+        device, determine_device_status(device)
+    )
+    return updated_device
+
+
+def device_populate_essential_info(device, value):
+    if not value:
+        return update_device_status(device)
+    updated_device_data = {
+        "id": device["id"],
+        "displayName": device.get("displayName"),
+        "instanceOf": device.get("instanceOf"),
+        "simulated": device.get("simulated"),
+        "deviceStatus": value,
+    }
+    return updated_device_data