TimescaleDB bridge

[BuxtonCalvin]

I'm writing a TimescaleDB bridge and have it working pretty well. I'm refining some of the compression and rollup features and once I get those going, I'll test it for a couple of weeks before submitting a pull request. Due to the many features of the database, it turned into something a little more complicated than what I first expected, and with complexity comes fragility. With this in mind, I have three ideas that I hope you can help me with. First, I found tracking down the movement of data in the program was somewhat difficult to follow as there are not a lot of comments describing how metric data and also device data moves through the class structures. And given that the data is the focus of the app, I believe it would be helpful to see more verbose descriptions in the classes concerning data flow--in my case just so I'm 100% on what's what with the bridge I wrote.

The second idea is that I believe there needs to be more separation between transports and bridges in the structure of the program. For example, there are many references to "write", "read", "connect" etc. where it's difficult to tell, at first, if you're referring to a transport that is connecting to an inverter, or to a bridge obtaining data produced from a transport, or a bridge connecting to an endpoint of some type--like a database. I realize that technically, these are all transports, but I believe it would be a lot clearer if you could re-factored "naming" to something like "transport_write", "bridge_write" "transport_connected", "bridge_connected" etc. As I said, I have a TDB bridge working, but it took some effort to understand what was what in the classes, and to be honest, I'm still not sure if I understand the data flow.

Finally, for us Docker users, could you create a config folder wherein all the user configurable files can sit. This would make it much simpler creating volumes if the variable mask, config, maps and any other configuration files were grouped together in one spot. I would imagine that the config folder would hold example placeholder files that would in turn be referenced programmatically within PPG: currently, these files are referenced in their respective folders. The Docker volume would then rename to whatever specific map, cfg etc that is in the local folder.

Other than all of that, a really nice job on this gateway. I think it has a lot of potential as there are numerous endpoints and inverters out there that someone may wish to connect to. I may even take a crack at a Prometheus bridge. Thx again for this app.

[HotNoob]

Sounds awesome.

Yes, documentation and what not is the weak point of ppg... and conceptially it's a bit difficult to explain.

https://github.com/HotNoob/PythonProtocolGateway/blob/v1.1.11/documentation/usage/configuration_examples/modbus_rtu_to_mqtt.md

i have an image here, a bit hard to find, that shows a bit about the flows. reading / writing.
generally speaking, reading is data from device to user, and writing is data from user to device.
since communication is bidirectional its a bit confusing.
it's also sometimes confusing to myself :P

the mixing of active and passive protocols...
ie, modbus, we actively "read" the registers
canbus, we wait for the devices to send data to ppg
mqtt, we use the subscribe topic to wait for the data to be sent to ppg... which is in a sense "reading" the data to be "written" to the bridge...
on an intuition level it seems a bit backwards. for mqtt :P

overall, the inter-transport behaviour is that everything is broadcasted to every other device on the same bridge...

Yeah... i confuse myself just thinking about it as well :P
yeah... maybe its backwards for mqtt... future me has to think more about that...

---

yes for the config, i think its not too difficult to add a /config folder without breaking things.
ppg could be made to check / and /config, for backwards compatability.

PythonProtocolGateway/Dockerfile

Line 11 in 29e5e21

COPY config.cfg /app/

and then /config could be added to the dockerfile.

[BuxtonCalvin]

I think the PPG documentation is fine. It was more than enough to get me started. What I had trouble with was sparse comments in the transport classes. If you could add some program flow comments to the functions within the classes to how the data moves within PPG, that would be a great help. For example the write_enabled flag is a bit confusing as I took it to mean writing or not writing back to the inverter. But there is some program logic that seems to stop other routines if it is set to false. Or maybe not, as it was hard for me to tell what was in scope and when, with the flag. Anyway, I do have TimescaleDB working, I'd just like to understand 100% of the program flow to track down inevitable bugs as they pop up. Thx

[BuxtonCalvin]

One other thing I hope you can help with. I currently batch together the various scrapes that populate "data" within the write_data method, for appending to a timescaledb table. For the MQTT bridge, this isn't important as it's data in and then data out. But for timescaledb, it's important to bundle all the register reads, in the given loop, so that all read metrics can be inserted as a single row in a table. I do this now by copying the inverter transports read_interval to a setting that controls the append of a given batch to the timescaledb table. However, this doesn't ensure that all the data within a particular inverter register read loop is captured in the same batch due to timing differences.

What would help here is if you could create an attribute that would perhaps be called ReadComplete that would initialize as false, and then go true at the end of the final register scrape in the inverter read loop. After a couple of seconds the flag would then go false again. I could use a callback of sorts to monitor the flag, where the state change would trigger my batch insert. Going off of the flag, I would capture all the data of a particular read loop without having to consider any timing discrepancies. My apologies if this functionality already exists, but I couldn't find anything in the code with a similar function.

[HotNoob]

hmm. i think in the canbus transport, i solve a similar issue to this by utilizing a cache.

