mqtt

cleanup_mqtt_values_field

cleanup_mqtt_values_field(
    values: Dict[str, Any]
) -> Dict[str, Any]

Cleanup the values from invalid keys

Drop the possible invalid keys from the values and normalize the keys.

Parameters:

• values (Dict[str, Any]) –

  Original dict

Returns:

• Dict[str, Any] –

  The cleaned-up values

Source code in src/grottext/ha/mqtt.py

def cleanup_mqtt_values_field(values: Dict[str, Any]) -> Dict[str, Any]:
    """Cleanup the values from invalid keys

    Drop the possible invalid keys from the values and normalize the keys.

    Parameters:
        values: Original dict

    Returns:
        The cleaned-up values
    """
    return {k.strip(): v for k, v in values.items() if is_valid_mqtt_topic(k)}

is_valid_mqtt_topic

is_valid_mqtt_topic(key_name: str) -> bool

Check if the key is a valid mqtt topic

Look If the possible topic is valid, based on the MQTT spec and reserved keywords

Parameters:

• key_name (str) –

  The value of the key (e.g. "ACDischarWatt")

Returns:

• bool –

  True if the key is a valid mqtt topic, False otherwise

Source code in src/grottext/ha/mqtt.py

def is_valid_mqtt_topic(key_name: str) -> bool:
    """Check if the key is a valid mqtt topic

    Look If the possible topic is valid, based on the MQTT spec and reserved keywords

    Parameters:
        key_name: The value of the key (e.g. "ACDischarWatt")

    Returns:
         True if the key is a valid mqtt topic, False otherwise
    """
    key_name = key_name.strip()
    # Character used to bind wildcard topics
    if key_name.startswith("#"):
        return False
    # Character used to bind single level topics
    if key_name.startswith("+"):
        return False
    # should not start or end with /
    if key_name.startswith("/"):
        return False
    if key_name.endswith("/"):
        return False
    # system topics
    if key_name.startswith("$"):
        return False
    return True

make_payload

make_payload(
    conf: FakeConf,
    device: str,
    key: str,
    name: Optional[str] = None,
) -> Dict[str, str]

Generate a MQTT payload for a sensor

Use default values to create a sensor payload, then update with custom attributes if they exist. E.g., unit_of_measurement/total increasing/etc.

Parameters:

• conf (FakeConf) –

  The configuration object, used to extract default divider

• device (str) –

  Use the device name as part of the sensor name + device

• key (str) –

  The key of the sensor sent by grott

• name (Optional[str], default: None ) –

  The name of the sensor, if you want something different

Returns:

• Dict[str, str] –

  A dictionary with the MQTT configuration payload

Source code in src/grottext/ha/mqtt.py

def make_payload(conf: FakeConf, device: str, key: str, name: Optional[str] = None) -> Dict[str, str]:
    """Generate a MQTT payload for a sensor

    Use default values to create a sensor payload, then update with custom
    attributes if they exist.
    E.g., unit_of_measurement/total increasing/etc.

    Parameters:
        conf: The configuration object, used to extract default divider
        device: Use the device name as part of the sensor name + device
        key: The key of the sensor sent by grott
        name: The name of the sensor, if you want something different

    Returns:
        A dictionary with the MQTT configuration payload
    """

    if not device:
        msg = "device is required"
        raise AttributeError(msg)
    if not key:
        msg = "key is required"
        raise AttributeError(msg)
    if not conf.layout:
        msg = "grott config class error"
        raise AttributeError(msg)
    if conf.layout not in conf.recorddict:
        msg = "grott config class error, missing record layout"
        raise AttributeError(msg)

    if name is None:
        name = key

    sensor = mapping.get(key, None)

    value_template = None
    if sensor and sensor.value_template:
        value_template = sensor.value_template
    else:
        # Reuse the existing divide value if available and not existing
        # and apply it to the HA config
        layout = conf.recorddict[conf.layout]
        if key in layout:
            # From grottdata:207, default type is num, also process numx
            register_type = layout[key].get("type", "num")
            # The register is a numerical type
            if register_type in ("num", "numx"):
                # default divide is 1, if not found
                divider = layout[key].get("divide", "1")
                value_template = f"{{{{ value_json.{key} | float / {divider} }}}}"
            elif register_type == "log":
                # register is already a float, no need to divide
                value_template = f"{{{{ value_json.{key} | float }}}}"
            elif register_type == "logpos":
                # register is already a float, no need to divide
                value_template = f"{{{{ [value_json.{key} | float, 0.0] | max }}}}"
            elif register_type == "logneg":
                # register is already a float, no need to divide
                value_template = f"{{{{ [value_json.{key} | float, 0.0] | min }}}}"

    if value_template is None:
        value_template = f"{{{{ value_json.{key} }}}}"

    # Default configuration payload
    payload = MQTTConfigPayload(
        name="{name}",
        unique_id=f"grott_{device}_{key}",  # Generate a unique device ID
        state_topic=f"homeassistant/grott/{device}/state",
        device=Device(
            identifiers=[device],  # Group under a device
            name=device,
            manufacturer="GrowWatt",
        ),
        value_template=value_template,
        expire_after=MQTT_EXPIRE_AFTER,
    )

    if sensor is not None:
        # Update the payload with the sensor configuration
        payload.name = sensor.name
        payload.icon = sensor.icon
        payload.state_class = sensor.state_class
        payload.device_class = sensor.device_class
        payload.icon = sensor.icon
        payload.entity_category = sensor.entity_category
        payload.unit_of_measurement = sensor.unit_of_measurement

    # Generate the name of the key, with all the param available
    payload.name = payload.name.format(device=device, name=name, key=key)
    # HA automatically group the sensors if the device name is prepended

    return to_dict(payload)