Class: Orocos::Log::Replay

Inherits:
Object
  • Object
show all
Includes:
Namespace, PortsSearchable
Defined in:
lib/orocos/log/replay.rb

Overview

Class for loading and replaying pocolog (Rock) log files.

This class creates objects whose API is compatible with TaskContext and OutputPort, using the log data.

By default, all tasks that are present in the log files provided to Replay.open can be resolved using the orocos name service. If this behaviour is unwanted, call #deregister_tasks after Replay.open or #load was called. To do it on a task-by-task basis, do the following after the call to Replay.open or #load

replay.name_service.deregister 'task_name'

Constant Summary

Constants included from Namespace

Namespace::DELIMATOR

Class Attribute Summary collapse

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from PortsSearchable

#find_all_input_ports, #find_all_output_ports, #find_all_ports, #find_input_port, #find_output_port, #find_port

Methods included from Namespace

#basename, #map_to_namespace, #namespace, #namespace=, #same_namespace?, #split_name, split_name, validate_namespace_name, #verify_same_namespace

Constructor Details

#initialize(*path) ⇒ Replay

Creates a new instance of Replay

If a path is givien load is called after creation to load the log files.



190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
 
# File 'lib/orocos/log/replay.rb', line 190

def initialize(*path)
    if !path.empty?
        raise ArgumentError, "Replay.new(*path) is deprecated, use Replay.open(*path) to create and load files at the same time"
    end

    @default_timestamp = nil
    @timestamps = Hash.new
    @tasks = Hash.new
    @annotations = Array.new
    @current_annotations = Hash.new
    @speed = 1
    @replayed_ports = Array.new
    @replayed_properties = Array.new
    @replayed_objects = Array.new
    @used_streams = Array.new
    @stream = nil
    @current_sample = nil
    @process_qt_events = false
    @log_config_file = Replay::log_config_file
    @namespace = ''
    reset_time_sync
    time_sync
end

Dynamic Method Handling

This class handles dynamic methods through the method_missing method

#method_missing(m, *args, &block) ⇒ Object

This is used to support the syntax. log_replay.task



1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
 
# File 'lib/orocos/log/replay.rb', line 1059

def method_missing(m,*args,&block) #:nodoc:
    task_name = full_name(m)
    return @tasks[task_name] if task_name

    begin
        super(m.to_sym,*args,&block)
    rescue  NoMethodError => e
        Log.error "#{m} is not a Log::Task of the current log file(s)"
        Log.error "The following tasks are availabe:"
        @tasks.each_value do |task|
            Log.error "  #{task.name}"
        end
        raise e
    end
end

Class Attribute Details

.log_config_file ⇒ Object

Returns the value of attribute log_config_file



60
61
62
 
# File 'lib/orocos/log/replay.rb', line 60

def log_config_file
  @log_config_file
end

Instance Attribute Details

#actual_speed ⇒ Float (readonly)

The actual replay speed

This is updated during replay, and reflects the actual replay speed

Returns:

  • (Float) 


126
127
128
 
# File 'lib/orocos/log/replay.rb', line 126

def actual_speed
  @actual_speed
end

#annotations ⇒ Object (readonly)

array of stream annotations



129
130
131
 
# File 'lib/orocos/log/replay.rb', line 129

def annotations
  @annotations
end

#current_sample ⇒ Object (readonly)

last replayed sample



82
83
84
 
# File 'lib/orocos/log/replay.rb', line 82

def current_sample
  @current_sample
end

#log_config_file ⇒ Object

name of the log file which holds the logged properties is converted into Regexp



79
80
81
 
# File 'lib/orocos/log/replay.rb', line 79

def log_config_file
  @log_config_file
end

#name_service ⇒ Orocos::Local

Returns a local nameservice on which the log tasks are registered. It is added to the global name service with #register_tasks and removed with #unregister_tasks

Returns:

  • (Orocos::Local) —

    a local nameservice on which the log tasks are registered. It is added to the global name service with #register_tasks and removed with #unregister_tasks



67
68
69
 
# File 'lib/orocos/log/replay.rb', line 67

def name_service
  @name_service
end

#name_service_async ⇒ Orocos::Async::Local

Returns a local async nameservice on which the log tasks are registered. It is added to the global name service with #register_tasks and removed with #unregister_tasks

Returns:

  • (Orocos::Async::Local) —

    a local async nameservice on which the log tasks are registered. It is added to the global name service with #register_tasks and removed with #unregister_tasks



72
73
74
 
# File 'lib/orocos/log/replay.rb', line 72

def name_service_async
  @name_service_async
end

#out_of_sync_delta ⇒ Float (readonly)

Measure of time synchronization during replay

This is updated during replay to reflect how fast the replay actually is. This is the difference (in seconds) between the replay time that we should have and the replay time that we actually have

