Acquire bulk readings in JSON format#40
Conversation
540da76 to
9f1eb9c
Compare
794b714 to
a6b252b
Compare
a6b252b to
16556ff
Compare
16556ff to
5954952
Compare
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #40 +/- ##
==========================================
+ Coverage 78.53% 78.60% +0.06%
==========================================
Files 56 56
Lines 3024 3047 +23
==========================================
+ Hits 2375 2395 +20
- Misses 649 652 +3 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
510d24c to
c7e5b43
Compare
📝 WalkthroughWalkthroughAdds bulk JSON reading support: InfluxDBAdapter can accept list-of-dicts and uses a new write_single method; Grafana dashboard field collection accepts dict or list-of-dicts; tests for MQTT→InfluxDB and MQTT→Grafana bulk ingestion were added; CHANGES.rst updated. Changes
Sequence Diagram(s)sequenceDiagram
participant MQTT as MQTT Client
participant Adapter as InfluxDBAdapter
participant Writer as write_single
participant Influx as InfluxDB
participant Grafana as Grafana Dashboard
MQTT->>Adapter: publish payload (dict or [dict,...])
Adapter->>Adapter: validate payload type
alt payload is list
loop for each reading
Adapter->>Writer: write_single(meta, reading)
Writer->>Writer: prepare point (deepcopy/format)
Writer->>Influx: write point (with retry)
end
else payload is dict
Adapter->>Writer: write_single(meta, dict)
Writer->>Influx: write point
end
Grafana->>Adapter: request fields for dashboard
Adapter->>Adapter: compute unique keys from dict or list
Grafana->>Grafana: aggregate fields and provision dashboard
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~22 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
Comment |
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
kotori/daq/graphing/grafana/dashboard.py (1)
275-278:⚠️ Potential issue | 🔴 CriticalBug:
use_fieldwill fail whendatais a list.The
use_fieldhelper callsdata.get(field_name), which assumesdatais a dict. After the changes on lines 284-293,datacan now be a list, causingAttributeError: 'list' object has no attribute 'get'whenuse_fieldis called on lines 300 and 305.Consider adapting
use_fieldto handle both data types, or skip field-type filtering when data is a list (since the value type cannot be determined reliably across items).🐛 Proposed fix
def use_field(field_name: str): - value = data.get(field_name) - return isinstance(value, (float, int)) + if isinstance(data, dict): + value = data.get(field_name) + return isinstance(value, (float, int)) + elif isinstance(data, list): + # For lists, check if any item has a numeric value for this field + for item in data: + value = item.get(field_name) + if isinstance(value, (float, int)): + return True + return False + return False
🧹 Nitpick comments (2)
kotori/daq/graphing/grafana/dashboard.py (1)
292-293: UseTypeErrorfor invalid type validation.Per Python conventions,
TypeErroris preferred when an argument has an unexpected type. This aligns with the static analysis hint (TRY004).♻️ Proposed fix
else: - raise ValueError(f"Type of data {type(data)} not accepted") + raise TypeError(f"Unsupported data type: {type(data).__name__}")test/test_daq_grafana.py (1)
206-211: Consider using the existing helper for consistency.Other tests in this file use
grafana.get_datasource_names()(e.g., line 36). Using the same pattern here would improve consistency.♻️ Suggested refactor
# Proof that Grafana is well provisioned. logger.info('Grafana: Checking datasource') - datasource_names = [] - for datasource in grafana.client.datasources.get(): - datasource_names.append(datasource['name']) - assert settings.influx_database in datasource_names + assert settings.influx_database in grafana.get_datasource_names()
c7e5b43 to
3edaeaa
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In `@kotori/daq/storage/influx.py`:
- Around line 73-83: The write method currently returns None while write_single
returns a success flag; update write so that when data is a dict it returns the
boolean result of write_single(meta, data), and when data is a list it iterates
calling write_single(meta, item) and returns an aggregate boolean (e.g.,
all(results) to indicate every item succeeded, or any(results) if you prefer
at-least-one success) instead of None; adjust the error branch to keep raising
ValueError and ensure the return type is consistently a bool from write and
write_single.
In `@test/test_daq_mqtt.py`:
- Around line 45-76: The test test_mqtt_to_influxdb_json_bulk is order-dependent
because messages sent without timestamps can be returned in any order; change
the assertion to be order-independent by retrieving all records (via
influx_sensors.get_record or a method that returns all entries for
settings.mqtt_topic_json), remove the 'time' field from each record, and assert
that the set/list of resulting dicts equals the expected set [{u'temperature':
21.42, u'humidity': 41.55}, {u'temperature': 42.84, u'humidity': 83.1}] (or
alternatively sort by a stable key such as 'temperature' before comparing)
instead of assuming index 0/1 mapping to mqtt_json_sensor send order.
| def write(self, meta, data): | ||
| if isinstance(data, dict): | ||
| self.write_single(meta, data) | ||
| elif isinstance(data, list): | ||
| for item in data: | ||
| self.write_single(meta, item) | ||
| else: | ||
| raise ValueError(f"Type of data {type(data)} not accepted") | ||
|
|
||
| def write_single(self, meta, data): | ||
|
|
There was a problem hiding this comment.
Return aggregated success from write() to avoid silent None.
write() currently returns None for both dict and list inputs, which is a behavior change from write_single() returning a success flag. This can break callers that rely on a boolean outcome (e.g., logging/metrics). Consider returning the single result or an aggregate for lists.
💡 Suggested fix (aggregate success)
def write(self, meta, data):
if isinstance(data, dict):
- self.write_single(meta, data)
+ return self.write_single(meta, data)
elif isinstance(data, list):
- for item in data:
- self.write_single(meta, item)
+ results = []
+ for item in data:
+ results.append(self.write_single(meta, item))
+ return all(results)
else:
raise ValueError(f"Type of data {type(data)} not accepted")🧰 Tools
🪛 Ruff (0.15.0)
[warning] 80-80: Prefer TypeError exception for invalid type
(TRY004)
[warning] 80-80: Avoid specifying long messages outside the exception class
(TRY003)
🤖 Prompt for AI Agents
In `@kotori/daq/storage/influx.py` around lines 73 - 83, The write method
currently returns None while write_single returns a success flag; update write
so that when data is a dict it returns the boolean result of write_single(meta,
data), and when data is a list it iterates calling write_single(meta, item) and
returns an aggregate boolean (e.g., all(results) to indicate every item
succeeded, or any(results) if you prefer at-least-one success) instead of None;
adjust the error branch to keep raising ValueError and ensure the return type is
consistently a bool from write and write_single.
| @pytest_twisted.inlineCallbacks | ||
| @pytest.mark.mqtt | ||
| def test_mqtt_to_influxdb_json_bulk(machinery, create_influxdb, reset_influxdb): | ||
| """ | ||
| Publish multiple readings in JSON format to MQTT broker | ||
| and proof it is stored in the InfluxDB database. | ||
| """ | ||
|
|
||
| # Submit multiple measurements, without timestamp. | ||
| data = [ | ||
| { | ||
| 'temperature': 21.42, | ||
| 'humidity': 41.55, | ||
| }, | ||
| { | ||
| 'temperature': 42.84, | ||
| 'humidity': 83.1, | ||
| }, | ||
| ] | ||
| yield threads.deferToThread(mqtt_json_sensor, settings.mqtt_topic_json, data) | ||
|
|
||
| # Wait for some time to process the message. | ||
| yield sleep(PROCESS_DELAY_MQTT) | ||
|
|
||
| # Proof that data arrived in InfluxDB. | ||
| record = influx_sensors.get_record(index=0) | ||
| del record['time'] | ||
| assert record == {u'temperature': 21.42, u'humidity': 41.55} | ||
|
|
||
| record = influx_sensors.get_record(index=1) | ||
| del record['time'] | ||
| assert record == {u'temperature': 42.84, u'humidity': 83.1} |
There was a problem hiding this comment.
Avoid order-dependent assertions when timestamps can collide.
Both readings are sent without timestamps, so InfluxDB may return them in nondeterministic order if they share the same second. This can make the test flaky. Prefer order-independent verification or add explicit timestamps.
✅ Order-independent assertion (example)
- record = influx_sensors.get_record(index=0)
- del record['time']
- assert record == {u'temperature': 21.42, u'humidity': 41.55}
-
- record = influx_sensors.get_record(index=1)
- del record['time']
- assert record == {u'temperature': 42.84, u'humidity': 83.1}
+ records = [influx_sensors.get_record(index=0), influx_sensors.get_record(index=1)]
+ for record in records:
+ del record['time']
+ records.sort(key=lambda r: r['temperature'])
+ assert records == [
+ {u'temperature': 21.42, u'humidity': 41.55},
+ {u'temperature': 42.84, u'humidity': 83.1},
+ ]🧰 Tools
🪛 Ruff (0.15.0)
[warning] 47-47: Unused function argument: machinery
(ARG001)
[warning] 47-47: Unused function argument: create_influxdb
(ARG001)
[warning] 47-47: Unused function argument: reset_influxdb
(ARG001)
🤖 Prompt for AI Agents
In `@test/test_daq_mqtt.py` around lines 45 - 76, The test
test_mqtt_to_influxdb_json_bulk is order-dependent because messages sent without
timestamps can be returned in any order; change the assertion to be
order-independent by retrieving all records (via influx_sensors.get_record or a
method that returns all entries for settings.mqtt_topic_json), remove the 'time'
field from each record, and assert that the set/list of resulting dicts equals
the expected set [{u'temperature': 21.42, u'humidity': 41.55}, {u'temperature':
42.84, u'humidity': 83.1}] (or alternatively sort by a stable key such as
'temperature' before comparing) instead of assuming index 0/1 mapping to
mqtt_json_sensor send order.
Coming from #39, this finally implements the acquisition of bulk readings in JSON format.
An example payload is:
[ { "time": 1614534635, "temperature": 45.2, "humidity": 83.1 }, { "time": 1614534660, "temperature": 45.3, "humidity": 83.2 } ]cc @valentinbarral