def GetRealtimeData()

in render-wallboard.py [0:0]

• 35 lines of code
• 14 McCabe index (conditional complexity)

def GetRealtimeData():
    global Logger,LastRealtimeRun,Data,DataSources,MetricUnitMapping

    #
    # We only want to poll the real-time API every so often.
    #
    Logger.debug("Last real-time poll at "+str(LastRealtimeRun)+", timeout is "+str(RealtimeTimeout)+", now is "+str(time.time()))
    
    if time.time() < LastRealtimeRun+RealtimeTimeout: return
    LastRealtimeRun = time.time()

    Connect = boto3.client("connect")
    
    #
    # Even though data sources are defined per wallboard we will always retrieve
    # all of them each time as they may be cross-referenced on other wallboards.
    #
    # First build a list of information we need from the API.
    #
    ConnectList = {}
    for WallboardName in DataSources:
        for Item in DataSources[WallboardName]:
            if Item not in Data: Data[Item] = "0"

            (ConnectARN,QueueARN,Metric) = DataSources[WallboardName][Item].split(":")
 
            if ConnectARN not in ConnectList: ConnectList[ConnectARN] = {}
            if QueueARN not in ConnectList[ConnectARN]: ConnectList[ConnectARN][QueueARN] = []
            ConnectList[ConnectARN][QueueARN].append({"Name":Metric, "Unit":MetricUnitMapping[Metric]})

    #
    # Now call the API for each Connect instance we're interested in.
    #
    for Instance in ConnectList:
        Logger.debug("Retrieving real-time data from "+Instance)
        
        MetricList = []
        for Queue in ConnectList[Instance]:
            MetricList += ConnectList[Instance][Queue]

        try:
            Response = Connect.get_current_metric_data(
                           InstanceId=Instance,
                           Groupings=["QUEUE"],
                           Filters={"Queues":list(ConnectList[Instance].keys())},
                           CurrentMetrics=MetricList)
        except Exception as e:
            Logger.error("Failed to get real-time data: "+str(e))
            continue

        if "MetricResults" not in Response: continue
        for Collection in Response["MetricResults"]:
            QueueARN   = Collection["Dimensions"]["Queue"]["Id"]

            for Metric in Collection["Collections"]:
                MetricName  = Metric["Metric"]["Name"]
                MetricValue = Metric["Value"]
                StoreMetric(Instance, QueueARN, MetricName, MetricValue)