In practice, negative values mean that the replayed samples are behind the simulated times, and positive values mean that the replayed samples are replayed to fast

Returns:

  • (Float) 


118
119
120
 
# File 'lib/orocos/log/replay.rb', line 118

def out_of_sync_delta
  @out_of_sync_delta
end

#process_qt_events ⇒ Object

set it to true if processing of qt events is needed during synced replay



97
98
99
 
# File 'lib/orocos/log/replay.rb', line 97

def process_qt_events
  @process_qt_events
end

#replayed_annotations ⇒ Object

array of all replayed annotaions this array is filled after #align was called



94
95
96
 
# File 'lib/orocos/log/replay.rb', line 94

def replayed_annotations
  @replayed_annotations
end

#replayed_ports ⇒ Object

array of all replayed ports this array is filled after #align was called



86
87
88
 
# File 'lib/orocos/log/replay.rb', line 86

def replayed_ports
  @replayed_ports
end

#replayed_properties ⇒ Object

array of all replayed properties this array is filled after #align was called



90
91
92
 
# File 'lib/orocos/log/replay.rb', line 90

def replayed_properties
  @replayed_properties
end

#speed ⇒ Object

desired replay speed = 1 –> record time



75
76
77
 
# File 'lib/orocos/log/replay.rb', line 75

def speed
  @speed
end

#timestamps ⇒ Hash<String,#call> (readonly)

Returns a mapping from a typelib type name to an object that allows to extract the timestamp from a value of that type

