class Pocolog::DataStream

Interface for reading a stream in a Pocolog::Logfiles

Attributes

index[R]

info[R]

The Logfiles::StreamInfo structure for that stream

logfile[R]

marshalled_registry[R]

metadata[R]

The stream associated metadata

name[R]

sample_index[R]

The index in the stream of the last read sample

It is equal to size if we are past-the-end, i.e. if one called next until it returned nil

time_getter[RW]

type_name[R]

Public Class Methods

new(logfile, index, name, type_name, marshalled_registry, metadata) click to toggle source

# File lib/pocolog/data_reader.rb, line 22
def initialize(logfile, index, name, type_name, marshalled_registry, metadata)
    @logfile, @index, @name, @type_name, @marshalled_registry, @metadata =
        logfile, index, name, type_name, marshalled_registry, metadata
    
    @data = nil
    @registry = nil
    @sample_index = -1
    @raw_data_buffer = ""
end

Public Instance Methods

[](sample_index) click to toggle source

Returns the sample_index sample of this stream

# File lib/pocolog/data_reader.rb, line 63
def [](sample_index)
    samples.between(sample_index, nil).
        find { true }
end

advance() click to toggle source

Reads the next sample in the file, and returns its header. Returns nil if the end of file has been reached. Unlike next, it does not decodes the data payload.

# File lib/pocolog/data_reader.rb, line 302
def advance
    if sample_index < size-1
        @sample_index += 1
        rio, file_pos = stream_index.file_position_by_sample_number(@sample_index)
        logfile.read_one_block(file_pos, rio)
        return logfile.data_header
    else
        @sample_index = size
    end
    nil
end

close() click to toggle source

# File lib/pocolog/data_reader.rb, line 44
def close
    logfile.close
end

closed?() click to toggle source

# File lib/pocolog/data_reader.rb, line 36
def closed?
    logfile.closed?
end

copy_to(index1,index2,stream) → true click to toggle source

copy_to(time1,time2,stream) → true

copies all blocks from start_index/time to end_index/time to the given stream for each block the given code block is called. If the code block returns 1 the copy process will be canceled and the method returns false

The given interval is automatically truncated if it is too big

# File lib/pocolog/data_reader.rb, line 352
def copy_to(start_index,end_index,stream,&block)
    if !samples?(start_index,end_index)
        raise "no samples for the given interval!"
    end
    interval = time_interval
    return unless interval.first
    start_index = if start_index.is_a? Time
                      if interval.first > start_index
                          0
                      else
                          stream_index.sample_number_by_time(start_index)
                      end
                  else
                      if start_index < 0
                          0
                      else
                          start_index
                      end
                  end
    end_index = if end_index.is_a? Time
                    if interval.last < end_index
                        size
                    else
                        stream_index.sample_number_by_time(end_index)
                    end
                else
                    if end_index >= size
                        size
                    else
                        end_index
                    end
                end
    
    data_buffer = String.new
    seek(start_index,false)
    counter = 0
    max = end_index-start_index
    begin
        if block
            return false if block.call(counter)
        end
        data = logfile.data(data_header, data_buffer)
        stream.write_raw(data_header.rt,data_header.lg,data)
        counter += 1
    end while advance && counter < max
    counter
end

data(data_header = nil) click to toggle source

# File lib/pocolog/data_reader.rb, line 219
def data(data_header = nil)
    Typelib.to_ruby(raw_data(data_header))
end

data_header() click to toggle source

The data header for the current sample. You can store a copy of this header to retrieve data later on with data:

# Don't forget to duplicate !
stored_header = stream.data_header.dup
...
data = stream.data(stored_header)

# File lib/pocolog/data_reader.rb, line 97
def data_header; logfile.data_header end

each_block(rewind = true) { || ... } click to toggle source

Enumerates the blocks of this stream

# File lib/pocolog/data_reader.rb, line 52
def each_block(rewind = true)
    if rewind
        self.rewind
    end

    while advance
        yield if block_given?
    end
end

empty?() click to toggle source

True if the size of this stream is zero

# File lib/pocolog/data_reader.rb, line 105
def empty?; size == 0 end

eof?() click to toggle source

True if we read past the last sample

# File lib/pocolog/data_reader.rb, line 103
def eof?; size == sample_index end

first → [time_rt, time_lg, data] click to toggle source

Returns the first sample in the stream, or nil if the stream is empty

It differs from rewind as it always decodes the data payload.

After a call to first, sample_index is 0

# File lib/pocolog/data_reader.rb, line 246
def first
    rewind
    self.next
end

has_type?() click to toggle source