PythonProtocolGateway/classes/transports/canbus.py

Line 43 in 29e5e21

cache : OrderedDict [int,(bytes, float)] = None

you could do something similar for your timescaledb transport.
just instead of a "read cache" you could implement a "write cache"

cache the data until a row is completed; flush the cache or just reuse the data in the cache shrugs

this way your timescaledb transport can handle all of that; wait for the cache to fill up before actually writting.

maybe a little bit more complicated, but it woud provide better isolation between the classes / transports.

[BuxtonCalvin]

Yes, I'm doing that now with appends to a temp dict as the scraped data is small enough to not hit performance. It's the "wait for the cache to fill up" that is unknown as you can't be certain when the cache is full. A state flag would definitively signal that the cache is full as the flag state would be based on the read loop close.

[HotNoob]

well... the read times can be variable (not everything is read every batch) - per register timing.

the "cache" in the case of canbus just holds one copy. it's not the most precise. different kind of cache.
fill the blanks with old data sort of thing.

then the actual read/writes are performed at an interval.

[HotNoob]

over simplified example:

cache :  OrderedDict [str,(str, float)] = None 

def write_data(self, data : dict[str, str], from_transport : transport_base):
    for key in data:
        self.cache[key] = (data[key], time.time())

    required_keys = ["a", "b", "c"]

    if not all(k in self.cache for k in required_keys):
        return #wait for all data

    #do write stuff here, use self.cache instead of data

[BuxtonCalvin]

Yes, I thought about pulling the filtered list of keys from the registry map-- I do that anyway to build the Timescaledb wide table columns dynamically, but that approach doesn't really guarantee where the underlying inverter data scrape starts and ends. This isn't a big problem with 10s poll loops (well for me anyway), but if you have a setup that polls every 10 minutes, then your data could get malformed quickly, as the entire table row is inserted from the cache with a single timestamp. The other way I considered is to insert the rows as the data comes in, but that can create enormous bloat in the transaction table as your overall scrape may be hundreds of metrics, sliced into any number of reads, meaning any number of rows for what really should be a single row. And the single row approach is what works best with the TimescaleDB compression algorithms.

What is the very first method in the modbus base class that starts a given inverter poll loop? I could at least read a start attribute there, which would trigger the start of a comparison loop in the TS class, wherein the TS read/write would wait until all the data is there. But it's klugey and has considerable associated loop overhead. Though frankly, if you have a start loop attribute, why not then flip the attribute at the end of the overall scrape loop and avoid all the headache.....

[HotNoob]

PythonProtocolGateway/classes/transports/modbus_base.py

Line 194 in 29e5e21