Returns:

  • (Hash<String,#call>) —

    a mapping from a typelib type name to an object that allows to extract the timestamp from a value of that type

See Also:

  • {timestamp}


104
105
106
 
# File 'lib/orocos/log/replay.rb', line 104

def timestamps
  @timestamps
end

#use_sample_time ⇒ Object

If true, the alignment algorithm is going to use the sample time for alignment. Otherwise, it uses the time at which the sample got written on disk (logical time)

See also #time_source



172
173
174
 
# File 'lib/orocos/log/replay.rb', line 172

def use_sample_time
  @use_sample_time
end

#used_streams ⇒ Array<Pocolog::DataStream> (readonly)

The streams that are actually replayed

Returns:

  • (Array<Pocolog::DataStream>) 


134
135
136
 
# File 'lib/orocos/log/replay.rb', line 134

def used_streams
  @used_streams
end

Class Method Details

.open(*path) ⇒ Object 



174
175
176
177
178
179
180
181
182
183
184
 
# File 'lib/orocos/log/replay.rb', line 174

def self.open(*path)
    replay = new
    replay.load(*path)
    replay
rescue ArgumentError => e
    Orocos.error "Cannot load logfiles"
    raise e
rescue Pocolog::Logfiles::MissingPrologue => e
    Orocos.error "Wrong log format"
    raise e
end

Instance Method Details

#add_intervals_as_log_markers(intervals, comment) ⇒ Object

Adds the given time intervals as LogMarkers

Parameters:

  • intervals (Array<Array<Time>>) —

    The intervals

  • comment (String) —

    Comment of the log markers



769
770
771
772
773
774
775
776
777
778
779
 
# File 'lib/orocos/log/replay.rb', line 769

def add_intervals_as_log_markers(intervals,comment)
    markers = log_markers # fill @markers from log file
    intervals.each do |interval|
        markers << LogMarker.new(interval.first,:start,-1,comment)
        markers << LogMarker.new(interval.last,:stop,-1,comment)
    end
    markers.sort! do |a,b|
        a.time <=> b.time
    end
    markers
end

#advance ⇒ Object 



454
455
456
457
458
459
460
 
# File 'lib/orocos/log/replay.rb', line 454

def advance
    if(@stream)
        return @stream.advance
    else
        throw "Stream is not initialized yet"
    end
end

#align(time_source = self.time_source) ⇒ Object

Aligns all streams which have at least:

  • one reader

  • or one connections

  • or track set to true.

After calling this method no more ports can be tracked.

time_source is passed through to the StreamAligner. It can be used to override the global #time_source parameter. See #time_source for available values.



400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
 
# File 'lib/orocos/log/replay.rb', line 400

def align( time_source = self.time_source )
    @replayed_ports = Array.new
    @used_streams = Array.new
    @replayed_annotations = Array.new

    if !replay?
        Log.warn "No ports are selected. Assuming that all ports shall be replayed."
        Log.warn "Connect port(s) or set their track flag to true to get rid of this message."
        track(true)
    end

    #get all properties which shall be replayed
    each_task do |task|
        if task.used?
            task.port("state").tracked=true if task.has_port?("state")
            task.properties.values.each do |property|
                property.tracked = true
                next if property.stream.empty?
                @replayed_properties << property
            end
        end
    end

    #get all streams which shall be replayed
    each_port do |port|
        @replayed_ports << port if port.used?
    end

    Log.info "Aligning streams --> all ports which are unused will not be loaded!!!"
    if @replayed_properties.empty? && @replayed_ports.empty?
        raise "No log data are replayed. All selected streams are empty."
    end

    # If we do have something to replay, then add the annotations as
    # well
    annotations.each do |annotation|
        next if annotation.stream.empty?
        @replayed_annotations << annotation
    end

    @replayed_objects = @replayed_properties + @replayed_ports + @replayed_annotations
    @used_streams = @replayed_objects.map(&:stream)

    Log.info "Replayed Ports:"
    @replayed_ports.each {|port| Log.info PP.pp(port,"")}

    #join streams
    @stream = Pocolog::StreamAligner.new(time_source, *@used_streams)
    @stream.rewind

    reset_time_sync
    return step
end

#aligned? ⇒ Boolean

Returns:

  • (Boolean) 


510
511
512
 
# File 'lib/orocos/log/replay.rb', line 510

def aligned?
    return @stream != nil
end

#calc_statistics(time = current_time) ⇒ Object 



585
586
587
588
589
590
591
592
593
594
595
 
# File 'lib/orocos/log/replay.rb', line 585

def calc_statistics(time = current_time)
    time = current_time

    @base_time ||= time
    @start_time ||= Time.now

    required_delta = (time - @base_time)/@speed
    actual_delta   = Time.now - @start_time
    @out_of_sync_delta = @time_sync_proc.call(time,actual_delta,required_delta)
    @actual_speed = required_delta/actual_delta*@speed
end

#clear_reader_buffers ⇒ Object

Clears all reader buffers. This is usefull if you are changing the replay direction.



1043
1044
1045
1046
1047
 
# File 'lib/orocos/log/replay.rb', line 1043

def clear_reader_buffers
    @tasks.each_value do |task|
        task.clear_reader_buffers
    end
end

#close ⇒ Object

close the log file, deregister from name service and also close all available streams



491
492
493
494
 
# File 'lib/orocos/log/replay.rb', line 491

def close
		deregister_tasks
		# TODO close all streams
end

#connect_to(task, port_mappings = Hash.new, port_policies = Hash.new, ports_ignored = Array.new) ⇒ Object

Tries to connect all input ports of the OROCOS task to simulated OutputPorts

Parameter:

*task => task to connect to *port_mappings => hash to define port mappings => dst_port_name *ports_ignored => array of port names which shall be ignored



326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
 
# File 'lib/orocos/log/replay.rb', line 326

def connect_to(task,port_mappings = Hash.new ,port_policies = Hash.new,ports_ignored = Array.new)
    #convenience block to do connect_to(task,:auto_ignore)
    if port_mappings == :auto_ignore
        ports_ignored = port_mappings
        port_mappings = Hash.new
    end
    ports_ignored = Array.new(ports_ignored)

    #start task if necessary
    if task.state == :PRE_OPERATIONAL
        task.configure
    end
    if task.state == :STOPPED
        task.start
    end

    #to have a better user interface the hash is inverted
    #port1 connect_to port2 is written as ('port1' => 'port2')
    port_mappings = port_mappings.invert

    task.each_port do |port|
        if port.to_orocos_port.kind_of?(Orocos::InputPort) && !ports_ignored.include?(port.name)
            target_port = find_port(port.type_name,port_mappings[port.name]||port.name)
            if target_port
                target_port.connect_to(port,port_policies[port.name])
            elsif !ports_ignored.include? :auto_ignore
                raise ArgumentError, "cannot find an output port for #{port.name}"
            else
                Log.warn "No input port can be found for output port #{port.full_name}."
            end
        end
    end
end

#current_annotations ⇒ Object

The current annotations

This is an aggregated version of #annotations, where the value for each key is the last value known (i.e. the value from the last annotation with that key that has a timestamp lower than the current time)



142
143
144
145
146
 
# File 'lib/orocos/log/replay.rb', line 142

def current_annotations
    annotations.inject(Hash.new) do |current, ann|
        current.merge(ann.current_state)
    end
end

#current_port ⇒ Object

returns the last port which recieved data



681
682
683
684
685
686
 
# File 'lib/orocos/log/replay.rb', line 681

def current_port
    if @current_sample
        stream_idx = @current_sample[0]
        replayed_ports[stream_idx]
    end
end

#current_sample_data ⇒ Object

returns the current data of the current sample



689
690
691
692
693
694
 
# File 'lib/orocos/log/replay.rb', line 689

def current_sample_data
    if @current_sample
        sample_info = @current_sample[2]
        sample_info[0].read_one_raw_data_sample(sample_info[1])
    end
end

#current_time ⇒ Object 



573
574
575
576
577
578
579
580
581
582
583
 
# File 'lib/orocos/log/replay.rb', line 573

def current_time
    stream_idx, time, sample_info = @current_sample
    return if !time

    stream_type = @stream.stream_by_index(stream_idx).type
    if getter = (timestamps[stream_type.name] || default_timestamp)
        data = sample_info[0].read_one_raw_data_sample(sample_info[1])
        getter[data]
    else time
    end
end

#default_timestamp(&block) ⇒ Object

Sets a code block to calculate the default timestamp duricng replay.



279
280
281
282
283
 
# File 'lib/orocos/log/replay.rb', line 279

def default_timestamp(&block)
    if block_given? then @default_timestamp = block
    else @default_timestamp
    end
end

#deregister_tasks ⇒ Object

deregister the local name service again



483
484
485
486
487
 
# File 'lib/orocos/log/replay.rb', line 483

def deregister_tasks
		if @name_service
 Orocos::name_service.delete @name_service
		end
end

#duration ⇒ Float

The total duration of the replayed data, in seconds

Returns:

  • (Float) 


517
518
519
520
521
522
523
524
525
 
# File 'lib/orocos/log/replay.rb', line 517

def duration
    intervals = used_streams.map { |s| s.info.interval_lg }
    min = intervals.map(&:first).min
    max = intervals.map(&:last).max
    if min && max
        max - min
    else 0
    end
end

#each_port(&block) ⇒ Object

Iterates through all simulated ports.



813
814
815
816
817
 
# File 'lib/orocos/log/replay.rb', line 813

def each_port(&block)
    @tasks.each_value do |task|
        task.each_port(&block)
    end
end

#each_task(&block) ⇒ Object

Iterates through all simulated tasks



851
852
853
854
855
 
# File 'lib/orocos/log/replay.rb', line 851

def each_task (&block)
    @tasks.each_value do |task|
        yield(task) if block_given?
    end
end

#eof? ⇒ Boolean

Returns true if the end of file is reached.

Returns:

  • (Boolean) 


873
874
875
 
# File 'lib/orocos/log/replay.rb', line 873

def eof?
    return @stream.eof?
end

#export_to_file(file, start_index = 0, end_index = 0, &block) ⇒ Object

exports all aligned stream to a new log file if no start and end index is given all data are exported otherwise the data are truncated according to the given global indexes the block is called for each sample to update a progress bar



1053
1054
1055
 
# File 'lib/orocos/log/replay.rb', line 1053

def export_to_file(file,start_index=0,end_index=0,&block)
    @stream.export_to_file(file,start_index,end_index,&block)
end

#extract_intervals(start_time = nil, end_time = nil, min_val = 0.8, kernel_size = 5.0) {|reader, sample| ... } ⇒ Array<Array<Time>>

Extracts time intervals from the log file where the given code block returns true.

For each sample the given code block is called with the current port and sample as parameter. After all samples were replayed the generated result vector (true is interpreted as 1) is filtered with a box filter of the given size. The returned intervals are these intervals where the filtered result vector is equal or bigger than min_val

Parameters:

  • start_time (Time) (defaults to: nil) —

    Start time of the interval which is replayed (nil = start of the log file)

  • end_time (Time) (defaults to: nil) —

    End time of the interval which is replayed (nil = end of the log file)

  • min_val (Float) (defaults to: 0.8) —

    Min value of the filtered result vector to be regarded as inlayer

  • kernel_size (Float) (defaults to: 5.0) —

    Filter kernel size of the box filter in seconds

Yields:

  • (reader, sample)

Yield Parameters:

  • reader —

    the data reader of the port from which the sample has been read

  • sample —

    the data sample

Yield Returns:

  • (Boolean)

Returns:

  • (Array<Array<Time>>) —

    extracted intervals



717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
 
# File 'lib/orocos/log/replay.rb', line 717

def extract_intervals(start_time=nil,end_time=nil, min_val=0.8,kernel_size=5.0,&block)
    #replay given intervals and collect results of the code block
    result,times = [],[]
    start_time ||= begin
                  rewind
                  time
              end
    seek(start_time)
    begin
        if block.call(current_port,current_sample_data)
            result << 1
        else
            result << 0
        end
        times << time
    end while(step && (!end_time || time <= end_time))

    #filter result vector
    idx,sum,size = 0,0,0
    filtered = result.map do |e|
        while times[idx+size] - times[idx] < kernel_size && idx+size < times.size-1
            size += 1
            sum += result[idx+size]
        end
        val = if size > 0
                sum/size
              else
                  0
              end
        sum -= result[idx]
        idx += 1
        size -= 1
        val
    end

    #extract intervals
    intervals,start = [],nil
    filtered.each_with_index do |e,i|
        if e >= min_val
            start ||= times[i]
        elsif start
            intervals << [start,times[i]] if times[i]-start >= kernel_size
            start = nil
        end
    end
    intervals
end

#first_sample_pos(stream) ⇒ Object 



462
463
464
 
# File 'lib/orocos/log/replay.rb', line 462

def first_sample_pos(stream)
    @stream.first_sample_pos(stream)
end

#full_name(name) ⇒ Object

returns the full name of a task reachable under the given name



361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
 
# File 'lib/orocos/log/replay.rb', line 361

def full_name(name)
    # use full name
    name = name.to_s
    return name if @tasks.has_key?(name)

    # use namespace of replay
    name2 = map_to_namespace(name)
    return name2 if @tasks.has_key?(name2)

    # use all namespaces
    t = @tasks.find do |key,task|
        task.basename == name
    end
    t.first if t
end

#generate_log_markers(comment, min_val = 0.8, kernel_size = 5.0) {|reader, sample| ... } ⇒ Array<Array<Time>>

Extracts time intervals from the log file where the given code block returns true and adds these interval as log markers to the log replay instance.

Parameters:

  • comment (String) —

    Comment of the log markers

  • min_val (Float) (defaults to: 0.8) —

    Min value of the filtered result vector to be regarded as inlayer

  • kernel_size (Float) (defaults to: 5.0) —

    Filter kernel size of the box filter in seconds

Yields:

  • (reader, sample)

Yield Parameters:

  • reader —

    the data reader of the port from which the sample has been read

  • sample —

    the data sample

Yield Returns:

  • (Boolean)

Returns:

  • (Array<Array<Time>>) —

    extracted intervals

See Also:



797
798
799
800
801
802
 
# File 'lib/orocos/log/replay.rb', line 797

def generate_log_markers(comment,min_val=0.8,kernel_size=5.0,&block)
    intervals = extract_intervals(nil,nil, min_val,kernel_size,&block)
    add_intervals_as_log_markers(intervals,comment)
    rewind
    intervals
end

#has_task?(name) ⇒ Boolean

Returns:

  • (Boolean) 


845
846
847
848
 
# File 'lib/orocos/log/replay.rb', line 845

def has_task?(name)
    name = map_to_namespace name
    @tasks.has_key?(name.to_s)
end

#last_sample_pos(stream) ⇒ Object 



466
467
468
 
# File 'lib/orocos/log/replay.rb', line 466

def last_sample_pos(stream)
    @stream.last_sample_pos(stream)
end

#load(*paths) ⇒ Object

Logs that share the same basename, will be joined such that the streams within the logs appear continous.

a collection of files/directories can be given as arguments, followed by a typelib registry and/or an options hash. The options can be given as:

:registry - same as providing the registry directly :multifile - set to :last if you don't want merging of logs with the same

basename

Raises:

  • (ArgumentError) 


979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
 
# File 'lib/orocos/log/replay.rb', line 979

def load(*paths)
    paths.flatten!
    raise ArgumentError, "No log file was given" if paths.empty?

    logreg = nil
    if paths.last.kind_of?(Typelib::Registry)
        logreg = paths.pop
    end
		opts = {}
		if paths.last.kind_of?(Hash)
		    opts = paths.pop
		    logreg = opts[:registry] if opts[:registry]
		end

    paths.each do |path|
        #check if path is a directory
        path = File.expand_path(path)
        if File.directory?(path)
            all_files = Dir.enum_for(:glob, File.join(path, '*.*.log'))
            by_basename = all_files.inject(Hash.new) do |h, path|
                split = path.match(/^(.*)\.(\d+)\.log$/)
                if split
                    basename, number = split[1], Integer(split[2])
                    h[basename] ||= Array.new
                    h[basename][number] = path
                    h
                else
                    Orocos.warn "invalid log file name #{path}. Expecting: /^(.*)\.(\d+)\.log$/"
                    h
                end
            end
            if by_basename.empty?
                Orocos.warn "empty directory: #{path}"
                next
            end

            by_basename.each_value do |files|
			    if opts[:multifile] == :last
				files = files[-1,1]
			    end
			    args = files.compact.map do |path|
				File.open(path)
			    end

                args << logreg

                logfile = Pocolog::Logfiles.new(*args.compact)
                load_log_file(logfile, files.first)
            end
        elsif File.file?(path)
            file = Pocolog::Logfiles.open(path, logreg)
            load_log_file(file, path)
        else
            raise ArgumentError, "Can not load log file: #{path} is neither a directory nor a file"
        end
    end
    raise ArgumentError, "Nothing was loaded from the following log files #{paths.join("; ")}" if @tasks.empty?

    #register task on the local name server
    register_tasks
end

#load_log_file(file, path) ⇒ Object

Loads all the streams defined in the provided log file



954
955
956
957
958
959
960
961
962
963
 
# File 'lib/orocos/log/replay.rb', line 954

def load_log_file(file, path)
    Log.info "  loading log file #{path}"
    file.streams.each do |s|
        if s.metadata["rock_stream_type"] == "annotations"
            @annotations << Annotations.new(path,s)
            next
        end
        load_task_from_stream(s,path)
    end
end

#load_task_from_stream(stream, path) ⇒ Object 



901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
 
# File 'lib/orocos/log/replay.rb', line 901

def load_task_from_stream(stream, path)
    #get the name of the task which was logged into the stream
    task_name = if stream.metadata.has_key? "rock_task_name"
                    begin
                        namespace, _ = Namespace.split_name(stream.metadata["rock_task_name"])
                        Namespace.validate_namespace_name(namespace)
                    rescue ArgumentError => e
                        Orocos.warn "invalid metadata rock_task_name:'#{stream.metadata["rock_task_name"]}' for stream #{stream.name}: #{e}"
                        stream.metadata.delete("rock_task_name")
                        return load_task_from_stream(stream,path)
                    end
                    stream.metadata["rock_task_name"]
                else
                    result = stream.name.to_s.match(/^(.*)\./)
                    result[1] if result
                end
    if task_name == nil
        task_name = "unknown"
        Log.warn "Stream name (#{stream.name}) does not follow the convention TASKNAME.PORTNAME and has no metadata, assuming as TASKNAME \"#{task_name}\""
    end

    #check if there is a namespace
    task_name = if task_name == basename(task_name)
                    map_to_namespace(task_name)
                else
                    task_name
                end

    task = @tasks[task_name]
    if !task
        task = @tasks[task_name]= TaskContext.new(self,task_name, path,@log_config_file)
    end

    # Guess the stream type if it's not stored in the file
    unless (stream_type = stream.metadata['rock_stream_type'])
        if File.basename(path).start_with?(@log_config_file)
            stream_type = 'property'
        else
            stream_type = 'port'
        end
    end

    begin
        task.add_stream(stream, type: stream_type)
        Log.info "    loading stream #{stream.name} (#{stream.type_name})"
    rescue InitializePortError => error
        Log.warn "    loading stream #{stream.name} (#{stream.type_name}) failed: #{error}. "\
            "Call the port for an error message."
    end
    task
end

#log_markers ⇒ Object

returns an array of log_markers



829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
 
# File 'lib/orocos/log/replay.rb', line 829

def log_markers
    @markers ||= Array.new
    return @markers if !@markers.empty?

    annotations.each do |annotation|
        #check if this is the right type
        if annotation.stream.type_name == "/logger/Annotations"
            @markers.concat LogMarker::parse(annotation.samples)
        end
    end
    @markers.sort! do |a,b|
        a.time <=> b.time
    end
    @markers
end

#port_for(type_name, port_name, precise = true) ⇒ Object

Tries to find a OutputPort for a specefic data type. For port_name Regexp is allowed. If precise is set to true an error will be raised if more than one port is matching type_name and port_name.



311
312
313
314
315
316
317
 
# File 'lib/orocos/log/replay.rb', line 311

def port_for(type_name, port_name, precise=true)
    Log.warn "#port_for is deprecated. Use either #find_all_ports or #find_port"
    if precise
        find_port(type_name, port_name)
    else find_all_ports(type_name, port_name)
    end
end

#ports ⇒ Object

Returns an array of all simulated ports



820
821
822
823
824
825
826
 
# File 'lib/orocos/log/replay.rb', line 820

def ports
    result = Array.new
    each_port do |port|
        result << port
    end
    result
end

#pretty_print(pp) ⇒ Object

pretty print for Replay



238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
 
# File 'lib/orocos/log/replay.rb', line 238

def pretty_print(pp)
    pp.text "Orocos::Log::Replay"
    pp.nest(2) do
        pp.breakable
        pp.text "replay speed = #{@speed}"
        pp.breakable
        pp.text "Markers = #{@markers}"
        pp.breakable
        @tasks.each_value do |task|
            pp.breakable
            task.pretty_print(pp)
        end
        pp.breakable
        pp.text "Stream Annotations:"
        @annotations.each do |a|
            pp.breakable
            a.pretty_print(pp)
        end
    end
end

#push_sample(stream_idx, sample_info) ⇒ Object 



649
650
651
652
653
654
655
656
657
658
 
# File 'lib/orocos/log/replay.rb', line 649

def push_sample(stream_idx, sample_info)
    #write sample to simulated ports or properties
    log_output = @replayed_objects[stream_idx]
    log_output.update(sample_info)
    if block_given?
        data = sample_info[0].read_one_raw_data_sample(sample_info[1])
        yield(log_output,Typelib.to_ruby(data))
    end
    return *@current_sample[0, 2]
end

#refresh ⇒ Object

replays the last sample to the log port



896
897
898
899
 
# File 'lib/orocos/log/replay.rb', line 896

def refresh
    index, _, sample_info = @current_sample
    @replayed_objects[index].update(sample_info)
end

#register_tasks ⇒ Object

registers all replayed log tasks on the local name server



471
472
473
474
475
476
477
478
479
480
 
# File 'lib/orocos/log/replay.rb', line 471

def register_tasks
    @name_service ||= Local::NameService.new
    @name_service_async ||= Orocos::Async::Local::NameService.new :tasks => @tasks.values if defined?(Orocos::Async)
    @tasks.each_pair do |name,task|
        @name_service.register task
        @name_service_async.register task if @name_service_async
    end
    Orocos::name_service.add @name_service
    Orocos::Async.name_service.add @name_service_async if @name_service_async
end

#replay? ⇒ Boolean

returns false if no ports are or will be replayed

Returns:

  • (Boolean) 


225
226
227
228
229
230
231
232
233
234
235
 
# File 'lib/orocos/log/replay.rb', line 225

def replay?
    #check if stream was initialized
    if @stream
        return true
    else
        each_task do |task|
            return true if task.used?
        end
    end
    return false
end

#reset_time_sync ⇒ Object

Resets the simulated time. This should be called after the replay was paused.



529
530
531
532
533
534
 
# File 'lib/orocos/log/replay.rb', line 529

def reset_time_sync
    @start_time = nil
    @base_time  = nil
    @actual_speed = 0
    @out_of_sync_delta = 0
end

#rewind ⇒ Object

Rewinds all streams and replays the first sample.



675
676
677
678
 
# File 'lib/orocos/log/replay.rb', line 675

def rewind()
    @stream.rewind
    step
end

#run(time_sync = false, speed = 1, &block) ⇒ Object

Runs through the log files until the end is reached.



805
806
807
808
809
810
 
# File 'lib/orocos/log/replay.rb', line 805

def run(time_sync = false,speed=1,&block)
    reset_time_sync
    @speed = speed
    while step(time_sync,&block) do
    end
end

#sample_index ⇒ Object

Returns the current position of the replayed sample.



858
859
860
861
 
# File 'lib/orocos/log/replay.rb', line 858

def sample_index()
    return @stream.sample_index if @stream
    return nil
end

#sample_index_for_time(time) ⇒ Object 



259
260
261
262
263
264
265
 
# File 'lib/orocos/log/replay.rb', line 259

def sample_index_for_time(time)
    prev_pos = sample_index
    seek(time)
    target_sample_pos = sample_index
    seek(prev_pos)
    return target_sample_pos
end

#seek(pos) ⇒ Object

Seeks to the given position



878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
 
# File 'lib/orocos/log/replay.rb', line 878

def seek(pos)
    #check if stream was generated otherwise call align
    align if @stream == nil
    @current_sample = @stream.seek(pos)
    if !@current_sample
        return
    end
    @current_sample[2] = @stream.sample_info(@current_sample[0])

    #write all data to the ports
    @stream.streams.length.times do |stream_idx|
        if info = @stream.sample_info(stream_idx)
            @replayed_objects[stream_idx].update(info)
        end
    end
end

#single_data(id) ⇒ Object 



272
273
274
275
276
 
# File 'lib/orocos/log/replay.rb', line 272

def single_data(id)
    if @stream
        return @stream.single_data(id)
    end
end

#size ⇒ Object

Returns the number of samples.



863
864
865
 
# File 'lib/orocos/log/replay.rb', line 863

def size
    return @stream.size
end

#step(time_sync = false) {|reader, sample| ... } ⇒ Object

Gets the next sample, writes it to the ports which are connected to the OutputPort and updates all its readers.

If a block is given it is called this the name of the replayed port.

Parameters:

  • time_sync (Boolean) (defaults to: false) —

    if true, the method will sleep as much time as required to match the time delta in the file

Yields:

  • (reader, sample)

Yield Parameters:

  • reader —

    the data reader of the port from which the sample has been read

  • sample —

    the data sample



610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
 
# File 'lib/orocos/log/replay.rb', line 610

def step(time_sync=false,&block)
    #check if stream was generated otherwise call align
    if @stream == nil
        return align
    end
    stream_idx, time = @stream.advance
    if !stream_idx
        @current_sample = nil
        return
    end
    sample_info = @stream.sample_info(stream_idx)
    @current_sample = [stream_idx, time, sample_info]
    calc_statistics(time)

    #wait if replay is faster than the desired speed and time_sync is set to true
    if time_sync && @out_of_sync_delta > 0.001
        if @process_qt_events == true
            start_wait = Time.now
            while true
                if $qApp
                    $qApp.processEvents()
                end
                break if !@start_time                           #break if start_time was reseted throuh processEvents
                wait = @out_of_sync_delta -(Time.now - start_wait)
                if wait > 0.001
                    sleep [0.01,wait].min
                else
                    break
                end
                calc_statistics(time)
            end
        else
            sleep(@out_of_sync_delta)
        end
    end

    push_sample(stream_idx, sample_info)
end

#step_back(time_sync = false, &block) ⇒ Object

Gets the previous sample and writes it to the ports which are connected to the OutputPort and updated its readers (see step).



662
663
664
665
666
667
668
669
670
671
672
 
# File 'lib/orocos/log/replay.rb', line 662

def step_back(time_sync=false,&block)
    #check if stream was generated otherwise call start
    if !@stream
        align
        return
    end
    @current_sample = @stream.step_back
    return if !@current_sample
    @current_sample[2] = @stream.sample_info(@current_sample[0])
    push_sample(@current_sample[0], @current_sample[2])
end

#stream_index_for_name(name) ⇒ Object 



496
497
498
499
500
501
 
# File 'lib/orocos/log/replay.rb', line 496

def stream_index_for_name(name)
    if @stream
        return @stream.stream_index_for_name(name)
    end
    throw "Stream is not initialized yet"
end

#stream_index_for_type(name) ⇒ Object 



503
504
505
506
507
508
 
# File 'lib/orocos/log/replay.rb', line 503

def stream_index_for_type(name)
    if @stream
        return @stream.stream_index_for_type(name)
    end
    throw "Stream is not initialized yet"
end

#sync_step? ⇒ Boolean

returns ture if the next sample must be replayed to meet synchronous replay

Returns:

  • (Boolean) 


564
565
566
567
568
569
570
571
 
# File 'lib/orocos/log/replay.rb', line 564

def sync_step?
    calc_statistics
    if @out_of_sync_delta > 0.001
        false
    else
        true
    end
end

#task(name) ⇒ Object

Returns the simulated task with the given namen.



383
384
385
386
387
 
# File 'lib/orocos/log/replay.rb', line 383

def task(name)
    name2 = full_name(name)
    raise "cannot find TaskContext called #{name}" if !name2
    return @tasks[name2]
end

#task?(name) ⇒ Boolean

returns true if a task with the given name exists

Returns:

  • (Boolean) 


378
379
380
 
# File 'lib/orocos/log/replay.rb', line 378

def task?(name)
    !!full_name(name)
end

#tasks ⇒ Object

returns an array of all simulated tasks



215
216
217
 
# File 'lib/orocos/log/replay.rb', line 215

def tasks
    @tasks.values
end

#time ⇒ Object

Returns the time of the current sample.



220
221
222
 
# File 'lib/orocos/log/replay.rb', line 220

def time
    @base_time
end

#time_source ⇒ Object

Returns where from the time used for alignment should be taken. It can be one of

false

use the time at which the logger received the data (“logical time”)

:use_sample_time

for streams whose data contains a field called “time” of type base/Time (from Rock's base/types package), use the time contained in that field. Otherwise, use the logical time.

See #use_sample_time, #use_sample_time=



160
161
162
163
164
165
 
# File 'lib/orocos/log/replay.rb', line 160

def time_source
    if use_sample_time
        return :use_sample_time
    else return false
    end
end

#time_sync(&block) ⇒ Object

this can be used to set a different time sync logic the code block has three parameters time = current sample time actual_delta = time between start of replay and now required_delta = time which should have elapsed between start and now to replay at the desired speed the code block must return the number of seconds which the replay shall wait before the sample is repalyed

Do not block the program otherwise qt events are no longer processed!!!

Example time_sync do |time,actual_delta,required_delta|

my_object.busy? ? 1 : 0

end



552
553
554
555
556
557
558
559
560
 
# File 'lib/orocos/log/replay.rb', line 552

def time_sync(&block)
    if block_given?
        @time_sync_proc = block
    else
        @time_sync_proc = Proc.new do |time,actual_delta,required_delta|
            required_delta - actual_delta
        end
    end
end

#timestamp(type_name, &block) ⇒ Object

Declares how the timestamp can be extracted out of values of a given type

Examples:

use the 'time' field in /base/samples/RigidBodyState as timestamp

replay.timestamp '/base/samples/RigidBodyState' do |rbs|
  rbs.time
end


292
293
294
 
# File 'lib/orocos/log/replay.rb', line 292

def timestamp(type_name, &block)
    timestamps[type_name] = block
end

#track(value, filter = Hash.new) ⇒ Object

If set to true all ports are replayed and are not filtered out by the filter otherwise only ports are replayed which have a reader or a connection to an other port



300
301
302
303
304
305
 
# File 'lib/orocos/log/replay.rb', line 300

def track(value,filter=Hash.new)
    options, filter = Kernel::filter_options(filter,[:tasks])
    @tasks.each_value do |task|
        task.track(value,filter) if !options.has_key?(:tasks) || task.name =~ options[:tasks]
    end
end