True if this data stream has a Typelib::Registry object associated

# File lib/pocolog/data_reader.rb, line 108
def has_type?; !marshalled_registry.empty? end

last → [time_rt, time_lg, data] click to toggle source

Returns the last sample in the stream, or nil if the stream is empty.

After a call to last, sample_index is size - 1

# File lib/pocolog/data_reader.rb, line 257
def last
    last_sample_pos = info.interval_io[1]
    logfile.seek(last_sample_pos[1], last_sample_pos[0])
    @sample_index = size - 2
    self.next
end

next → [time_rt, time_lg, data] click to toggle source

Reads the next sample in the file, and returns it. It differs from advance as it always decodes the data sample.

# File lib/pocolog/data_reader.rb, line 319
def next
    header = advance
    if(header) 
      return [header.rt, Time.at(header.lg - logfile.time_base), data]
    end
end

open() click to toggle source

# File lib/pocolog/data_reader.rb, line 40
def open
    logfile.open
end

previous → [time_rt, time_lg, data] click to toggle source

Reads the previous sample in the file, and returns it.

# File lib/pocolog/data_reader.rb, line 330
def previous
    if sample_index < 0
        # Just rewind, never played
        return nil
    elsif sample_index == 0
        # Beginning of file reached
        rewind
        return nil 
    else
        seek(sample_index - 1)
    end
end

raw_data(data_header = nil, sample = nil) click to toggle source

Returns the decoded data sample associated with the given block header.

Block headers are returned by rewind

# File lib/pocolog/data_reader.rb, line 184
def raw_data(data_header = nil, sample = nil)
    if(@data && !data_header) then @data
    else
        marshalled_data = logfile.data(data_header, @raw_data_buffer)
        data = sample || type.new
        data.from_buffer_direct(marshalled_data)
        if logfile.endian_swap
            data = data.endian_swap
        end
        data
    end
rescue Interrupt
    raise
rescue Exception => e
    raise e, "failed to unmarshal sample at #{(data_header || logfile.data_header).payload_pos}: #{e.message}", e.backtrace
end

read_one_raw_data_sample(position, sample = nil) click to toggle source

# File lib/pocolog/data_reader.rb, line 201
def read_one_raw_data_sample(position, sample = nil)
    rio, block_pos = stream_index.file_position_by_sample_number(position)
    marshalled_data = logfile.read_one_data_payload(rio, block_pos, @raw_data_buffer)
    data = sample || type.new
    data.from_buffer_direct(marshalled_data)
    if logfile.endian_swap
        data = data.endian_swap
    end
    data
rescue Interrupt
    raise
rescue Exception => e
    if rio.respond_to?(:path)
        file = "in #{rio.path} "
    end
    raise e, "#{file}failed to unmarshal sample for block position #{block_pos}: #{e.message}", e.backtrace
end

registry() click to toggle source

Get the Typelib::Registry object for this stream

# File lib/pocolog/data_reader.rb, line 119
def registry
    if !@registry
        @registry = logfile.registry || Typelib::Registry.new

        stream_registry = Typelib::Registry.new

        if has_type?
            begin
                stream_registry.merge_xml(marshalled_registry)
            rescue ArgumentError
                Typelib::Registry.add_standard_cxx_types(stream_registry)
                stream_registry.merge_xml(marshalled_registry)
            end

            stream_registry = stream_registry.minimal(typename, false)

            # if we do have a registry, then adapt it to the local machine
            # if needed. Right now, this is required if containers changed
            # size.
            resize_containers = Hash.new
            stream_registry.each do |type|
                if type <= Typelib::ContainerType && type.size != type.natural_size
                    resize_containers[type] = type.natural_size
                end
            end
            stream_registry.resize(resize_containers)

            begin
                @registry.merge(stream_registry)
            rescue RuntimeError => e
                if e.message =~ /but with a different definition/
                    raise e, e.message + ". Are you mixing 32 and 64 bit data ?", e.backtrace
                end
            end
        end
    end
    @registry
end

reload_registry() click to toggle source

Reload the registry. Can be useful if new convertions have been added to the Typelib system

# File lib/pocolog/data_reader.rb, line 112
def reload_registry
    @registry = nil
    @type = nil
    registry
end

rewind → data_header click to toggle source

Goes to the first sample in the stream, and returns its header. Returns nil if the stream is empty.

It differs from first as it does not decode the data payload.

# File lib/pocolog/data_reader.rb, line 230
def rewind
    # The first sample in the file has index 0, so set sample_index to
    # -1 so that (@sample_index += 1) sets the index to 0 for the first
    # sample
    @sample_index = -1
    nil
