Processing encoded payload data

I have a Ruuvi gateway and sensors.
The sensors (I have 2 at the moment) update around once per second and thepublish rate cannot be changed.
The gateway publishes sensor updates to a mosquito server.
I control the mosquito server and it runs on the same system as my Indigo server.
The MQTT messages look like:

Code: Select all

 ruuvi/EC:2E:89:E4:21:C7 {
   "gw_mac":   "FC:E7:EA:35:6C:32",
   "rssi":   -14,
   "aoa":   [],
   "gwts":   "1700598017",
   "ts":   "1700598017",
   "data":   "0201061BFF990405124E4BB2BF880020FFD803E4BE56AC31B0EC2E89E421C7",
   "coords":   ""
}

The only information I am interested in is the data key in the payload. The data key is encoded, and when decoded to a Python dict, yields:

Code: Select all

{'data_format': 5, 'humidity': 47.6, 'temperature': 23.38, 'pressure': 988.5, 'acceleration': 1008.7140328160405, 'acceleration_x': 32, 'acceleration_y': -92, 'acceleration_z': 1004, 'tx_power': 4, 'battery': 3192, 'movement_counter': 119, 'measurement_sequence_number': 7458, 'mac': 'ec2e89e421c7'}

The Python code to decode the data is simple:

Code: Select all

from ruuvi_decoders import Df5Decoder
decoder = Df5Decoder()
data = MQTT_Data
data = data.split("FF9904")[1]
sensor = decoder.decode_data(data)
print(sensor)

What I would like is to have an MQTT Shim device with states for the encoded keys. The question is can I do that, and if so, how?
I know I can grab the data key and save it to an Indigo variable and then create a trigger to process the variable and save the encoded keys into other Indigo variables. But, that is not very elegant and doesn't end up with a pseudo device (I.e., one Indigo object), which is what I would prefer. Also, I would like to control the update rate, at least in Indigo (I cannot change the publishing rate in the Ruuvi gateway). Once a minute, or even once every 5 minutes would be fine.

I am open to ideas.

TIA

Hmm. I'm not going to add specialized Shim devices that require a proprietary decoder, so that's out.

You can certainly do it with a Python script that updates variables, There's no way for code outside of a plugin (ie, a script) to updated the states of a plugin device. Just not possible.

So if you don't want to use variables, the only way to do it is to write a custom minimal plugin that creates the Ruvi devices and maintains the sensor devices.

Also, since the data is coming in once per second, you might as well update it that fast. But you could certainly skip updates to get the rate you want. Only reason I would do that is if you're collecting the updates in a database and just don't want that much data.

I need to think about if it's possible to do some sort of "plugin" to a plugin so that users could do custom processing in Shim devices. That might be possible. If so, I could probably do this one as a sample so that other users could do their own later on. I just don't want to have to maintain all the possible custom ones.

#1. Thanks for the quick response.

Hmm. I'm not going to add specialized Shim devices that require a proprietary decoder, so that's out.

Totally understood.

You can certainly do it with a Python script that updates variables, There's no way for code outside of a plugin (ie, a script) to updated the states of a plugin device. Just not possible.

Yes, of course. That is why the Shim option is so attractive.

So if you don't want to use variables, the only way to do it is to write a custom minimal plugin that creates the Ruvi devices and maintains the sensor devices.

Probably more effort than this is worth for just one person and one application.

Also, since the data is coming in once per second, you might as well update it that fast. But you could certainly skip updates to get the rate you want. Only reason I would do that is if you're collecting the updates in a database and just don't want that much data.

No, I am not storing the data. Just not a fan of overdoing things. But, you are probably right here.

I need to think about if it's possible to do some sort of "plugin" to a plugin so that users could do custom processing in Shim devices. That might be possible. If so, I could probably do this one as a sample so that other users could do their own later on. I just don't want to have to maintain all the possible custom ones.

Well, should you think this is worth the effort, I'll certainly be glad to help in any way needed.

Again, thanks

So that was an interesting coding project. I had never done anything with loading code modules on the fly, as it were. I haven't really documented this yet, so I hope this makes sense.

1. Install this pre-release: https://github.com/FlyingDiver/Indigo-S ... g/2023.0.0

2. In a "Decoders" folder inside the plugin, make a copy of the "Expand.py" file.

3. Create a "Decoders" folder in the Indigo install directory. Put the Expand.py file in there, then rename it for your specific purpose. Maybe "Ruuvi.py".

4. Edit that file, and change the class name to the same as the file name. See comments in file.

5. Edit the code per your requirements. The input is the entire payload. The output is dict with one or more key/value pairs. Make sure the key names don't conflict with any existing states.

6. Edit the Shim device definition to specify the decoder you just wrote.

That should be it. You might need to edit the plugin config to specify the "Decoders" folder you created in #3. I think the default is set correctly, but...

The location(s) of the decoders might change. I'm looking at ways to protect it during plugin and Indigo updates. You could put it inside the plugin where the sample one is, but that would get erased when the plugin is updated. I need to make sure it's someplace that gets copied to the new Indigo install folder during Indigo updates.

For discussion purposes, here's the sample decoder included in the plugin. Which is actually quite useful for anyone dealing with MQTT payloads with nested dicts.

Code: Select all

# Example decoder that expands nested dictionaries into a flat dictionary
#
# The class name must be the same as the file name, and the class must have a static method called "decode"
# that takes a single argument, which is the payload of the MQTT message.  The method must return a dictionary
# of states to update, or None if there are no states to update.

