Raw/Derived data feed redistribution mechanism

[webbpinner]

Attempt to re-create the SCS Message functionality currently missing within OpenRVDAS.

End Goal:
Allow users to rebroadcast one or more raw and/or custom data feeds to UDP/TCP/Serial/Websocket destinations without needing to modify the primary logger_config.yaml file.

Possible Scenarios:

• Ship needs a real-time, unmolested copy of SSV data sent to a serial port on the MBES DAQ workstation

• Science party needs a real-time, unmolested copy of ONLY position/heading NMEA messages (GGA, HDT) forwarded to a 3rd-party sensor/system via direct UDP. The copy should only include the GGA and HDT messages and NOT the VTG, GLL, ZDA, RMS messages contained in the raw feed from the INS.

• Ship needs a UDP feed containing the GGA messages from a GPS receiver and the HDT from a Gyro multiplexed into a single data feed sent via UDP broadcast (192.168.0.255).

• Science party needs a NMEA DBT formatted message built from a corrected/calibrated depth value stored within the CDS sent via serial port to a science-supplied workstation.

• Ship needs GGA messages from a GPS sent to multiple discrete UDP destination addresses.

• Web-based clients needs access to raw/derived data feeds via websocket.

[davidpablocohn]

Thanks for this. Clarification: in terms of "unmolested" data - do you think users will want raw, raw, pre-timestamped strings? Or a tap somewhere in the logger pipeline - timestamped, filtered for some message type? Both are challenging to do cleanly from a software architecture perspective, but the challenges are rather different.

…

On Thu, Dec 4, 2025 at 10:51 AM Webb Pinner ***@***.***> wrote: *webbpinner* created an issue (OceanDataTools/openrvdas#480) <#480> Attempt to re-create the SCS Message functionality currently missing within OpenRVDAS. End Goal: Allow users to rebroadcast one or more raw and/or custom data feeds to UDP/TCP/Serial/Websocket destinations without needing to modify the primary logger_config.yaml file. Possible Scenarios: - Ship needs a real-time, unmolested copy of SSV data sent to a serial port on the MBES DAQ workstation - Science party needs a real-time, unmolested copy of ONLY position/heading NMEA messages (GGA, HDT) forwarded to a 3rd-party sensor/system via direct UDP. The copy should only include the GGA and HDT messages and NOT the VTG, GLL, ZDA, RMS messages contained in the raw feed from the INS. - Ship needs a UDP feed containing the GGA messages from a GPS receiver and the HDT from a Gyro multiplexed into a single data feed sent via UDP broadcast (192.168.0.255). - Science party needs a NMEA DBT formatted message built from a corrected/calibrated depth value stored within the CDS sent via serial port to a science-supplied workstation. - Ship needs GGA messages from a GPS sent to multiple discrete UDP destination addresses. — Reply to this email directly, view it on GitHub <#480>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AFO7V3QH5BMXMTP6VMJXE434AB7BNAVCNFSM6AAAAACOCIFSR2VHI2DSMVQWIX3LMV43ASLTON2WKOZTGY4TMMBSGY4TGNY> . You are receiving this because you are subscribed to this thread.Message ID: ***@***.***>

[webbpinner]

great question... if you asked the community I'm sure they would say "both". However I think 90% of the needs could be met by just exposing the output from the reader classes.

My preference would be that this new mechanism allows users to build independent loggers-like pipelines that can read, filter and mutate the incoming raw data as needed by the output destination(s).

Here's a more in-depth example:
OpenRVDAS is setup to log GGA, VTG, HDT, RMS, ZDA from a Seapath380 to CDS, InfluxDB and logfiles.

The science party brings on a device that needs copies of just the GGA, HDT feeds. They don't tell the ship until day 3 of the cruise. The ship doesn't want to modify their logger_config.yaml mid-cruise and risk data gaps while modifying and debugging the logger to include a UDPWriter.... which might involve refactoring composed writers depending on how the logger is setup.

Second scenario...
The science party wants a GGA and HDT for an ROV. The GGA comes from the USBL but the heading comes from the ROV gyro and is NOT in HDT format. Ideally there would be a way to multiplex the raw feed from the USBL and a HDT formatted version of the ROV Gyro feed.

...and before we get too far, I'd also like to fresh out the config file syntax that ships will use to define these redistributed feeds.

[davidpablocohn]

Okay, I've had a good long think on this, as it completely throws a wrench in the traditional architecture. The way I explained it to my wife is: "Imagine you've built a railway. Now, customers want to be able to get off, not only at the stations, but anywhere in between. And you can't slow the train down, and you have to use the same rails and cars as before."

I think I've got an idea on how to do it cleanly and efficiently (that's why I get the big bucks, right?)

But before I dive into exploring that, could you clarify "I'd also like to fresh out the config file syntax that ships will use to define these redistributed feeds."

[webbpinner]

