def update_prometheus_metrics()

in agora/cerebral_simulator/src/previous_version/app.py [0:0]

• 37 lines of code
• 13 McCabe index (conditional complexity)

def update_prometheus_metrics(equipment_type, device_id, data):
    try:
        if equipment_type == "Refrigerator":
            FRIDGE_TEMP.labels(device_id=device_id).set(data["temperature_celsius"])
            FRIDGE_DOOR_OPEN.labels(device_id=device_id).set(1 if data["door_open"] else 0)
            FRIDGE_POWER_USAGE.labels(device_id=device_id).set(data["power_usage_kwh"])
        elif equipment_type == "Scale":
            SCALE_WEIGHT.labels(device_id=device_id).set(data["weight_kg"])
            SCALE_TARE_WEIGHT.labels(device_id=device_id).set(data["tare_weight_kg"])
        elif equipment_type == "POS":
            POS_ITEMS_SOLD.labels(device_id=device_id).inc(data["items_sold"])
            POS_TOTAL_AMOUNT.labels(device_id=device_id).inc(data["total_amount_usd"])
            POS_PAYMENT_METHOD.labels(method=data["payment_method"], device_id=device_id).inc()
        elif equipment_type == "SmartShelf":
            SMART_SHELF_STOCK_LEVEL.labels(product_id=data["product_id"], device_id=device_id).set(data["stock_level"])
            SMART_SHELF_THRESHOLD.labels(product_id=data["product_id"], device_id=device_id).set(data["threshold_stock_level"])
        elif equipment_type == "HVAC":
            HVAC_TEMP.labels(device_id=device_id).set(data["temperature_celsius"])
            HVAC_HUMIDITY.labels(device_id=device_id).set(data["humidity_percent"])
            HVAC_POWER_USAGE.labels(device_id=device_id).set(data["power_usage_kwh"])
            HVAC_MODE.labels(device_id=device_id).set(1 if data["operating_mode"] == "heating" else 0)
        elif equipment_type == "LightingSystem":
            LIGHTING_BRIGHTNESS.labels(device_id=device_id).set(data["brightness_level"])
            LIGHTING_POWER_USAGE.labels(device_id=device_id).set(data["power_usage_kwh"])
            LIGHTING_STATUS.labels(device_id=device_id).set(1 if data["status"] == "on" else 0)
        elif equipment_type == "AutomatedCheckout":
            AUTO_CHECKOUT_ITEMS_SCANNED.labels(device_id=device_id).inc(data["items_scanned"])
            AUTO_CHECKOUT_TOTAL_AMOUNT.labels(device_id=device_id).inc(data["total_amount_usd"])
            AUTO_CHECKOUT_PAYMENT_METHOD.labels(method=data["payment_method"], device_id=device_id).inc()
            AUTO_CHECKOUT_ERRORS.labels(device_id=device_id).inc(data["errors"])
        
        DATA_POINTS_GENERATED.labels(equipment_type=equipment_type, device_id=device_id).inc()
    except KeyError as e:
        logger.error(f"Error updating Prometheus metrics for {device_id}: Missing key {str(e)}")
        LOG_MESSAGES.labels(level="error").inc()
    except Exception as e:
        logger.error(f"Error updating Prometheus metrics for {device_id}: {str(e)}")
        LOG_MESSAGES.labels(level="error").inc()