class Expand(object):
    @staticmethod
    def decode(payload):
   
        new_states = {}
        for key in payload:
            if type(payload[key]) is dict:
                for subkey in payload[key]:
                    new_key = key + "_" + subkey
                    new_states[new_key] = payload[key][subkey]
        if len(new_states):
            return new_states
        else:
            return None


... 1. Install this pre-release: https://github.com/FlyingDiver/Indigo-S ... g/2023.0.0 ....

On it!

... 1. Install this pre-release: https://github.com/FlyingDiver/Indigo-S ... g/2023.0.0 ....

On it!

Ok. Took a while because of problems getting the right message matching. Now, the packets seem to be fed to the decoder. And I have a different problem. My decoder assumed it would only receive the specific encoded payload item. However, I now realize you are simply feeding the entire payload to the decoder, and it is my job to dig out the one item I want. So, a little work here. But, so far it looks really good. Thanks.!

Should just be payload['data'] to get the part you wanted.

Should just be payload['data'] to get the part you wanted.

And, voila...

Screen Shot 2023-11-22 at 19.32.35.png (55.38 KiB) Viewed 906 times

Thank you. This is perfect. Hopefully, others can benefit from this as well.

Should just be payload['data'] to get the part you wanted.

And, voila...
Thank you. This is perfect. Hopefully, others can benefit from this as well.

And, a question... is the dict returned from the Decoder accessible for any other use in the shim? For example, as the Payload key in a value sensor device, or energy usage, etc? I assume the answer is no because the data is not available at the time those fields are populated. But, just wondering. Probably the single most interesting use would be in the State field displayed in the Indigo devices list. BTW, this is NOT a feature request. Really, just a question.

And, a question... is the dict returned from the Decoder accessible for any other use in the shim? For example, as the Payload key in a value sensor device, or energy usage, etc? I assume the answer is no because the data is not available at the time those fields are populated. But, just wondering. Probably the single most interesting use would be in the State field displayed in the Indigo devices list. BTW, this is NOT a feature request. Really, just a question.

Not the way it's currently written. The value state is processed before all the additional/added states. It's possible that the order of operations could be changed so that the device state is handled last, after all the additional custom states (multi-key and decoded) are created. I'll look at that.

... and, a new wrinkle. At the current message rate, ~2 per second, the queue is getting backed up and throwing errors in the indigo log. Eg:

Code: Select all

   MQTT Connector Warning          Queue for message type '##ruuvi##' has 86 messages pending

The way this is setup at the moment I have one trigger that reads the message queue from an MQTT connector device that only looks for Ruuvi messages, and then 2 shims. I suppose I could simply enable/disable the trigger so that it only ran once a minute or so. Or set/reset a variable to use as a condition in the trigger (same effect). But, is there a better way?

The only way that would happen is if the Shims processing is taking much longer than it should. First, restart the connector plugin (but not the Shims plugin) and see how long it takes for the errors to start appearing.

Second, post the code for your Decoder so I can see if something there is taking a long time.

Also, turn off debug logging (if you have it on) and see if it still gets backed up. The logging adds processing time.

Logging is at informational

Here is the Decoder code:

Code: Select all

from ruuvi_decoders import Df5Decoder

class Ruuvi(object):
    @staticmethod
    def decode(payload):
   
        new_states = {}
        decoder = Df5Decoder()
        ruuvi_data = payload['data']
        ruuvi_data = ruuvi_data.split("FF9904")[1]
        new_states = decoder.decode_data(ruuvi_data)
        if len(new_states):
            return new_states
        else:
            return None

And, just under a minute to start getting the errors.

Code: Select all

23 Nov 2023 at 11:38:15
   Starting plugin "MQTT Connector 2022.1.1" (pid 1510)
   Started plugin "MQTT Connector 2022.1.1"
   MQTT Connector                  CM LinkTap MQTT Responses: Starting Device
   MQTT Connector                  CM MQTT ruuvi connector: Starting Device
   MQTT Connector                  CM OSPi MQTT Receiver: Starting Device
   MQTT Connector                  Shelly MQTT connector: Starting Device
   MQTT Connector                  CM LinkTap MQTT Responses: Subscribing to: cm/LinkTap/up/# (0)
   MQTT Connector                  CM MQTT ruuvi connector: Subscribing to: ruuvi/# (0)
   MQTT Connector                  CM OSPi MQTT Receiver: Subscribing to: opensprinkler/station/# (0)
   MQTT Connector                  CM LinkTap MQTT Responses: Subscribing to: cm/LinkTap2/up/# (0)
   MQTT Connector                  Shelly MQTT connector: Subscribing to: shellies (0)
   MQTT Connector                  CM LinkTap MQTT Responses: Subscribing to: cm/LinkTap_reply/up/# (0)
   MQTT Connector                  Shelly MQTT connector: Subscribing to: shellies/# (0)
   MQTT Connector                  CM LinkTap MQTT Responses: Subscribing to: cm/LinkTap2_reply/up/# (0)

23 Nov 2023 at 11:39:02
   MQTT Connector Warning          Queue for message type '##ruuvi##' has 31 messages pending


Another bit of info. Opening a shim device for editing now hangs for at least a minute, maybe longer. Restarting the server makes no difference.

And, last, here is a picture of the trigger.