I'm just asking for the interface... how are users going to define their data stream redistributions?

...and based on what I'm hearing from users, it needs to be a separate file from the logger_config.yaml. Users don't want to mess with working data logging configs during a cruise. They'd prefer it's an independent configuration.

[davidpablocohn]

Okay - that's a tall order. It would be easy to do in a hacked-together manner that could come back to bite us, but I'm going to have to put my head down to figure out a clean way to go about it.

…

On Sun, Dec 7, 2025 at 2:10 PM Webb Pinner ***@***.***> wrote: *webbpinner* left a comment (OceanDataTools/openrvdas#480) <#480 (comment)> I'm just asking for the interface... how are users going to define their data stream redistributions? ...and based on what I'm hearing from users, it needs to be a separate file from the logger_config.yaml. Users don't want to mess with working data logging configs during a cruise. They'd prefer it's an independent configuration. — Reply to this email directly, view it on GitHub <#480 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AFO7V3SICSKMCCNZ7KRHEFL4ASQVHAVCNFSM6AAAAACOCIFSR2VHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZTMMRTGU3DKNZYGY> . You are receiving this because you commented.Message ID: ***@***.***>

[webbpinner]

Would it be possible to just write the raw string to the CDS using the logger name as the key? Then the second file would contain loggers that start with a CDSReader.

This approach would give the user the options to pull raw data and/or parsed values from the CDS to format and redistribute as needed.

Let's schedule a call to talk through this some more. My descriptions may be confusing.

[davidpablocohn]

The question is how to seamlessly "tap" readers, and how to specify which readers should be tapped.

…

On Mon, Dec 8, 2025 at 7:52 AM Webb Pinner ***@***.***> wrote: *webbpinner* left a comment (OceanDataTools/openrvdas#480) <#480 (comment)> Would it be possible to just write the raw string to the CDS using the logger name as the key? Then the second file would contain loggers that start with a CDSReader. This approach would give the user the options to pull raw data and/or parsed values from the CDS to format and redistribute as needed. Let's schedule a call to talk through this some more. My descriptions may be confusing. — Reply to this email directly, view it on GitHub <#480 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AFO7V3TEYXHQFXANNOPDBKL4AWNDNAVCNFSM6AAAAACOCIFSR2VHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZTMMRXGY2DGNZXHA> . You are receiving this because you commented.Message ID: ***@***.***>

[webbpinner]

Could you add an option to the ReaderClass for the CDS server ip/port?

Only forward to CDS if var != None.

[davidpablocohn]

I'm hesitant to hard wire it to CDS. There's also the question of how you'd identify which reader a feed was coming from without tweaking/preprocessing the main config file.

I'm appending my initial idea below of using a superclass method to allow installing a "tap" into any reader or transform that would shunt a copy of its output to an arbitrary writer. It could be defined in the cruise config file itself, or in a separate yaml file that could be applied by preprocessing the cruise config, much in the way we apply templates:

Basic idea is to add an optional "tap" argument to Readers and Transforms that allows specifying that, in addition to returning their normal output, it gets shunted off to a Writer (including, if we wanted to get medieval, a composed writer).

There are a few ways we could do this that would be more or less general and/or succinct:

readers:
  - class: SerialReader
    args:
      port: /dev/ttys0
      ...
      tap:
        class: UDPWriter
        args:
          port: 6224

Or more compactly, but less generally:

readers:
  - class: SerialReader
    args:
      port: /dev/ttys0
      ...
      udp_tap: 6224

and similarly with websocket_tap, textfile_tap and logfile_tap.

It could be implemented by adding **kwargs to the end of all Reader and transform arg lists, and calling super().init(**kwargs) at the end of their inits, which would pass all unrecognized args to the superclass Reader and Transform.

Then right before the return record line, a line could be added along the lines of superclass_process(record) or some such.

The specific readers and transforms could be completely agnostic about what that "process" was. The base classes could interpret the arguments it gets from **kwargs (whether tap or udp_tap or whatever) and create the appropriate writer. When their superclass_process() was called, they could send the passed record off to that writer. NOTE: prob best to do asynchronously, dropping it in a queue to be read by a separate thread so it can return quickly and not block if the writer is slow or problematic.

Problem: how to allow tapping without messing with the cruise config file?

A possible solution: create a separate "tap.yaml" file to specify which components of which loggers should be tapped.

(config_name, component): {class: ..., kwargs: ...}

We could create (yet) another pre-processing script that would look for a "tap" file and insert the taps in the appropriate loggers. But it would require some careful architecting to make sure it's well behaved on corner cases (e.g. a single config can have multiple readers of the same class), concise and sufficiently expressive. And how will it interact with templated loggers?

[webbpinner]

I like the tap concept and being able to "tap" at any point in the logger config but I'm struggling to come up with strong use-cases where having taps beyond the initial reader are necessary.

I also get the not wanting to dump stuff into the CDS but JUST FOR BRAIN STORMING PURPOSES I'm going to use the CDS cache the raw feeds in the following example. This example uses the new default value syntax

Here's what I'd want my logger_config.yaml like:

cruise:
  id: <<cruise>>
  start: '<<cruise_start>>'
  end: '<<cruise_end>>'

###########################################################
# Global variables - can be overridden by individual loggers
variables:
  cruise: ODT2501
  cruise_start: '2025-12-01'
  cruise_end: '2025-12-31'

  parser_udp_port: 6224

  file_root: /var/tmp/log
  parse_definition_path: local/devices.yaml

  data_server: localhost:8766
  influx_bucket_name: openrvdas

###################
config_templates:
  #################
  udp_to_file_config_template: &udp_to_file_base
    readers: 
    - class: UDPReader
      kwargs:
        port: <<udp_port>>
        eol: "<<eol|'\r\n'>>"
        tap: <<tap|null>>
    transforms:
    - class: SplitTransform
      module: logger.transforms.split_transform
    - class: TimestampTransform
      kwargs:
        sep: ','
    writers:
    - class: LogfileWriter
      kwargs:
        filebase: <<filebase>>
        split_char: ','
        suffix: '.txt'

  udp_to_net_config_template:
    <<: *udp_to_file_base
    writers:
    - class: ComposedWriter
      kwargs:
        transforms:
        - class: PrefixTransform
          kwargs:
            prefix: <<logger>>
        writers:
        - class: UDPWriter
          kwargs:
            port: <<parser_udp_port>>

  udp_to_file+net_config_template:
    <<: *udp_to_file_base
    writers:
    - class: LogfileWriter
      kwargs:
        filebase: <<filebase>>
        split_char: ','
        suffix: '.txt'
    - class: ComposedWriter
      kwargs:
        transforms:
        - class: PrefixTransform
          kwargs:
            prefix: <<logger>>
        writers:
        - class: UDPWriter
          kwargs:
            port: <<parser_udp_port>>

###################
logger_templates:
  #################
  udp_logger_template:
    configs:
      'off': {}
      file:
        config_template: udp_to_file_config_template
        variables:
          udp_port: <<udp_port>>
          tap: <<tap|null>>
          filebase: <<file_root>>/raw/<<cruise>>_<<logger>>
      net:
        config_template: udp_to_net_config_template
        variables:
          udp_port: <<udp_port>>
          tap: <<tap|null>>
      file+net:
        config_template: udp_to_file+net_config_template
        variables:
          udp_port: <<udp_port>>
          tap: <<tap|null>>
          filebase: <<file_root>>/raw/<<cruise>>_<<logger>>

  parse_data_logger_template:
    configs:
      'off': {}
      'on':
        readers:
        - class: UDPReader
          kwargs:
            port: <<parser_udp_port>>
        transforms:
        - class: ParseTransform
          kwargs:
            metadata_interval: 10
            definition_path: <<parse_definition_path>>
            record_format: '{data_id:w} {timestamp:ti},{field_string}'
            return_json: true
        writers:
        - class: CachedDataWriter
          kwargs:
            data_server: <<data_server>>
        - class: InfluxDBWriter
          kwargs:
            bucket_name: <<influx_bucket_name>>

###########################################################
loggers:
  gps01:
    logger_template: udp_logger_template
    variables:
      udp_port: 16001
      tap: <<data_server>>/<<logger>>

  gps02:
    logger_template: udp_logger_template
    variables:
      udp_port: 16002

  parse_data:
    logger_template: parse_data_logger_template

###########################################################
modes:
  'off':
    [gps01-off, parse_data-off]

  no_write:
    [gps01-net, parse_data-on]

  write:
    [gps01-net+file, parse_data-on]

Then for my redistribution yaml:

###########################################################
loggers
  gps01:
    'off': {}
    on:
      readers:
      - class: CachedDataReader
        kwargs:
          data_server: localhost:8766
          subscription:
          fields:
            gps01:
              seconds: 0
      transforms:
      - class: FormatTransform
        module: logger.transforms.format_transform  # where the definition is
        kwargs:
          format_str: '{gps01}'
      writers:
      - class: UDPWriter # send everything
        kwargs:
          destination: 192.168.4.42
          port: 20021
      - class: ComposedWriter  # send only GGA
        kwargs:
          transforms:
          - class RegexFilterTransform
            kwargs:
              pattern: 'GGA'
          writers:
          - class: UDPWriter
            kwargs:
              destination: 192.168.4.35
              port: 16011

###########################################################
modes:
  'off':
    [gps01-off]

  on:
    [gps01-on]

If there's an additional request for GPS data, the user only has to add it to the redistribution yaml. This examples also shows how to opt-in to the mechanism through the: tap: <<data_server>>/<<logger>> and underlying tap: <<tap|null>> config syntax