def _convert_sub_points_from_csv()

in azext_edge/edge/providers/rpsaas/adr/assets.py [0:0]

• 36 lines of code
• 9 McCabe index (conditional complexity)

def _convert_sub_points_from_csv(sub_points: List[Dict[str, str]]):
    csv_conversion_map = {
        "CapabilityId": "capabilityId",
        "Capability Id": "capabilityId",
        "Data Source": "dataSource",
        "EventName": "name",
        "EventNotifier": "eventNotifier",
        "Event Notifier": "eventNotifier",
        "Name": "name",
        "NodeID": "dataSource",
        "ObservabilityMode": "observabilityMode",
        "Observability Mode": "observabilityMode",
        "QueueSize": "queueSize",
        "Queue Size": "queueSize",
        "Sampling Interval Milliseconds": "samplingInterval",
        "TagName" : "name",
    }
    for point in sub_points:
        # point has csv values
        point.pop("", None)
        for key, value in csv_conversion_map.items():
            if key in point:
                point[value] = point.pop(key)
        # now the point has the normal values - do some final transformations
        if point.get("observabilityMode"):
            point["observabilityMode"] = point["observabilityMode"].capitalize()
        configuration = {}
        if point.get("samplingInterval"):
            configuration["samplingInterval"] = int(point.pop("samplingInterval"))
        else:
            point.pop("samplingInterval", None)
        if point.get("queueSize"):
            configuration["queueSize"] = int(point.pop("queueSize"))
        else:
            point.pop("queueSize", None)
        if configuration:
            config_key = "dataPointConfiguration" if "dataSource" in point else "eventConfiguration"
            point[config_key] = json.dumps(configuration)