end

samples(read_data = true) click to toggle source

Returns a SampleEnumerator object for this stream

# File lib/pocolog/data_reader.rb, line 49
def samples(read_data = true); SampleEnumerator.new(self, read_data) end

samples?(pos1,pos2) → true click to toggle source

samples?(time1,time2) → true

returns true if stream samples lies insight the given time or position interval

# File lib/pocolog/data_reader.rb, line 405
def samples?(start_index,end_index)
    if start_index.is_a? Time
        interval = time_interval
        return unless interval.first
        start_index <= interval.last && start_index <= end_index && end_index >= interval.first
    else
        start_index < size && start_index <= end_index && end_index >= 0
    end
end

seek(pos, decode_data = true) click to toggle source

Seek the stream at the given position

If pos is a Time object, seeks to the last sample whose logical time is not greater than pos

If pos is an integer, it is interpreted as an index and the stream goes to the sample that has this index.

Returns [rt, lg, data] for the current sample (if there is one), and nil otherwise

# File lib/pocolog/data_reader.rb, line 274
def seek(pos, decode_data = true)
    if pos.kind_of?(Time)
        return nil if(time_interval.empty? || time_interval[0] > pos || time_interval[1] < pos)
        @sample_index = stream_index.sample_number_by_time(pos)
    else
        @sample_index = pos
    end

    rio, file_pos = stream_index.file_position_by_sample_number(@sample_index)
    block_info = logfile.read_one_block(file_pos, rio)
    if block_info.index != self.index
        raise InternalError, "index returned index=#{@sample_index} and pos=#{file_pos} as position for seek(#{pos}) but it seems to be a sample in stream #{logfile.stream_from_index(block_info.index).name} while we were expecting #{name}"
    end
    if header = self.data_header
        header = header.dup

        if(decode_data)
            data = self.data(header)
            return [header.rt, Time.at(header.lg - logfile.time_base), data]
        else
            header
        end
    end
end

size() click to toggle source

The size, in samples, of data in this stream

# File lib/pocolog/data_reader.rb, line 100
def size; info.size end

stream_index() click to toggle source

# File lib/pocolog/data_reader.rb, line 32
def stream_index
    info.index
end

sub_field(fieldname, data_header = nil) click to toggle source

Returns the decoded subfield specified by 'fieldname' for the given data header. If no header is given, the current last read data header is used

# File lib/pocolog/data_reader.rb, line 164
def sub_field(fieldname, data_header = nil)
    header = data_header || logfile.data_header
    if( header.compressed )
        data(data_header).send(fieldname)
    elsif(type.is_a?(Typelib::CompoundType) and type.has_field?(fieldname))
        offset = type.offset_of(fieldname)
        subtype = type[fieldname]
        rawData = logfile.sub_field(offset, subtype.size, data_header)
        wrappedType = subtype.wrap(rawData)
        rubyType = Typelib.to_ruby(wrappedType)
        rubyType
    else
        nil
    end
end

time() click to toggle source

Returns the time of the current sample

# File lib/pocolog/data_reader.rb, line 71
def time
    header = logfile.data_header
    if !time_getter
        [header.rt, Time.at(header.lg - logfile.time_base)]
    else
        [header.rt, time_getter[data(header)]]
    end
end

time_interval(rt = false) click to toggle source

Get the logical time of first and last samples in this stream. If rt is true, returns the interval for the wall-clock time

Returns nil if the stream is empty

# File lib/pocolog/data_reader.rb, line 84
def time_interval(rt = false)
    if rt then info.interval_rt
    else info.interval_lg
    end
end

type() click to toggle source

Get a Typelib object describing the type of this data stream

# File lib/pocolog/data_reader.rb, line 159
def type; @type ||= registry.get(typename) end

typename() click to toggle source

Provided for backward compatibility reasons. Use type_name

# File lib/pocolog/data_reader.rb, line 9
def typename; type_name end

write(rt, lg, data) click to toggle source

Write a sample in this stream, with the rt and lg timestamps. data can be either a Typelib::Type object of the right type, or a String (in which case we consider that it is the raw data)

# File lib/pocolog/data_writer.rb, line 7
def write(rt, lg, data)
    data = Typelib.from_ruby(data, type)
    write_raw(rt, lg, data.to_byte_array)
end

write_raw(rt, lg, data) click to toggle source

Write an already marshalled sample. data is supposed to be a typelib-marshalled value of the stream type

# File lib/pocolog/data_writer.rb, line 14
def write_raw(rt, lg, data)
    logfile.write_data_block(self, rt, lg, data)
end