Ingest data received on MQTT

[maaikelimper]

wis2box should be able to receive data over MQTT

Maaike is currently working with Campbell on a demo-setup sending data over MQTT to help develop this new feature.

I suggest that messages be published using the following topic-hierarchy:
data-incoming///
so for Zambia it would be something like :
data-incoming/zmb/campbell-v1/0-20000-0-67665

And Maaike will update the the wis2box-subcriber to subcribe to 'data-incoming/#' to parse incoming data ...

[tomkralidis]

• copy to wis2box-incoming
• existing workflow takes over
• could be a MQTT-forwarder container

[golfvert]

As we are moving toward an IOT type of solution, may I suggest having a look at https://github.com/eclipse-sparkplug/sparkplug ?
It standardizes the relationship between the IOT device (for us AWS) and the collection point.
Mosquitto, as a broker, is compliant with Sparkplug. I haven't looked into the specs very thoroughly, however, in the "standard first approach", it makes sense to look into it.

[maaikelimper]

Adding @efucile @david-i-berry for info

Here are the result of my tests with Campbell:

In the test we receive data on the topic:
data-incoming/zmb/campbell-v1/MetDpt-WIS2Test/data/cr1000x/34769/SYNOP

The message has the following format:

{
  "type": "Feature",
  "geometry": {
    "type": "Point",
    "coordinates": [
      28.320101,
      -15.4158,
      1280
    ]
  },
  "properties": {
    "loggerID": "CR1000X_34769",
    "observationNames": [
      "WMO_Block",
      "Station_ID",
      "Station_Name",
      "WMO_Station_Type",
      "M_Year",
      "M_Month",
      "M_DayOfMonth",
      "M_HourOfDay",
      "M_Minutes",
      "Latitude",
      "Longitude",
      "Elevation",
      "BP_Elevation",
      "BP",
      "QNH",
      "BP_Change",
      "BP_Tendency",
      "Temp_H",
      "AirTempK",
      "DewPointTempK",
      "RH",
      "Sun_hr",
      "SunHrs",
      "Sun_hr24",
      "SunHrs24",
      "Rain_H",
      "Rain_hr",
      "Rain_mm_Tot",
      "Temp_hr24",
      "Temp24T",
      "AirTempMaxK",
      "AirTempMinK",
      "WSpeed_height",
      "Wind_Type",
      "Wind_Sig",
      "Wind_T",
      "WSpeed",
      "WindDir",
      "WSpeed10M_Avg",
      "WindG_Sig",
      "WindGust",
      "Solar_hr",
      "SlrJ",
      "Solar_hr24",
      "SlrJ24"
    ],
    "observations": {
      "2023-04-05T11:29:00Z": [
        null,
        "MetDpt-WIS2Test",
        "MetDpt-WIS2Test",
        "0",
        2023,
        4,
        5,
        9,
        29,
        -15.42,
        28.32,
        1280,
        137,
        99940,
        101577.6,
        0,
        4,
        4,
        302.1,
        290,
        48.1,
        -1,
        1,
        -24,
        12.5,
        4,
        -1,
        0,
        -24,
        6,
        303.4,
        300.4,
        4,
        "0",
        2,
        -10,
        0.223,
        97.3,
        0.335,
        "None",
        0.76,
        -1,
        42950,
        -24,
        529810
      ]
    }
  }
}

I propose to prepare a new service "wis2box-data-subscriber" that subscribes to "data-incoming/zmb/#/SYNOP", parses the messages and write the data as csv into MinIO in the directory:
"zmb/<center_id>/data/core/weather/surface-based-observations/synop"
such that the data is ingested as usual.

The service will require an environment variable CENTER_ID to be provided by the user.
If the user wants to ingest data for multiple centers into a single wis2box, multiple "wis2box-data-subscriber"-services would have to be defined ... Acceptable?

[golfvert]

Is the format of the data open for discussion ?
The challenge will be to agree on a standard between the AWS (Campbell and others...) and the wis2box. This geoJSON looks valid, however the two arrays with observationsName and then observations is a weird choice.
There might be very good reasons, however, like this is, it might not be the most convenient format to transform... No ?

[tomkralidis]

At this stage of the data management lifecycle, I think the absolute lowest barrier should be put forth. GeoJSON doesn't provide much value at this phase (it does during dissemination, of course). The Malawi CSV example provides a dead simple encoding, and compatible/ready with wis2box workflow, for example.

[golfvert]

For me, the end goal is for "any" AWS to talk to "any" data acquisition system (not only wis2box...).
The same way we have standardize the MQTT topic hierarchy, the message format... This allows any implementation (wis2box, moving weather,...) to talk with the broker, cache,...
So, to show it is doable, any format can (almost) do. On the long run, a standard will be needed (hence my suggestion to look at sparkplug).

[tomkralidis]

Agree. CSV is the lowest barrier approach for long term sustainability.