Use the te/+/+/+/+/m/+/meta topics to set measurement units

didier-wenzek · didier-wenzek · commit 94058fffd92c · 2025-09-12T15:28:55.000+02:00
Signed-off-by: Didier Wenzek &lt;didier.wenzek@free.fr&gt;
diff --git a/crates/common/tedge_config/src/tedge_toml/tedge_config.rs b/crates/common/tedge_config/src/tedge_toml/tedge_config.rs
@@ -303,8 +303,8 @@ define_tedge_config! {
         mqtt: HostPort<MQTT_TLS_PORT>,
 
         /// Set of MQTT topics the Cumulocity mapper should subscribe to
-        #[tedge_config(example = "te/+/+/+/+/a/+,te/+/+/+/+/m/+,te/+/+/+/+/e/+")]
-        #[tedge_config(default(value = "te/+/+/+/+,te/+/+/+/+/twin/+,te/+/+/+/+/m/+,te/+/+/+/+/e/+,te/+/+/+/+/a/+,te/+/+/+/+/status/health"))]
+        #[tedge_config(example = "te/+/+/+/+/a/+,te/+/+/+/+/m/+,te/+/+/+/+/m/+/meta,te/+/+/+/+/e/+")]
+        #[tedge_config(default(value = "te/+/+/+/+,te/+/+/+/+/twin/+,te/+/+/+/+/m/+,te/+/+/+/+/m/+/meta,te/+/+/+/+/e/+,te/+/+/+/+/a/+,te/+/+/+/+/status/health"))]
         topics: TemplatesSet,
 
         enable: {
diff --git a/crates/extensions/c8y_mapper_ext/src/actor.rs b/crates/extensions/c8y_mapper_ext/src/actor.rs
@@ -177,7 +177,7 @@ impl C8yMapperActor {
                     }
                 }
             } else {
-                self.process_message(message.clone()).await?;
+                self.process_message(message).await?;
             }
         } else {
             self.convert_and_publish(&message).await?;
diff --git a/crates/extensions/c8y_mapper_ext/src/converter.rs b/crates/extensions/c8y_mapper_ext/src/converter.rs
@@ -12,6 +12,7 @@ use crate::entity_cache::UpdateOutcome;
 use crate::error::ConversionError;
 use crate::error::MessageConversionError;
 use crate::json;
+use crate::json::Units;
 use crate::operations;
 use crate::operations::OperationHandler;
 use crate::supported_operations::operation::get_child_ops;
@@ -190,6 +191,8 @@ pub struct CumulocityConverter {
 
     supported_operations: SupportedOperations,
     pub operation_handler: OperationHandler,
+
+    units: Units,
 }
 
 impl CumulocityConverter {
@@ -277,6 +280,7 @@ impl CumulocityConverter {
             recently_completed_commands: HashMap::new(),
             active_commands_last_cleared: Instant::now(),
             operation_handler,
+            units: Units::default(),
         })
     }
 
