def get_subscription_definition()

in groups/group_setup.py [0:0]

• 54 lines of code
• 1 McCabe index (conditional complexity)

    def get_subscription_definition(self, config):
        """
        Get the Arm Group Type's subscription definition

        :param config: gg_group_setup.GroupConfigFile used with the Group Type
        :return: the subscription definition used to provision the group
        """
        cfg = config
        d = cfg['devices']
        l = cfg['lambda_functions']
        s = cfg['subscriptions']

        definition = [
            {  # from Group's arm device to bridge device
                "Id": "40",
                "Source": d[self.arm_ggd_name]['thing_arn'],
                "Subject": s["all"],
                "Target": d[self.bridge_ggd_name]['thing_arn']
            },
            {  # from Group's arm device to AWS cloud
                "Id": "41",
                "Source": d[self.arm_ggd_name]['thing_arn'],
                "Subject": s['stages'],
                "Target": "cloud"
            },
            {  # from Group's arm device ArmErrorDetector Lambda
                "Id": "42",
                "Source": d[self.arm_ggd_name]['thing_arn'],
                "Subject": s['telemetry'],
                "Target": l['ArmErrorDetector']['arn']
            },
            {  # from ArmErrorDetector Lambda to bridge device
                "Id": "50",
                "Source": l['ArmErrorDetector']['arn'],
                "Subject": s['errors'],
                "Target": d[self.bridge_ggd_name]['thing_arn']
            },
            {  # from ArmErrorDetector Lambda to AWS cloud
                "Id": "51",
                "Source": l['ArmErrorDetector']['arn'],
                "Subject": s['errors'],
                "Target": "cloud"
            },
            {  # from Group's heartbeat device to bridge device
                "Id": "95",
                "Source": d[self.heartbeat_ggd_name]['thing_arn'],
                "Subject": "heart/beat",
                "Target": d[self.bridge_ggd_name]['thing_arn']
            },
            {  # from Group's heartbeat device to AWS cloud
                "Id": "97",
                "Source": d[self.heartbeat_ggd_name]['thing_arn'],
                "Subject": "heart/beat",
                "Target": "cloud"
            }
        ]
        logging.debug(
            '[arm.get_subscription_definition] definition:{0}'.format(
                definition)
        )

        return definition