Skip to content

Commit

Permalink
integrate options-server into test-options.py
Browse files Browse the repository at this point in the history
  • Loading branch information
@maloel
maloel committed Dec 14, 2023
1 parent 7908809 commit 12ab55d
Show file tree
Hide file tree
Showing 2 changed files with 104 additions and 116 deletions.
90 changes: 0 additions & 90 deletions unit-tests/dds/options-server.py
View file

This file was deleted.

130 changes: 104 additions & 26 deletions unit-tests/dds/test-options.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,27 +7,106 @@
import pyrealdds as dds
from rspy import log, test


dds.debug( log.is_debug_on(), 'C ' )
log.nested = 'C '


participant = dds.participant()
participant.init( 123, "test-options" )

info = dds.message.device_info()
info.name = "Test Device"
info.topic_root = "realdds/device/topic-root"
info.topic_root = "realsense/options-device"

with test.remote.fork( nested_indent=None ) as remote:
if remote is None: # we're the fork

dds.debug( log.is_debug_on(), log.nested )

participant = dds.participant()
participant.init( 123, "server" )


def test_no_options():
# Create one stream with one profile so device init won't fail
# No device options, no stream options
s1p1 = dds.video_stream_profile( 9, dds.video_encoding.rgb, 10, 10 )
s1profiles = [s1p1]
s1 = dds.depth_stream_server( "s1", "sensor" )
s1.init_profiles( s1profiles, 0 )
dev_opts = []
global server
server = dds.device_server( participant, info.topic_root )
server.init( [s1], dev_opts, {} )

def test_device_options_discovery( values ):
# Create one stream with one profile so device init won't fail, no stream options
s1p1 = dds.video_stream_profile( 9, dds.video_encoding.rgb, 10, 10 )
s1profiles = [s1p1]
s1 = dds.depth_stream_server( "s1", "sensor" )
s1.init_profiles( s1profiles, 0 )
dev_opts = []
for index, value in enumerate( values ):
option = dds.option( f'opt{index}', dds.option_range( value, value, 0, value ), f'opt{index} description' )
dev_opts.append( option )
global server
server = dds.device_server( participant, info.topic_root )
server.init( [s1], dev_opts, {})

def test_stream_options_discovery( value, min, max, step, default, description ):
s1p1 = dds.video_stream_profile( 9, dds.video_encoding.rgb, 10, 10 )
s1profiles = [s1p1]
s1 = dds.depth_stream_server( "s1", "sensor" )
s1.init_profiles( s1profiles, 0 )
so1 = dds.option( "opt1", dds.option_range( value, value, 0, value ), "opt1 is const" )
so1.set_value( value )
so2 = dds.option( "opt2", dds.option_range( min, max, step, default ), "opt2 with range" )
so3 = dds.option( "opt3", dds.option_range( 0, 1, 0.05, 0.15 ), description )
s1.init_options( [so1, so2, so3] )
global server
server = dds.device_server( participant, info.topic_root )
server.init( [s1], [], {} )

def test_device_and_multiple_stream_options_discovery( dev_values, stream_values ):
dev_options = []
for index, value in enumerate( dev_values ):
option = dds.option( f'opt{index}', dds.option_range( value, value, 0, value ), f'opt{index} description' )
dev_options.append( option )

s1p1 = dds.video_stream_profile( 9, dds.video_encoding.rgb, 10, 10 )
s1profiles = [s1p1]
s1 = dds.depth_stream_server( "s1", "sensor" )
s1.init_profiles( s1profiles, 0 )
stream_options = []
for index, value in enumerate( stream_values ):
option = dds.option( f'opt{index}', dds.option_range( value, value, 0, value ), f'opt{index} description' )
stream_options.append( option )
s1.init_options( stream_options )

s2p1 = dds.video_stream_profile( 9, dds.video_encoding.rgb, 10, 10 )
s2profiles = [s2p1]
s2 = dds.depth_stream_server( "s2", "sensor" )
s2.init_profiles( s2profiles, 0 )
stream_options = []
for index, value in enumerate( stream_values ):
option = dds.option( f'opt{index}', dds.option_range( value, value, 0, value ), f'opt{index} description' )
stream_options.append( option )
s2.init_options( stream_options )

global server
server = dds.device_server( participant, info.topic_root )
server.init( [s1, s2], dev_options, {} )

def close_server():
global server
server = None

raise StopIteration() # exit the 'with' statement


###############################################################################################################
#
dds.debug( log.is_debug_on(), 'C ' )
log.nested = 'C '

participant = dds.participant()
participant.init( 123, "test-options" )


import os.path
cwd = os.path.dirname(os.path.realpath(__file__))
remote_script = os.path.join( cwd, 'options-server.py' )
with test.remote( remote_script, nested_indent=" S" ) as remote:
remote.wait_until_ready()
#
#############################################################################################
#
with test.closure( "Test no options" ):
remote.run( 'test_no_options()' )
device = dds.device( participant, info )
Expand All @@ -45,9 +124,9 @@

remote.run( 'close_server()' )
device = None
#


#############################################################################################
#
with test.closure( "Test device options discovery" ):
test_values = list(range(17))
remote.run( 'test_device_options_discovery(' + str( test_values ) + ')' )
Expand All @@ -68,9 +147,9 @@

remote.run( 'close_server()' )
device = None
#


#############################################################################################
#
with test.closure( "Test stream options discovery" ):
#send values to be checked later as string parameter to the function
remote.run( 'test_stream_options_discovery(1, 0, 123456, 123, 12, "opt3 of s1")' )
Expand Down Expand Up @@ -99,9 +178,9 @@

remote.run( 'close_server()' )
device = None
#


#############################################################################################
#
with test.closure( "Test device and multiple stream options discovery" ):
test_values = list(range(5))
remote.run( 'test_device_and_multiple_stream_options_discovery(' + str( test_values ) + ', ' + str( test_values ) + ')' )
Expand All @@ -121,9 +200,8 @@

remote.run( 'close_server()' )
device = None
#
#############################################################################################
participant = None


participant = None
test.print_results_and_exit()
#############################################################################################
test.print_results()

0 comments on commit 12ab55d

Please sign in to comment.