ranges = self.protocolSettings.calculate_registry_ranges(self.protocolSettings.registry_map[registry_type],

everything is dynamic, so there isnt exactly a "start of the loop" sort of thing.

i'm not sure what sort of table structure is done. hard to say how to handle the data destortion.
if it can be structured so that every item has a timestamp, than its not so hard :P but obviously heavier on the db.

maybe a summary table & a table for each variable?
i only know mysql :P shrugs

if each table holds each variable, could check if a value changes before inserting. at least keeping the db smaller.
not sure how strict your table structure requirements are.

or a table with variable name as a column.

---

what about null values? if the info is stale? can timescaledb handle partial rows \w nullables?

---

could implement all 3 table structures.
table_structure = "narrow" # "wide" | "narrow" | "multitable"
a lot of extra work :P

[BuxtonCalvin]

yeah, the metrics columns (based on variable name) are nullable, timestamps are not, though I do trap for nulls as there really shouldn't be any. If the info is stale, then I imagine something is wrong with the scrape with errors showing in the log. I could do a 1 hour period comparison of all values to flag stale data, perhaps in a separate thread, and I've considered incorporating some sort of messaging like pushover, but right now just want to get aligned data into the wide table format.

I'm currently implementing narrow and wide tables, with variable names as wide table columns, because there are different SQL use cases for each, with not a huge amount of extra disk space use for the double tables.
I'm assuming from log entries that, where there is a filter match, registers are read sequentially starting at 1 (or 0) in blocks of whatever your batch read settings are, and then finishing at max registers. Would it be possible to listen for register 1's batch group, set a flag, then listen for max register (prior to restarting at 1) and then flip the flag. That should give me the parameters for a cache of aligned data. BTW, what is the easiest method to pull an init list of final, safe, filtered variable names.

[BuxtonCalvin]

Here's a diagram of what I'm doing:

`┌──────────────────────────────────────────────────────────────────────────────┐
│ 1. Gateway Startup │
└──────────────────────────────────────────────────────────────────────────────┘
protocol_gateway.py
|
|-- load config
|-- dynamically import "classes.transports.modbus_tcp"
|-- dynamically import "classes.transports.timescaledb_out"
|
|-- instantiate modbus_tcp( cfg )
|-- instantiate timescaledb_out( cfg )
|
|-- timescaledb_out.init()
|
|-- start flush thread
|-- start autorefresh thread
|-- does NOT connect yet (connect() called later)

┌──────────────────────────────────────────────────────────────────────────────┐
│ 2. Modbus Polling Loop (Separate Thread) │
└──────────────────────────────────────────────────────────────────────────────┘
modbus_tcp._poll_worker()
|
|-- poll device → decode registers → build "data packet" dict
e.g. { ts: 1710000000, pointName: "V1", value: 234.1 }
|
|-- send packet to ProtocolGateway.on_message()

┌──────────────────────────────────────────────────────────────────────────────┐
│ 3. Bridge Routing │
└──────────────────────────────────────────────────────────────────────────────┘
protocol_gateway.on_message()
|
|-- find output transports where:
output.match_name == bridge_target
(ex: "transport.timescaledb")
|
|-- call:
timescaledb_out.write( data_dict_from_modbus )

┌──────────────────────────────────────────────────────────────────────────────┐
│ 4. Timescale Output: Write → Queue │
└──────────────────────────────────────────────────────────────────────────────┘
timescaledb_out.write()
|
|-- convert Modbus message → internal MetricPoint structure
|
|-- put MetricPoint into batch_queue
(thread-safe queue.Queue())

┌──────────────────────────────────────────────────────────────────────────────┐
│ 5. Flush Worker Thread │
└──────────────────────────────────────────────────────────────────────────────┘
timescaledb_out._flush_worker() (always running, daemon thread)
LOOP:
|
|-- dequeue up to batch_size points
|
|-- call:
_flush_batch_wide( points )

┌──────────────────────────────────────────────────────────────────────────────┐
│ 6. Wide Table Writer (device_metrics_wide) │
└──────────────────────────────────────────────────────────────────────────────┘
timescaledb_out._flush_batch_wide()
|
|-- ensure table exists: device_metrics_wide
|
|-- detect unknown metric keys
|-- ALTER TABLE ADD COLUMN IF NOT EXISTS float8
|
|-- build row:
device_id = <bridge/device identifier>
ts = timestamp of sample
= value
= value
|
|-- INSERT INTO device_metrics_wide( device_id, ts, metric... )
|
|-- COMMIT

┌──────────────────────────────────────────────────────────────────────────────┐
│ 7. On Error → Backlog + Reconnect │
└──────────────────────────────────────────────────────────────────────────────┘
if INSERT/connection fails:
|
|-- _add_to_backlog() for each failed point
|-- _trigger_reconnect()
|
|-- spawn reconnect thread (only one allowed)
|
|-- reconnect sequence:
connect()
flush_backlog()
resume normal flush loop

┌──────────────────────────────────────────────────────────────────────────────┐
│ 8. Data Ends Up in TimescaleDB Wide Table │
└──────────────────────────────────────────────────────────────────────────────┘
device_metrics_wide
id bigserial PK
device_id text
ts timestamptz
voltage float8
current float8
temp float8
...
`

[BuxtonCalvin]

Here's a png diagram

[HotNoob]

somewhat correct.

the modbus transport doesnt send any data to other transports until its completely done, so you shouldnt have to worry about batch start/stop. that's only effecting the modbus transport itself.

yeah its a little messy

PythonProtocolGateway/classes/transports/modbus_base.py

Line 198 in 29e5e21

registry = self.read_modbus_registers(ranges=ranges, registry_type=registry_type)

PythonProtocolGateway/classes/transports/modbus_base.py

Line 209 in 29e5e21

return info

everything is returned outside of the batch read loop.

if your seeing "partial" data results, that'll be from the variable timing, depending on the protocol, not everything is read every time.

if its really problematic for timescaledb \w widemode, we could add a "variable_timing=off" sort of setting.
and apply it around here:

PythonProtocolGateway/classes/protocol_settings.py

Line 831 in 29e5e21

if init: #add but do not update timestamp; can maybe rename init to no timestamp at this point

---

and example of accessing transport_settings from protocol_settings class:

PythonProtocolGateway/classes/protocol_settings.py

Line 408 in 29e5e21

transport_read_interval = self.transport_settings.getint("read_interval", transport_read_interval)

---

just worry about data in/out within timescaledb class.

transport.protocolSettings.registry_map should contain the complete list.

the variable masks and screens are applied when its loaded.

PythonProtocolGateway/classes/protocol_settings.py

Line 778 in 29e5e21

if self.variable_mask:

from within a transport, after transport_base has been initialized:
super().__init__(settings)
self.protocolSettings.registry_map

dict of registry_map_entry's - attribute variable_name

if you want, you can add a get_variable_names helper function to the protocol_settings class.
i could see that coming in handy.

[HotNoob]

its best to design everything so that data / partial data can be handle randomly, since modbus isnt the only protocol that sends partial results.

the variable timing is to help address speed limitations with modbus. ie, read important / often changing variables more often.