Downloading logged packets from a script

[hmaclachlan]

Hello
I am looking to automate processing and downloading packets from the Data Extractor so it can be performed within scripts that are being ran, based off of variables defined within that script (ie. start and end time). I am using the method described in Issue 937 (#937) and adding the items to a .csv, but it takes a while to add all rows of telemetry. One of the lines in that script references not using this method for large queries. Is there a better way to do this, for large queries? I am unable to locate any code that details how the Data Extractor performs on the backend. I tried using batch/parallel processing methods to speed up this process, but COSMOS does not seem to recognize the 'concurrent' library. Here is what I am currently working with:

def download_tlm(tlm_items, test_start, test_end, headers, filepath)
  OpenC3::StreamingWebSocketApi.new do |api|
    api.add(items: tlm_items)
  end
  
  data = OpenC3::StreamingWebSocketApi.read_all(items: tlm_items, start_time: test_start, end_time: test_end) # Warning this saves all data to RAM. Do not use for large queries
  
  csv_content = CSV.generate do |csv|
    csv << headers
    data.each do |entry|
      row = [entry[tlm_items[0]]] + tlm_items[1..-1].map { |item| entry[item] }
      csv << row
    end
  end
  
  put_target_file(filepath, csv_content)
  download_file(filepath)
end

[jmthomas]

#937 is how you do it. You can use the read_all method but just be aware of how many items you're trying to read over the time period you select. It will pull all the items into memory before writing them to a file so you just don't want to pull thousands of items over days.

[hmaclachlan]

@jmthomas are there any plans to create a method that would work for pulling large amounts of telemetry items? I am looking for some scripting functionality that would do the equivalent of pressing the "Process" button in the Data Extractor (which downloads large amounts of telemetry instantly). I need to be able to process and download telemetry items from a packet that collects every tenth of a second and runs for 3 hours. The process above of putting all items into memory and adding each entry to a .csv takes forever. If there is not an option to perform the "Process" functionality of the Data Extractor within a script, I will resort to doing it manually, but am wondering if I am missing something here

[jmthomas]

I would recommend using the StreamingAPI with your test's start and end time like so (works in the COSMOS Demo):

start_time = Time.now.to_nsec_from_epoch
wait 10 # wait for your test
end_time = Time.now.to_nsec_from_epoch

file = Tempfile.new('test')
OpenC3::StreamingWebSocketApi.new do |api|
  api.add(items: ['DECOM__TLM__INST__HEALTH_STATUS__TEMP1__CONVERTED', 'DECOM__TLM__INST__HEALTH_STATUS__TEMP2__CONVERTED'], start_time: start_time, end_time: end_time)
  file.write api.read
end
put_target_file("INST/test.txt", file)

[hmaclachlan]

@jmthomas when I execute this against the COSMOS Demo, the put_target_file line runs but does not populate the "INST/test.txt" with anything. But what is odd is that when I download that file after, the downloaded file is populated. This is still an issue, however, as I need to later be able to pull from that file for further manipulation of the data. I am working in 5.19.0. Why is the file not populating in targets_modified for INST?

start_time = Time.now.to_nsec_from_epoch
wait 10 # wait for your test
end_time = Time.now.to_nsec_from_epoch

file = Tempfile.new('test')
OpenC3::StreamingWebSocketApi.new do |api|
  api.add(items: ['DECOM__TLM__INST__HEALTH_STATUS__TEMP1__CONVERTED', 'DECOM__TLM__INST__HEALTH_STATUS__TEMP2__CONVERTED'], start_time: start_time, end_time: end_time)
  file.write api.read
end
put_target_file("INST/test.txt", file)
download_file("INST/test.txt")

[jmthomas]

Before calling put_target_file try to rewind the tempfile first:

file.rewind
put_target_file("INST/test.txt", file)
download_file("INST/test.txt")

[hmaclachlan]

@jmthomas that worked, thanks!

[hmaclachlan]

@jmthomas Thanks for your guidance with this. I actually have a few follow up questions:

1. I notice that when I have more than 8 telemetry items defined for 'items' in the api.add, the file populates with hashes (ie. a string of [#Hash:66960, #Hash:66980, #Hash:67000, #Hash:67020]). Is the OpenC3::StreamingWebSocketApi only able to input a maximum of 8 telemetry items?

2. Is there a limit regarding the number of lines returned per telemetry request within this API? After performing some json post processing of the tempfile output, it seems as though the there is always a limit of 100 lines, no matter how long the timeframe defined is.

3. In looking into the OpenC3::StreamingWebSocketApi code examples, I see you can define a whole packet as opposed to list of telemetry items. When trying this with raw and decom versions of the same packet, however, I do not get internal telemetry items for the packet. Here is an example output:

                [{"__type"=>"PACKET", "__packet"=>"RAW__TLM__GSE__GSE", "__time"=>1732228901959368600, "buffer"=>"CAoJtAD9VGolRfUAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA\nAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAQBAPADEm6Xj\nVD/05WBBiTdMQDwAAAAAAABAFAAAAAAAAEBAAAAAAAAAAAAAAAAAAABSRU1P\nVEUAAAAAAAAAAAAAAAAAACswLCJObyBlcnJvciIAAAAAAAAALCwsLCwAAAAA\nAAAAAAAAAAAAAAABAQAAAQEAAAEAAEFjdGl2ZSBDb250cm9sbGVyOiBDT1NN\nT1MAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=\n"}, {"__type"=>"PACKET", "__packet"=>"RAW__TLM__GSE__GSE", "__time"=>1732228902962954200, "buffer"=>"CAoJtQD9VGolRvUAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA\nAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAQBAPADEm6Xj\nVD/0sCDEm6XjQDwAAAAAAABAFAAAAAAAAEBAAAAAAAAAAAAAAAAAAABSRU1P\nVEUAAAAAAAAAAAAAAAAAACswLCJObyBlcnJvciIAAAAAAAAALCwsLCwAAAAA\nAAAAAAAAAAAAAAABAQAAAQEAAAEAAEFjdGl2ZSBDb250cm9sbGVyOiBDT1NN\nT1MAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=\n"}, {"__time"=>1732228901959368600, "DECOM__TLM__GSE__GSE"=>nil, "__type"=>"ITEMS"}, {"__time"=>1732228902962954200, "DECOM__TLM__GSE__GSE"=>nil, "__type"=>"ITEMS"}]
   

4. Given the issues above, is there any additional method to downloading all telemetry packets for a target within a script (based on a timeframe defined in that script)? Or is the best way to do this truly via manual user inputs in the Data Extractor.

[jmthomas]

1. Wow I'm glad we kept pulling on this thread because this is a pretty big bug. We have code that truncates a hash output if there are more than 10 items. When you do a api.add we always include the __time and __type so that leaves 8 more before we do the [#<Hash:21280>, ... thing. Thank you for finding this!

2. I gave you a bad example. The batch size is 100 but you just have to keep calling api.read until it stops returning data. A better implementation is:

end_time = Time.now.to_nsec_from_epoch
start_time = (Time.now - 600).to_nsec_from_epoch

items = ['DECOM__TLM__INST__HEALTH_STATUS__TEMP1__CONVERTED']

file = Tempfile.new('test')
disable_instrumentation do
  OpenC3::StreamingWebSocketApi.new do |api|
    api.add(items: items, start_time: start_time, end_time: end_time)
    done = false
    while(!done)
      data = api.read
      if data.length == 0
        done = true
      else
        file.write(data)
      end
    end
  end
end
file.rewind
put_target_file("INST/test.txt", file)
download_file("INST/test.txt")

Note I put the looping inside of disable_instrumentation so you avoid the gui updating over and over again.

3. In your example it looks like you were requesting the RAW packet which gives you the raw buffer. It is supposed to work where you give packets: rather than items: to add but I'm afraid we just run into the same issue as 1 where we get a bunch of [#<Hash:21280>, ... output.

packets = [ 'DECOM__TLM__INST__HEALTH_STATUS__CONVERTED' ]
...
api.add(packets: packets, start_time: start_time, end_time: end_time)

4. I think until we fix the hash bug you'll need to use Data Extractor.

[jmthomas]

I found a work around that can work until we fix our implementation (should work for both items and packets):

  OpenC3::StreamingWebSocketApi.new do |api|
    # api.add(items: items, start_time: start_time, end_time: end_time)
    api.add(packets: packets, start_time: start_time, end_time: end_time)
    done = false
    while(!done)
      data = api.read
      if data.length == 0
        done = true
      else
        data.each do |item|
          file.write(item.old_inspect())
        end
      end
    end
  end

[hmaclachlan]

@jmthomas thanks for the input on work arounds. I was unable to get the most recent method provided working for more than 8 item inputs and for the packet. It would run to completion but not populate the file (code below)

def download_tlm(items, start_time, end_time, filepath)
  file = Tempfile.new('temp')
  OpenC3::StreamingWebSocketApi.new do |api|
    api.add(items: items, start_time: start_time, end_time: end_time)
    #api.add(packets: packets, start_time: start_time, end_time: end_time)
    done = false
    while(!done)
      data = api.read
      if data.length == 0
        done = true
      else
        data.each do |item|
          file.write(item.old_inspect())
        end
      end
    end
  end
file.rewind
put_target_file(filepath, file)
download_file("filepath)
end

The previous work around that you gave for the batch size limitation worked great. No need to dig further into this though - I have opted to split items from the desired packet into two separate files - works for what I need at the moment.

[hmaclachlan]

@jmthomas I have another telemetry logging question as well. Is there any way to mass download raw & decom logs from the bucket explorer (either from a script or another function)? Or is the best way to do this by manually downloading each log

[jmthomas]

We don't provide an interface to mass download files from MINIO. That said this is open source software and MINIO is pretty well documented so it would awesome if you could create something and share it with the community! :-)

[hmaclachlan]

@jmthomas thanks for the recommendation, I will look into it!