@@ -470,8 +474,9 @@ impl CumulocityConverter {
 
         if let Some(entity) = self.entity_cache.get(source) {
             // Need to check if the input Thin Edge JSON is valid before adding a child ID to list
+            let units = self.get_measurement_type_units(source, measurement_type);
             let c8y_json_payload =
-                json::from_thin_edge_json(input.payload_str()?, entity, measurement_type, None)?;
+                json::from_thin_edge_json(input.payload_str()?, entity, measurement_type, units)?;
 
             if c8y_json_payload.len() < self.size_threshold.0 {
                 mqtt_messages.push(MqttMessage::new(
@@ -1163,6 +1168,29 @@ impl CumulocityConverter {
         self.active_commands.contains_key(cmd_id)
             || self.recently_completed_commands.contains_key(cmd_id)
     }
+
+    fn get_measurement_type_units(
+        &self,
+        source: &EntityTopicId,
+        measurement_type: &str,
+    ) -> Option<&Units> {
+        let group = format!("{source}/{measurement_type}");
+        self.units.get_group_units(&group)
+    }
+
+    fn set_measurement_type_units(
+        &mut self,
+        source: &EntityTopicId,
+        measurement_type: &str,
+        message: &MqttMessage,
+    ) {
+        if let Ok(payload) = message.payload_str() {
+            if let Ok(meta) = serde_json::from_str(payload) {
+                let group = format!("{source}/{measurement_type}");
+                self.units.set_group_units(group, meta)
+            }
+        }
+    }
 }
 
 #[derive(Error, Debug)]
@@ -1205,6 +1233,10 @@ impl CumulocityConverter {
     ) -> Result<Vec<MqttMessage>, ConversionError> {
         match &channel {
             Channel::EntityMetadata => self.try_convert_entity_registration(source, message),
+            Channel::MeasurementMetadata { measurement_type } => {
+                self.set_measurement_type_units(&source, measurement_type, message);
+                Ok(vec![])
+            }
             _ => {
                 self.try_convert_data_message(source, channel, message)
                     .await
diff --git a/crates/extensions/c8y_mapper_ext/src/json.rs b/crates/extensions/c8y_mapper_ext/src/json.rs
@@ -62,7 +62,7 @@ fn from_thin_edge_json_with_timestamp(
 #[derive(Default)]
 pub struct Units {
     units: HashMap<String, String>,
-    groups: HashMap<String, Units>,
+    group_units: HashMap<String, Units>,
 }
 
 impl Units {
@@ -72,13 +72,13 @@ impl Units {
     pub fn new() -> Units {
         Units {
             units: HashMap::new(),
-            groups: HashMap::new(),
+            group_units: HashMap::new(),
         }
     }
 
     /// True if no units are actually defined
     pub fn is_empty(&self) -> bool {
-        self.units.is_empty() && self.groups.is_empty()
+        self.units.is_empty() && self.group_units.is_empty()
     }
 
     /// Measurement units as defined by metadata published on a measurement topic
@@ -93,12 +93,11 @@ impl Units {
     }
 
     pub fn set_unit(&mut self, measurement: String, meta: serde_json::Value) {
-        if let serde_json::Value::String(unit) = meta {
-            // "Temperature": "°C"
-            self.units.insert(measurement, unit);
-        } else if let Some(unit) = meta.get("unit") {
+        if let Some(unit) = meta.get("unit") {
             // "Temperature": {"unit": "°C"},
-            self.set_unit(measurement, unit.clone());
+            if let serde_json::Value::String(unit_name) = unit {
+                self.units.insert(measurement, unit_name.to_owned());
+            }
         } else {
             // "Climate": { "Temperature": {"unit": "°C"}, "Humidity": {"unit": "%RH"} }
             let group = measurement;
@@ -109,7 +108,7 @@ impl Units {
     pub fn set_group_units(&mut self, group: String, meta: serde_json::Value) {
         let units = Units::from_metadata(meta);
         if !units.is_empty() {
-            self.groups.insert(group, units);
+            self.group_units.insert(group, units);
         }
     }
 
@@ -120,7 +119,7 @@ impl Units {
 
     /// Retrieve the units to be used for a measurement group, if any
     pub fn get_group_units(&self, group: &str) -> Option<&Units> {
-        self.groups.get(group)
+        self.group_units.get(group)
     }
 }
 
diff --git a/tests/RobotFramework/tests/cumulocity/telemetry/thin-edge_device_telemetry.robot b/tests/RobotFramework/tests/cumulocity/telemetry/thin-edge_device_telemetry.robot
@@ -20,6 +20,44 @@ Thin-edge devices support sending simple measurements
     ...    series=temperature
     Log    ${measurements}
 
+Thin-edge devices support sending simple measurements with units
+    # Using a test-specific measurement is required to not interfer with other tests
+    Execute Command    tedge mqtt pub -r te/device/main///m/t1/meta '{ "temperature": { "unit": "°C" } }'
+    Execute Command    tedge mqtt pub te/device/main///m/t1 '{ "temperature": 25.111 }'
+    Execute Command    tedge mqtt pub te/device/main///m/t1 '{ "temperature": 25.222 }'
+    ${measurements}=    Device Should Have Measurements
+    ...    minimum=2
+    ...    maximum=2
+    ...    type=t1
+    ...    value=temperature
+    ...    series=temperature
+    Log    ${measurements}
+    Should Be Equal As Numbers    ${measurements[0]["temperature"]["temperature"]["value"]}    25.111
+    Should Be Equal    ${measurements[0]["temperature"]["temperature"]["unit"]}    °C
+    Should Be Equal As Numbers    ${measurements[1]["temperature"]["temperature"]["value"]}    25.222
+    Should Be Equal    ${measurements[1]["temperature"]["temperature"]["unit"]}    °C
+
+Thin-edge devices support sending complex measurements with units
+    # Using a test-specific measurement is required to not interfer with other tests
+    Execute Command    tedge mqtt pub -r te/device/main///m/t2/meta '{ "Climate": { "Temperature": { "unit": "°C" }, "Pressure": { "unit": "bar" }}}'
+    Execute Command    tedge mqtt pub te/device/main///m/t2 '{ "Climate": { "Temperature": 19.42, "Pressure": 1.013 }, "Engine": { "Temperature": 350.42, "Pressure": 321.0 }}'
+    ${measurements}=    Device Should Have Measurements
+    ...    minimum=1
+    ...    maximum=1
+    ...    type=t2
+    ...    value=Climate
+    ...    series=Temperature
+    Log    ${measurements}
+    Should Be Equal As Numbers    ${measurements[0]["Climate"]["Temperature"]["value"]}    19.42
+    Should Contain    ${measurements[0]["Climate"]["Temperature"]}    unit
+    Should Be Equal    ${measurements[0]["Climate"]["Temperature"]["unit"]}    °C
+    Should Be Equal As Numbers    ${measurements[0]["Climate"]["Pressure"]["value"]}    1.013
+    Should Be Equal    ${measurements[0]["Climate"]["Pressure"]["unit"]}    bar
+    Should Be Equal As Numbers    ${measurements[0]["Engine"]["Temperature"]["value"]}    350.42
+    Should Not Contain    ${measurements[0]["Engine"]["Temperature"]}    unit
+    Should Be Equal As Numbers    ${measurements[0]["Engine"]["Pressure"]["value"]}    321.0
+    Should Not Contain    ${measurements[0]["Engine"]["Pressure"]}    unit
+
 Thin-edge devices support sending simple measurements with custom type
     Execute Command    tedge mqtt pub te/device/main///m/ '{ "type":"CustomType", "temperature": 25 }'
     ${measurements}=    Device Should Have Measurements
diff --git a/tests/RobotFramework/tests/tedge/call_tedge_config_list.robot b/tests/RobotFramework/tests/tedge/call_tedge_config_list.robot
@@ -97,7 +97,7 @@ set/unset c8y.topics
     ${unset}    Execute Command    tedge config list
     Should Contain
     ...    ${unset}
-    ...    c8y.topics=["te/+/+/+/+", "te/+/+/+/+/twin/+", "te/+/+/+/+/m/+", "te/+/+/+/+/e/+", "te/+/+/+/+/a/+", "te/+/+/+/+/status/health"]
+    ...    c8y.topics=["te/+/+/+/+", "te/+/+/+/+/twin/+", "te/+/+/+/+/m/+", "te/+/+/+/+/m/+/meta", "te/+/+/+/+/e/+", "te/+/+/+/+/a/+", "te/+/+/+/+/status/health"]
 
 set/unset c8y.proxy.bind.address
     Execute Command    sudo tedge config set c8y.proxy.bind.address 127.1.1.1    # Changing c8y.proxy.bind.address

Original file line number	Diff line number	Diff line change
`@@ -177,7 +177,7 @@ impl C8yMapperActor {`
`177`	`177`	`}`
`178`	`178`	`}`
`179`	`179`	`} else {`
`180`		`- self.process_message(message.clone()).await?;`
	`180`	`+ self.process_message(message).await?;`
`181`	`181`	`}`
`182`	`182`	`} else {`
`183`	`183`	`self.convert_and_publish(&message).await?;`
Original file line number	Diff line number	Diff line change
`@@ -62,7 +62,7 @@ fn from_thin_edge_json_with_timestamp(`
`62`	`62`	`#[derive(Default)]`
`63`	`63`	`pub struct Units {`
`64`	`64`	`units: HashMap<String, String>,`
`65`		`- groups: HashMap<String, Units>,`
	`65`	`+ group_units: HashMap<String, Units>,`
`66`	`66`	`}`
`67`	`67`
`68`	`68`	`impl Units {`
`@@ -72,13 +72,13 @@ impl Units {`
`72`	`72`	`pub fn new() -> Units {`
`73`	`73`	`Units {`
`74`	`74`	`units: HashMap::new(),`
`75`		`- groups: HashMap::new(),`
	`75`	`+ group_units: HashMap::new(),`
`76`	`76`	`}`
`77`	`77`	`}`
`78`	`78`
`79`	`79`	`/// True if no units are actually defined`
`80`	`80`	`pub fn is_empty(&self) -> bool {`
`81`		`- self.units.is_empty() && self.groups.is_empty()`
	`81`	`+ self.units.is_empty() && self.group_units.is_empty()`
`82`	`82`	`}`
`83`	`83`
`84`	`84`	`/// Measurement units as defined by metadata published on a measurement topic`
`@@ -93,12 +93,11 @@ impl Units {`
`93`	`93`	`}`
`94`	`94`
`95`	`95`	`pub fn set_unit(&mut self, measurement: String, meta: serde_json::Value) {`
`96`		`- if let serde_json::Value::String(unit) = meta {`
`97`		`- // "Temperature": "°C"`
`98`		`- self.units.insert(measurement, unit);`
`99`		`- } else if let Some(unit) = meta.get("unit") {`
	`96`	`+ if let Some(unit) = meta.get("unit") {`
`100`	`97`	`// "Temperature": {"unit": "°C"},`
`101`		`- self.set_unit(measurement, unit.clone());`
	`98`	`+ if let serde_json::Value::String(unit_name) = unit {`
	`99`	`+ self.units.insert(measurement, unit_name.to_owned());`
	`100`	`+ }`
`102`	`101`	`} else {`
`103`	`102`	`// "Climate": { "Temperature": {"unit": "°C"}, "Humidity": {"unit": "%RH"} }`
`104`	`103`	`let group = measurement;`
`@@ -109,7 +108,7 @@ impl Units {`
`109`	`108`	`pub fn set_group_units(&mut self, group: String, meta: serde_json::Value) {`
`110`	`109`	`let units = Units::from_metadata(meta);`
`111`	`110`	`if !units.is_empty() {`
`112`		`- self.groups.insert(group, units);`
	`111`	`+ self.group_units.insert(group, units);`
`113`	`112`	`}`
`114`	`113`	`}`
`115`	`114`
`@@ -120,7 +119,7 @@ impl Units {`
`120`	`119`
`121`	`120`	`/// Retrieve the units to be used for a measurement group, if any`
`122`	`121`	`pub fn get_group_units(&self, group: &str) -> Option<&Units> {`
`123`		`- self.groups.get(group)`
	`122`	`+ self.group_units.get(group)`
`124`	`123`	`}`
`125`	`124`	`}`
`126`	`125`