def get_subscription_definition()

in notebooks/iot_greengress/health_tracker/group_setup.py [0:0]

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

    def get_subscription_definition(self, config):
        """
        Get the Tracker 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 TrackerErrorDetector to TrackerBrain Lambda
                "Id": "12",
                "Source": l['TrackerErrorDetector']['arn'],
                "Subject": s['errors'],
                "Target": l['TrackerBrain']['arn']
            },
            {  # from TrackerErrorDetector to AWS cloud
                "Id": "13",
                "Source": l['TrackerErrorDetector']['arn'],
                "Subject": s['errors'],
                "Target": "cloud"
            },
            {  # from Tracker web device to Greengrass Core local shadow
                "Id": "16",
                "Source": d[self.web_ggd_name]['thing_arn'],
                "Subject": "$aws/things/"+ self.tracker_brain_shadow + "/shadow/get",
                "Target": "GGShadowService"
            },
            {  # from Greengrass Core local shadow to Tracker web device
                "Id": "17",
                "Source": "GGShadowService",
                "Subject": "$aws/things/"+ self.tracker_brain_shadow + "/shadow/get/#",
                "Target": d[self.web_ggd_name]['thing_arn']
            },
            {  # from Tracker heartbeat device to AWS cloud
                "Id": "97",
                "Source": d[self.heartbeat_ggd_name]['thing_arn'],
                "Subject": "heart/beat",
                "Target": "cloud"
            },
            {  # from Tracker heart rate device to TrackerBrain Lambda
                "Id": "98",
                "Source": d[self.heartrate_ggd_name]['thing_arn'],
                "Subject": "hr",
                "Target": l['TrackerBrain']['arn']
            }
        ]
        logging.debug(
            '[tracker.get_subscription_definition] definition:{0}'.format(
                definition)
        )
        return definition