Add provisioning filters

azext_iot/central/commands_device.py

@@ -7,6 +7,7 @@
 
 from knack.util import CLIError
 from azext_iot.central.providers import CentralDeviceProvider
+from azext_iot.common.shared import DeviceStatus
 
 
 def list_devices(
@@ -68,10 +69,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/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.common.shared 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", "-ds"],
+            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.common.shared import DeviceStatus
 
 logger = get_logger(__name__)
 
@@ -170,11 +171,15 @@ def get_device_credentials(
         return credentials
 
     def get_device_registration_info(
-        self, device_id, central_dns_suffix="azureiotcentral.com",
+        self, device_id, device_status, central_dns_suffix="azureiotcentral.com"
     ):
         info = self._device_registration_info.get(device_id)
 
         if not info:
+            central_info = self.get_device(device_id)
+            central_info = self.update_device_info_filter_status(
+                central_info, device_status
+            )
             credentials = self.get_device_credentials(
                 device_id=device_id, central_dns_suffix=central_dns_suffix
             )
@@ -183,7 +188,9 @@ def get_device_registration_info(
             dps_state = dps_global_service.get_registration_state(
                 id_scope=id_scope, key=key, device_id=device_id
             )
-            central_info = self.get_device(device_id)
+            dps_state = self.filter_dps_info(
+                dps_state, central_info.get("deviceStatus")
+            )
             info = {
                 "@device_id": device_id,
                 "dps_state": dps_state,
@@ -194,20 +201,92 @@ def get_device_registration_info(
 
         return info
 
-    def get_all_registration_info(self, central_dns_suffix="azureiotcentral.com"):
+    def filter_dps_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"),
+        }
+
+        filtered_dps_info.update({"error": error.get(device_status)})
+        return filtered_dps_info
+
+    def update_device_info_filter_status(self, device, value):
+        if value is None:
+            return self.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
+
+    def update_device_status(self, device):
+        if device["approved"] is False:
+            updated_device = self.update_device_info_filter_status(
+                device, DeviceStatus.blocked.value
+            )
+        else:
+            if device.get("instanceOf") is None:
+                updated_device = self.update_device_info_filter_status(
+                    device, DeviceStatus.unassociated.value
+                )
+            else:
+                if device["provisioned"] is False:
+                    updated_device = self.update_device_info_filter_status(
+                        device, DeviceStatus.registered.value
+                    )
+                else:
+                    updated_device = self.update_device_info_filter_status(
+                        device, DeviceStatus.provisioned.value
+                    )
+        return updated_device
+
+    def update_all_device_status(self, devices):
+        filtered_device_list = []
+        for device in devices:
+            filtered_device_list.append(self.update_device_status(device))
+        return filtered_device_list
+
+    def filter_device_list(self, devices, device_status):
+        filtered_device_list = []
+        updated_device_list = self.update_all_device_status(devices)
+        if device_status is None:
+            return updated_device_list
+
+        for device in updated_device_list:
+            if device.get("deviceStatus") is device_status:
+                filtered_device_list.append(device)
+        return filtered_device_list
+
+    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):
+
+        filtered_devices = self.filter_device_list(real_devices, 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])
+                "{}".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/common/shared.py

@@ -214,3 +214,14 @@ class JobStatusType(Enum):
     cancelled = "cancelled"
     scheduled = "scheduled"
     queued = "queued"
+
+
+class DeviceStatus(Enum):
+    """
+    Type of Device status.
+    """
+
+    provisioned = "provisioned"
+    registered = "registered"
+    blocked = "blocked"
+    unassociated = "unassociated"