Skip to content

Python Examples

The following examples demonstrate how the Sonardyne API Python library can be used to read, write, and stream data in and out of a device compatible with version of the Sonardyne API, which can be installed by running:

pip install git+https://github.com/Sonardyne/son-idl.git@2.1.0#subdirectory=libraries/python/sonardyne_api

See the Quick Start guide for information on setting up your device for use with the API.

Algorithms

The Algorithm component represents some internal algorithm e.g. an INS algorithm.

AHRS Algorithm

AHRS Algorithm Command Reset

Example code: ahrs_algorithm_command_reset.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.send_command(son.AhrsAlgorithmCommandReset()))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.send_command(son.AhrsAlgorithmCommandReset()))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.send_command(son.AhrsAlgorithmCommandReset()))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

AHRS Algorithm Configuration

Example code: ahrs_algorithm_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.AhrsAlgorithmConfiguration(
        coarse_position_latitude_radians=0.1,
        coarse_position_longitude_radians=0.2
    )))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.AhrsAlgorithmConfiguration(
        coarse_position_latitude_radians=0.1,
        coarse_position_longitude_radians=0.2
    )))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.AhrsAlgorithmConfiguration(
        coarse_position_latitude_radians=0.1,
        coarse_position_longitude_radians=0.2
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

DVL Algorithm

DVL Algorithm Configuration

Example code: dvl_algorithm_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.DvlAlgorithmConfiguration(
        dvl_rate=son.DvlRate(value=son.DvlRate.DvlRateEnum.DVL_RATE_ENUM_FIXED_5HZ),
        dvl_mode=son.DvlMode(value=son.DvlMode.DVL_MODE_ENUM_DVL),
        adcp_parameters=son.DvlAdcpParameters(
            number_of_cells=son.BoundedUInt32(value=100),
            preferred_cell_width_metres=son.BoundedDouble(value=0.1),
            pings_to_average=son.BoundedUInt32(value=10),
            dvl_adcp_ping_ratio=son.BoundedUInt32(value=10),
            velocity_limit=son.AdcpVelocityLimit(value=son.AdcpVelocityLimit.ADCP_VELOCITY_LIMIT_ENUM_STANDARD)
        ),
        input_trigger_parameters=son.DvlInputTriggerParameters(
            input_trigger_port=son.TriggerPortReference(set_uid=son.UniqueID(name="Trigger A1")),
            input_trigger_edge=son.TRIGGER_EDGE_RISING,
            input_trigger_delay_seconds=son.BoundedDouble(value=0.1)
        ),
        dvl_blanking_distance_metres=son.BoundedDouble(value=5.687)
    )))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.DvlAlgorithmConfiguration(
        dvl_rate=son.DvlRate(value=son.DvlRate.DvlRateEnum.DVL_RATE_ENUM_FIXED_5HZ),
        dvl_mode=son.DvlMode(value=son.DvlMode.DVL_MODE_ENUM_DVL),
        adcp_parameters=son.DvlAdcpParameters(
            number_of_cells=son.BoundedUInt32(value=100),
            preferred_cell_width_metres=son.BoundedDouble(value=0.1),
            pings_to_average=son.BoundedUInt32(value=10),
            dvl_adcp_ping_ratio=son.BoundedUInt32(value=10),
            velocity_limit=son.AdcpVelocityLimit(value=son.AdcpVelocityLimit.ADCP_VELOCITY_LIMIT_ENUM_STANDARD)
        ),
        input_trigger_parameters=son.DvlInputTriggerParameters(
            input_trigger_port=son.TriggerPortReference(set_uid=son.UniqueID(name="Trigger A1")),
            input_trigger_edge=son.TRIGGER_EDGE_RISING,
            input_trigger_delay_seconds=son.BoundedDouble(value=0.1)
        ),
        dvl_blanking_distance_metres=son.BoundedDouble(value=5.687)
    )))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.DvlAlgorithmConfiguration(
        dvl_rate=son.DvlRate(value=son.DvlRate.DvlRateEnum.DVL_RATE_ENUM_FIXED_5HZ),
        dvl_mode=son.DvlMode(value=son.DvlMode.DVL_MODE_ENUM_DVL),
        adcp_parameters=son.DvlAdcpParameters(
            number_of_cells=son.BoundedUInt32(value=100),
            preferred_cell_width_metres=son.BoundedDouble(value=0.1),
            pings_to_average=son.BoundedUInt32(value=10),
            dvl_adcp_ping_ratio=son.BoundedUInt32(value=10),
            velocity_limit=son.AdcpVelocityLimit(value=son.AdcpVelocityLimit.ADCP_VELOCITY_LIMIT_ENUM_STANDARD)
        ),
        input_trigger_parameters=son.DvlInputTriggerParameters(
            input_trigger_port=son.TriggerPortReference(set_uid=son.UniqueID(name="Trigger A1")),
            input_trigger_edge=son.TRIGGER_EDGE_RISING,
            input_trigger_delay_seconds=son.BoundedDouble(value=0.1)
        ),
        dvl_blanking_distance_metres=son.BoundedDouble(value=5.687)
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

INS Algorithm

INS Algorithm Command Reset

Example code: ins_algorithm_command_reset.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.send_command(son.InsAlgorithmCommandReset()))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.send_command(son.InsAlgorithmCommandReset()))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.send_command(son.InsAlgorithmCommandReset()))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

INS Algorithm Configuration

Example code: ins_algorithm_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.InsAlgorithmConfiguration(
        is_gnss_enabled=True,
        is_xpos_enabled=False,
        is_external_depth_enabled=False,
        is_internal_depth_enabled=False,
        is_gnss_height_enabled=True
    )))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.InsAlgorithmConfiguration(
        is_gnss_enabled=True,
        is_xpos_enabled=False,
        is_external_depth_enabled=False,
        is_internal_depth_enabled=False,
        is_gnss_height_enabled=True
    )))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.InsAlgorithmConfiguration(
        is_gnss_enabled=True,
        is_xpos_enabled=False,
        is_external_depth_enabled=False,
        is_internal_depth_enabled=False,
        is_gnss_height_enabled=True
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

INS Algorithm Observation

Example code: ins_algorithm_observation.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(
            observation_subscriptions=[
                son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="AidingObservation"))
            ]
        )
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(
            observation_subscriptions=[
                son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="AidingObservation"))
            ]
        )
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(
            observation_subscriptions=[
                son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="AidingObservation"))
            ]
        )
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))
while True:
    if data := socket.recv(10240):
        print(data.decode())

Modules

The Module component represents some non-algorithm, non-sensor, non-data source component e.g. a BatteryModule which is responsible for managing a battery and enabling/disabling charging.

Battery Module

Battery Module Configuration

Example code: battery_module_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.BatteryModuleConfiguration(
        is_battery_charging_enabled=False
    )))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.BatteryModuleConfiguration(
        is_battery_charging_enabled=False
    )))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.BatteryModuleConfiguration(
        is_battery_charging_enabled=False
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Device Module

Device Module Command Factory Reset

Example code: device_module_command_factory_reset.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.send_command(son.DeviceModuleCommandFactoryReset()))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.send_command(son.DeviceModuleCommandFactoryReset()))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.send_command(son.DeviceModuleCommandFactoryReset()))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Device Module Command Shutdown

Example code: device_module_command_shutdown.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.send_command(son.DeviceModuleCommandShutdown()))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.send_command(son.DeviceModuleCommandShutdown()))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.send_command(son.DeviceModuleCommandShutdown()))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Device Module Status

Example code: device_module_status.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    status_stream = wrapper.open_status_stream(
        son.StatusSubscriptionRequest(status_subscriptions=[
            son.StatusSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="DeviceModuleStatus"))
        ])
    )
    print("\n(listening to status stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in status_stream.recv():
                print(type(message), message)
                print("(listening to status stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    status_stream = wrapper.open_status_stream(
        son.StatusSubscriptionRequest(status_subscriptions=[
            son.StatusSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="DeviceModuleStatus"))
        ])
    )
    print("\n(listening to status stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in status_stream.recv():
                print(type(message), message)
                print("(listening to status stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    status_stream = wrapper.open_status_stream(
        son.StatusSubscriptionRequest(status_subscriptions=[
            son.StatusSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="DeviceModuleStatus"))
        ])
    )
    print("\n(listening to status stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in status_stream.recv():
                print(type(message), message)
                print("(listening to status stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))
while True:
    if data := socket.recv(10240):
        print(data.decode())

External Logging Module

External Logging Module Configuration

Example code: external_logging_module_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.ExternalLoggingModuleConfiguration(
        ethernet_logging_port=son.EthernetPort(
            source_port=8105,
            tcp_server=son.TcpServerParameters()
        )
    )))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.ExternalLoggingModuleConfiguration(
        ethernet_logging_port=son.EthernetPort(
            source_port=8105,
            tcp_server=son.TcpServerParameters()
        )
    )))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.ExternalLoggingModuleConfiguration(
        ethernet_logging_port=son.EthernetPort(
            source_port=8105,
            tcp_server=son.TcpServerParameters()
        )
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Navigation Module Observation

Example code: navigation_module_observation.py

import sonardyne_api as son
import time

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(
            observation_subscriptions=[
                son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="NavigationModuleObservation"))
            ]
        )
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
                print(time.time())
    except KeyboardInterrupt:
        pass

import sonardyne_api as son
import time

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(
            observation_subscriptions=[
                son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="NavigationModuleObservation"))
            ]
        )
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
                print(time.time())
    except KeyboardInterrupt:
        pass

import sonardyne_api as son
import time

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(
            observation_subscriptions=[
                son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="NavigationModuleObservation"))
            ]
        )
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
                print(time.time())
    except KeyboardInterrupt:
        pass

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))
while True:
    if data := socket.recv(10240):
        print(data.decode())

Network Module

Network Module Configuration

Example code: network_module_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.NetworkModuleConfiguration(
        subnet_mask="255.255.255.0",
        ip_address_list=son.IPAddressList(ip_addresses=["192.168.179.56"]),
        gateway_ip="192.168.179.1",
        dhcp_enabled=False
    )))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.NetworkModuleConfiguration(
        subnet_mask="255.255.255.0",
        ip_address_list=son.IPAddressList(ip_addresses=["192.168.179.56"]),
        gateway_ip="192.168.179.1",
        dhcp_enabled=False
    )))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.NetworkModuleConfiguration(
        subnet_mask="255.255.255.0",
        ip_address_list=son.IPAddressList(ip_addresses=["192.168.179.56"]),
        gateway_ip="192.168.179.1",
        dhcp_enabled=False
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Output Message Module

Output Message Module Command Add Output

Example code: output_message_module_command_add_output.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    wrapper.send_command(son.OutputMessageModuleCommandAddOutput(
        output_message=
        son.OutputMessage(
            output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_GGA,
            rate_hz=20.0,
            remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Remote Point 3")),
            data_port_output=son.DataPortReference(set_any_ethernet_port=True)
        )
    ))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    wrapper.send_command(son.OutputMessageModuleCommandAddOutput(
        output_message=
        son.OutputMessage(
            output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_GGA,
            rate_hz=20.0,
            remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Remote Point 3")),
            data_port_output=son.DataPortReference(set_any_ethernet_port=True)
        )
    ))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    wrapper.send_command(son.OutputMessageModuleCommandAddOutput(
        output_message=
        son.OutputMessage(
            output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_GGA,
            rate_hz=20.0,
            remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Remote Point 3")),
            data_port_output=son.DataPortReference(set_any_ethernet_port=True)
        )
    ))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Output Message Module Command Remove Output

Example code: output_message_module_command_remove_output.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    wrapper.set_configuration(son.OutputMessageModuleConfiguration(
        output_message_list=son.OutputMessageList(
            output_messages=[
                son.OutputMessage(
                    output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_GGA,
                    rate_hz=10.0,
                    remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Remote Point 2")),
                    data_port_output=son.DataPortReference(set_any_ethernet_port=True)
                ),
                son.OutputMessage(
                    output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_LNAV,
                    rate_hz=5.0,
                    remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Common Reference Point")),
                    data_port_output=son.DataPortReference(set_any_ethernet_port=True)
                )
            ]
        )
    ))
    wrapper.send_command(son.OutputMessageModuleCommandRemoveOutput(
        output_message=son.OutputMessage(uid=son.UniqueID(name="LNAV Output Message"))
    ))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    wrapper.set_configuration(son.OutputMessageModuleConfiguration(
        output_message_list=son.OutputMessageList(
            output_messages=[
                son.OutputMessage(
                    output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_GGA,
                    rate_hz=10.0,
                    remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Remote Point 2")),
                    data_port_output=son.DataPortReference(set_any_ethernet_port=True)
                ),
                son.OutputMessage(
                    output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_LNAV,
                    rate_hz=5.0,
                    remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Common Reference Point")),
                    data_port_output=son.DataPortReference(set_any_ethernet_port=True)
                )
            ]
        )
    ))
    wrapper.send_command(son.OutputMessageModuleCommandRemoveOutput(
        output_message=son.OutputMessage(uid=son.UniqueID(name="LNAV Output Message"))
    ))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    wrapper.set_configuration(son.OutputMessageModuleConfiguration(
        output_message_list=son.OutputMessageList(
            output_messages=[
                son.OutputMessage(
                    output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_GGA,
                    rate_hz=10.0,
                    remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Remote Point 2")),
                    data_port_output=son.DataPortReference(set_any_ethernet_port=True)
                ),
                son.OutputMessage(
                    output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_LNAV,
                    rate_hz=5.0,
                    remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Common Reference Point")),
                    data_port_output=son.DataPortReference(set_any_ethernet_port=True)
                )
            ]
        )
    ))
    wrapper.send_command(son.OutputMessageModuleCommandRemoveOutput(
        output_message=son.OutputMessage(uid=son.UniqueID(name="LNAV Output Message"))
    ))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Output Message Module Configuration

Example code: output_message_module_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.OutputMessageModuleConfiguration(
        output_message_list=son.OutputMessageList(
            output_messages=[
                son.OutputMessage(
                    output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_GGA,
                    rate_hz=10.0,
                    remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Remote Point 2")),
                    data_port_output=son.DataPortReference(set_any_ethernet_port=True)
                ),
                son.OutputMessage(
                    output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_LNAV,
                    rate_hz=5.0,
                    remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Common Reference Point")),
                    data_port_output=son.DataPortReference(set_any_ethernet_port=True)
                )
            ]
        )
    )))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.OutputMessageModuleConfiguration(
        output_message_list=son.OutputMessageList(
            output_messages=[
                son.OutputMessage(
                    output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_GGA,
                    rate_hz=10.0,
                    remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Remote Point 2")),
                    data_port_output=son.DataPortReference(set_any_ethernet_port=True)
                ),
                son.OutputMessage(
                    output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_LNAV,
                    rate_hz=5.0,
                    remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Common Reference Point")),
                    data_port_output=son.DataPortReference(set_any_ethernet_port=True)
                )
            ]
        )
    )))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.OutputMessageModuleConfiguration(
        output_message_list=son.OutputMessageList(
            output_messages=[
                son.OutputMessage(
                    output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_GGA,
                    rate_hz=10.0,
                    remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Remote Point 2")),
                    data_port_output=son.DataPortReference(set_any_ethernet_port=True)
                ),
                son.OutputMessage(
                    output_message_type=son.OutputMessageType.OUTPUT_MESSAGE_TYPE_LNAV,
                    rate_hz=5.0,
                    remote_point=son.RemotePointReference(set_uid=son.UniqueID(name="Common Reference Point")),
                    data_port_output=son.DataPortReference(set_any_ethernet_port=True)
                )
            ]
        )
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Time Module

Time Module Configuration

Example code: time_module_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.TimeModuleConfiguration(
        input_time_sources=son.TimeSourceInputReferences(
            input_ntp_time_source=son.NtpTimeSourceReference(set_uid=son.UniqueID(name="NTP Time Source")),
            input_zda_time_source=son.ZdaTimeSourceReference(set_none=True),
            input_pps_time_source=son.PpsTimeSourceReference(set_none=True)
        )
    )))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.TimeModuleConfiguration(
        input_time_sources=son.TimeSourceInputReferences(
            input_ntp_time_source=son.NtpTimeSourceReference(set_uid=son.UniqueID(name="NTP Time Source")),
            input_zda_time_source=son.ZdaTimeSourceReference(set_none=True),
            input_pps_time_source=son.PpsTimeSourceReference(set_none=True)
        )
    )))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.TimeModuleConfiguration(
        input_time_sources=son.TimeSourceInputReferences(
            input_ntp_time_source=son.NtpTimeSourceReference(set_uid=son.UniqueID(name="NTP Time Source")),
            input_zda_time_source=son.ZdaTimeSourceReference(set_none=True),
            input_pps_time_source=son.PpsTimeSourceReference(set_none=True)
        )
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Ports

The Port List component represents some collection of data inputs/outputs.

Some port lists, e.g. ethernet port lists, can have entries added and removed. Other port lists, e.g. serial port lists, represent physical ports on the device, so are fixed-length.

On a data port such as a SerialPort or an EthernetPort, it is possible to send and receive PortComms using the CommsService.

Ethernet Port

Ethernet Port List Configuration

Example code: ethernet_port_list_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as device: # SPRINT-Nav ip address & port
    print(device.set_configuration(son.EthernetPortListConfiguration(ethernet_ports=[
        son.EthernetPort(source_port=8110, tcp_server=son.TcpServerParameters()),
        son.EthernetPort(source_port=8111, tcp_client=son.TcpClientParameters(server_ip_address="192.168.179.60")),
        son.EthernetPort(source_port=8112, udp_socket=son.UdpSocketParameters(destination_ip_address="192.168.179.70")),
    ])))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as device: # SPRINT-Nav ip address & port
    print(device.set_configuration(son.EthernetPortListConfiguration(ethernet_ports=[
        son.EthernetPort(source_port=8110, tcp_server=son.TcpServerParameters()),
        son.EthernetPort(source_port=8111, tcp_client=son.TcpClientParameters(server_ip_address="192.168.179.60")),
        son.EthernetPort(source_port=8112, udp_socket=son.UdpSocketParameters(destination_ip_address="192.168.179.70")),
    ])))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as device: # SPRINT-Nav ip address & port
    print(device.set_configuration(son.EthernetPortListConfiguration(ethernet_ports=[
        son.EthernetPort(source_port=8110, tcp_server=son.TcpServerParameters()),
        son.EthernetPort(source_port=8111, tcp_client=son.TcpClientParameters(server_ip_address="192.168.179.60")),
        son.EthernetPort(source_port=8112, udp_socket=son.UdpSocketParameters(destination_ip_address="192.168.179.70")),
    ])))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

External Control Port

External Control Port List Configuration

Example code: external_control_port_list_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.get_configuration(son.ExternalControlPortListConfiguration))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.get_configuration(son.ExternalControlPortListConfiguration))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.get_configuration(son.ExternalControlPortListConfiguration))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Power Pass Port

Power Pass Port List Configuration

Example code: power_pass_port_list_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.PowerPassPortListConfiguration(power_pass_ports=[
        son.PowerPassPort(power_pass_state=son.PowerPassPort.POWER_PASS_STATE_TRIPPED),
        son.PowerPassPort(power_pass_state=son.PowerPassPort.POWER_PASS_STATE_ENABLED)
    ])))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.PowerPassPortListConfiguration(power_pass_ports=[
        son.PowerPassPort(power_pass_state=son.PowerPassPort.POWER_PASS_STATE_TRIPPED),
        son.PowerPassPort(power_pass_state=son.PowerPassPort.POWER_PASS_STATE_ENABLED)
    ])))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.PowerPassPortListConfiguration(power_pass_ports=[
        son.PowerPassPort(power_pass_state=son.PowerPassPort.POWER_PASS_STATE_TRIPPED),
        son.PowerPassPort(power_pass_state=son.PowerPassPort.POWER_PASS_STATE_ENABLED)
    ])))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Serial Port

Serial Port List Configuration

Example code: serial_port_list_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.SerialPortListConfiguration(serial_ports=[
        son.SerialPort(baud_rate=son.SerialPort.BAUD_RATE_9600),
        son.SerialPort(baud_rate=son.SerialPort.BAUD_RATE_38400),
    ])))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.SerialPortListConfiguration(serial_ports=[
        son.SerialPort(baud_rate=son.SerialPort.BAUD_RATE_9600),
        son.SerialPort(baud_rate=son.SerialPort.BAUD_RATE_38400),
    ])))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.SerialPortListConfiguration(serial_ports=[
        son.SerialPort(baud_rate=son.SerialPort.BAUD_RATE_9600),
        son.SerialPort(baud_rate=son.SerialPort.BAUD_RATE_38400),
    ])))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Trigger Port

Trigger Port List Configuration

Example code: trigger_port_list_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.get_configuration(son.TriggerPortListConfiguration))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.get_configuration(son.TriggerPortListConfiguration))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.get_configuration(son.TriggerPortListConfiguration))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Sensors

The Sensor component represents some physical device which observes data.

The API can be used to get and set the configuration of a particular sensor. This could include, for example, enabling raw data logging for a Doppler Velocity Log (DVL).

Sensors can also send and receive SensorsObservations using the ObservationService.

DVL Sensor

DVL Sensor Configuration

Example code: dvl_sensor_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.DvlSensorConfiguration(
        is_raw_logging_enabled=False
    )))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.DvlSensorConfiguration(
        is_raw_logging_enabled=False
    )))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.DvlSensorConfiguration(
        is_raw_logging_enabled=False
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Pressure Sensor

Pressure Sensor Configuration

Example code: pressure_sensor_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.PressureSensorConfiguration(
        manual_depth_offset_metres=2
    )))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.PressureSensorConfiguration(
        manual_depth_offset_metres=2
    )))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.PressureSensorConfiguration(
        manual_depth_offset_metres=2
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Services

Command Service

Example code: command_service.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    wrapper.send_command(son.InsAlgorithmCommandReset())

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    wrapper.send_command(son.InsAlgorithmCommandReset())

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    wrapper.send_command(son.InsAlgorithmCommandReset())

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Comms Service

Example code: comms_service.py

import sonardyne_api as son
import time
import random

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    comms_stream  = wrapper.open_comms_stream(
        son.CommsSubscriptionRequest(comms_subscriptions=[
            son.CommsSubscription(matching_criteria=son.MatchingCriteria(match_uids=[son.UniqueID(name="TCP-8110")]))
        ])
    )

    # write to stream
    print("\n(writing to comms stream) press ctrl-c to stop stream")
    try:
        while True:
            comms_stream.send(
                son.DataPortComms(
                    uid=son.UniqueID(name="TCP-8110"),
                    data_direction=son.DataDirection.DATA_DIRECTION_OUTPUT,
                    data=f"Hello World! {random.randrange(100)}".encode()
                )
            )
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    print("\n(listening to comms stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in comms_stream.recv():
                print(type(message), message)
                print("(listening to comms stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son
import time
import random

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    comms_stream  = wrapper.open_comms_stream(
        son.CommsSubscriptionRequest(comms_subscriptions=[
            son.CommsSubscription(matching_criteria=son.MatchingCriteria(match_uids=[son.UniqueID(name="TCP-8110")]))
        ])
    )

    # write to stream
    print("\n(writing to comms stream) press ctrl-c to stop stream")
    try:
        while True:
            comms_stream.send(
                son.DataPortComms(
                    uid=son.UniqueID(name="TCP-8110"),
                    data_direction=son.DataDirection.DATA_DIRECTION_OUTPUT,
                    data=f"Hello World! {random.randrange(100)}".encode()
                )
            )
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    print("\n(listening to comms stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in comms_stream.recv():
                print(type(message), message)
                print("(listening to comms stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son
import time
import random

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    comms_stream  = wrapper.open_comms_stream(
        son.CommsSubscriptionRequest(comms_subscriptions=[
            son.CommsSubscription(matching_criteria=son.MatchingCriteria(match_uids=[son.UniqueID(name="TCP-8110")]))
        ])
    )

    # write to stream
    print("\n(writing to comms stream) press ctrl-c to stop stream")
    try:
        while True:
            comms_stream.send(
                son.DataPortComms(
                    uid=son.UniqueID(name="TCP-8110"),
                    data_direction=son.DataDirection.DATA_DIRECTION_OUTPUT,
                    data=f"Hello World! {random.randrange(100)}".encode()
                )
            )
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    print("\n(listening to comms stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in comms_stream.recv():
                print(type(message), message)
                print("(listening to comms stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))
while True:
    if data := socket.recv(10240):
        print(data.decode())

Configuration Service

Example code: configuration_service.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  

    print("all configurations:")
    configurations = wrapper.get_configurations("Configuration")
    for configuration in configurations:
        print(f"    • {type(configuration).__name__}")

    print("all configurations ending in SourceConfiguration:")
    configurations = wrapper.get_configurations("SourceConfiguration")
    for configuration in configurations:
        print(f"    • {type(configuration).__name__}")

    configuration_stream = wrapper.open_configuration_stream(
        son.ConfigurationSubscriptionRequest(configuration_subscriptions=[
            son.ConfigurationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="Configuration"))
        ])
    )
    print("\n(listening to configuration stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in configuration_stream.recv():
                print(type(message), message)
                print("(listening to configuration stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  

    print("all configurations:")
    configurations = wrapper.get_configurations("Configuration")
    for configuration in configurations:
        print(f"    • {type(configuration).__name__}")

    print("all configurations ending in SourceConfiguration:")
    configurations = wrapper.get_configurations("SourceConfiguration")
    for configuration in configurations:
        print(f"    • {type(configuration).__name__}")

    configuration_stream = wrapper.open_configuration_stream(
        son.ConfigurationSubscriptionRequest(configuration_subscriptions=[
            son.ConfigurationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="Configuration"))
        ])
    )
    print("\n(listening to configuration stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in configuration_stream.recv():
                print(type(message), message)
                print("(listening to configuration stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  

    print("all configurations:")
    configurations = wrapper.get_configurations("Configuration")
    for configuration in configurations:
        print(f"    • {type(configuration).__name__}")

    print("all configurations ending in SourceConfiguration:")
    configurations = wrapper.get_configurations("SourceConfiguration")
    for configuration in configurations:
        print(f"    • {type(configuration).__name__}")

    configuration_stream = wrapper.open_configuration_stream(
        son.ConfigurationSubscriptionRequest(configuration_subscriptions=[
            son.ConfigurationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="Configuration"))
        ])
    )
    print("\n(listening to configuration stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in configuration_stream.recv():
                print(type(message), message)
                print("(listening to configuration stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))
while True:
    if data := socket.recv(10240):
        print(data.decode())

Observation Service

Example code: observation_service.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="GnssSourceObservation")),
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="XposSourceObservation")),
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="SusblSourceObservation")),
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="GnssSourceObservation")),
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="XposSourceObservation")),
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="SusblSourceObservation")),
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="GnssSourceObservation")),
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="XposSourceObservation")),
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="SusblSourceObservation")),
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))
while True:
    if data := socket.recv(10240):
        print(data.decode())

Status Service

Example code: status_service.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    status_stream = wrapper.open_status_stream(
        son.StatusSubscriptionRequest(status_subscriptions=[
            son.StatusSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="InsAlgorithmStatus"))
        ])
    )
    print("\n(listening to status stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in status_stream.recv():
                print(type(message), message)
                print("(listening to status stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    status_stream = wrapper.open_status_stream(
        son.StatusSubscriptionRequest(status_subscriptions=[
            son.StatusSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="InsAlgorithmStatus"))
        ])
    )
    print("\n(listening to status stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in status_stream.recv():
                print(type(message), message)
                print("(listening to status stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    status_stream = wrapper.open_status_stream(
        son.StatusSubscriptionRequest(status_subscriptions=[
            son.StatusSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="InsAlgorithmStatus"))
        ])
    )
    print("\n(listening to status stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in status_stream.recv():
                print(type(message), message)
                print("(listening to status stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))
while True:
    if data := socket.recv(10240):
        print(data.decode())

Sources

The Source component represents some "aiding" or "input" source e.g. GNSS or Time.

The API can be used to get and set the configuration of a particular source. This could include, for example, the source's lever arms; the lever arms relate the position of the source to the central reference point of the vessel.

Sources can also send and receive SourceObservations using the ObservationService.

Depth Source

Depth Source Configuration

Example code: depth_source_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.DepthSourceConfiguration(
        input_depth_data_source_reference=son.DepthDataSourceReference(
            set_uid=son.UniqueID(name="TCP-8110")
        ),
        lever_arms=son.LeverArms(forward_metres=1.5, starboard_metres=2, down_metres=3.2),
        manual_depth_offset_metres=0
    )))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.DepthSourceConfiguration(
        input_depth_data_source_reference=son.DepthDataSourceReference(
            set_uid=son.UniqueID(name="TCP-8110")
        ),
        lever_arms=son.LeverArms(forward_metres=1.5, starboard_metres=2, down_metres=3.2),
        manual_depth_offset_metres=0
    )))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.DepthSourceConfiguration(
        input_depth_data_source_reference=son.DepthDataSourceReference(
            set_uid=son.UniqueID(name="TCP-8110")
        ),
        lever_arms=son.LeverArms(forward_metres=1.5, starboard_metres=2, down_metres=3.2),
        manual_depth_offset_metres=0
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Depth Source External Depth Calculation

Example code: depth_source_external_depth_calculation.py

import sonardyne_api as son

def pascals_to_depth(pressure_pa):
    """
    Converts pressure in Pascals to depth in meters.
    Default is sea water, set sea_water=False for fresh water.
    """
    g = 9.81  # Acceleration due to gravity (m/s^2)
    rho = 1000  # Density of fresh water (kg/m^3)
    depth = pressure_pa / (rho * g)
    return depth

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    # Set up depth source to use internal sensor with external depth conversion.
    wrapper.set_configuration(son.DepthSourceConfiguration(
        input_depth_data_source_reference=son.DepthDataSourceReference(set_any_pressure_sensor=True),
        pressure_to_depth_conversion=son.DepthSourceConfiguration.PRESSURE_TO_DEPTH_CONVERSION_EXTERNAL
    ))
    # Open stream to start listening for pressure sensor observations, so we can give them an externally calculated depth.
    observation_stream = wrapper.open_observation_stream(son.ObservationSubscriptionRequest(observation_subscriptions=[
        son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="PressureSensorObservation"))
    ]))
    try:
        while True:
            for observation in observation_stream.recv():
                if isinstance(observation, son.PressureSensorObservation):
                    # ignore messages which already have externally_calculated_depth_metres set.
                    if not observation.pressure_data.HasField("externally_calculated_depth_metres"):
                        depth = pascals_to_depth(observation.pressure_data.pressure_pascals)
                        observation.pressure_data.externally_calculated_depth_metres = depth + 10 # Add some offset so the difference is obvious.
                        observation_stream.send(observation)
    except KeyboardInterrupt:
        pass

import sonardyne_api as son

def pascals_to_depth(pressure_pa):
    """
    Converts pressure in Pascals to depth in meters.
    Default is sea water, set sea_water=False for fresh water.
    """
    g = 9.81  # Acceleration due to gravity (m/s^2)
    rho = 1000  # Density of fresh water (kg/m^3)
    depth = pressure_pa / (rho * g)
    return depth

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    # Set up depth source to use internal sensor with external depth conversion.
    wrapper.set_configuration(son.DepthSourceConfiguration(
        input_depth_data_source_reference=son.DepthDataSourceReference(set_any_pressure_sensor=True),
        pressure_to_depth_conversion=son.DepthSourceConfiguration.PRESSURE_TO_DEPTH_CONVERSION_EXTERNAL
    ))
    # Open stream to start listening for pressure sensor observations, so we can give them an externally calculated depth.
    observation_stream = wrapper.open_observation_stream(son.ObservationSubscriptionRequest(observation_subscriptions=[
        son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="PressureSensorObservation"))
    ]))
    try:
        while True:
            for observation in observation_stream.recv():
                if isinstance(observation, son.PressureSensorObservation):
                    # ignore messages which already have externally_calculated_depth_metres set.
                    if not observation.pressure_data.HasField("externally_calculated_depth_metres"):
                        depth = pascals_to_depth(observation.pressure_data.pressure_pascals)
                        observation.pressure_data.externally_calculated_depth_metres = depth + 10 # Add some offset so the difference is obvious.
                        observation_stream.send(observation)
    except KeyboardInterrupt:
        pass

import sonardyne_api as son

def pascals_to_depth(pressure_pa):
    """
    Converts pressure in Pascals to depth in meters.
    Default is sea water, set sea_water=False for fresh water.
    """
    g = 9.81  # Acceleration due to gravity (m/s^2)
    rho = 1000  # Density of fresh water (kg/m^3)
    depth = pressure_pa / (rho * g)
    return depth

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    # Set up depth source to use internal sensor with external depth conversion.
    wrapper.set_configuration(son.DepthSourceConfiguration(
        input_depth_data_source_reference=son.DepthDataSourceReference(set_any_pressure_sensor=True),
        pressure_to_depth_conversion=son.DepthSourceConfiguration.PRESSURE_TO_DEPTH_CONVERSION_EXTERNAL
    ))
    # Open stream to start listening for pressure sensor observations, so we can give them an externally calculated depth.
    observation_stream = wrapper.open_observation_stream(son.ObservationSubscriptionRequest(observation_subscriptions=[
        son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="PressureSensorObservation"))
    ]))
    try:
        while True:
            for observation in observation_stream.recv():
                if isinstance(observation, son.PressureSensorObservation):
                    # ignore messages which already have externally_calculated_depth_metres set.
                    if not observation.pressure_data.HasField("externally_calculated_depth_metres"):
                        depth = pascals_to_depth(observation.pressure_data.pressure_pascals)
                        observation.pressure_data.externally_calculated_depth_metres = depth + 10 # Add some offset so the difference is obvious.
                        observation_stream.send(observation)
    except KeyboardInterrupt:
        pass

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))
while True:
    if data := socket.recv(10240):
        print(data.decode())

External Depth Source

External Depth Source Configuration

Example code: external_depth_source_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.ExternalDepthSourceConfiguration(
        input_data_port=son.DataPortReference(set_uid=son.UniqueID(name="TCP-8110")),
        lever_arms=son.LeverArms(forward_metres=1.5, starboard_metres=2, down_metres=3.2),
        manual_depth_offset_metres=0,
        pressure_to_depth_conversion=son.PRESSURE_TO_DEPTH_CONVERSION_FIXED_SCALING
    )))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.ExternalDepthSourceConfiguration(
        input_data_port=son.DataPortReference(set_uid=son.UniqueID(name="TCP-8110")),
        lever_arms=son.LeverArms(forward_metres=1.5, starboard_metres=2, down_metres=3.2),
        manual_depth_offset_metres=0,
        pressure_to_depth_conversion=son.PRESSURE_TO_DEPTH_CONVERSION_FIXED_SCALING
    )))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.ExternalDepthSourceConfiguration(
        input_data_port=son.DataPortReference(set_uid=son.UniqueID(name="TCP-8110")),
        lever_arms=son.LeverArms(forward_metres=1.5, starboard_metres=2, down_metres=3.2),
        manual_depth_offset_metres=0,
        pressure_to_depth_conversion=son.PRESSURE_TO_DEPTH_CONVERSION_FIXED_SCALING
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

GNSS Source

GNSS Source Configuration

Example code: gnss_source_configuration.py

import sonardyne_api as son
import time

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.GnssSourceConfiguration(
        input_data_port=son.DataPortReference(
            set_uid=son.UniqueID(name="External Control")
        )
    )))

import sonardyne_api as son
import time

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.GnssSourceConfiguration(
        input_data_port=son.DataPortReference(
            set_uid=son.UniqueID(name="External Control")
        )
    )))

import sonardyne_api as son
import time

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.GnssSourceConfiguration(
        input_data_port=son.DataPortReference(
            set_uid=son.UniqueID(name="External Control")
        )
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

GNSS Source Observation

Example code: gnss_source_observation.py

import sonardyne_api as son
import time

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation_stream.send(son.GnssSourceObservation(gga_telegram=son.GgaTelegram(
                time_of_validity=son.Timestamp(common_time_seconds=time.time()),
                latitude_radians=0.0,
                longitude_radians=0.0,
                altitude_metres=0.0,
                geoid_separation_metres=0.0,
                hdop=0.5,
                age_seconds=0.0,
                number_of_satellites=10,
                reference_station_id=None,
                fix_quality=son.GgaTelegram.FIX_QUALITY_GNSS_FIX
            )))
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="GnssSourceObservation"))
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son
import time

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation_stream.send(son.GnssSourceObservation(gga_telegram=son.GgaTelegram(
                time_of_validity=son.Timestamp(common_time_seconds=time.time()),
                latitude_radians=0.0,
                longitude_radians=0.0,
                altitude_metres=0.0,
                geoid_separation_metres=0.0,
                hdop=0.5,
                age_seconds=0.0,
                number_of_satellites=10,
                reference_station_id=None,
                fix_quality=son.GgaTelegram.FIX_QUALITY_GNSS_FIX
            )))
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="GnssSourceObservation"))
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son
import time

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation_stream.send(son.GnssSourceObservation(gga_telegram=son.GgaTelegram(
                time_of_validity=son.Timestamp(common_time_seconds=time.time()),
                latitude_radians=0.0,
                longitude_radians=0.0,
                altitude_metres=0.0,
                geoid_separation_metres=0.0,
                hdop=0.5,
                age_seconds=0.0,
                number_of_satellites=10,
                reference_station_id=None,
                fix_quality=son.GgaTelegram.FIX_QUALITY_GNSS_FIX
            )))
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="GnssSourceObservation"))
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))
while True:
    if data := socket.recv(10240):
        print(data.decode())

Internal Depth Source

Internal Depth Source Configuration

Example code: internal_depth_source_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.InternalDepthSourceConfiguration(
        manual_depth_offset_metres=1,
        pressure_to_depth_conversion=son.PRESSURE_TO_DEPTH_CONVERSION_FIXED_SCALING
    )))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.InternalDepthSourceConfiguration(
        manual_depth_offset_metres=1,
        pressure_to_depth_conversion=son.PRESSURE_TO_DEPTH_CONVERSION_FIXED_SCALING
    )))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.InternalDepthSourceConfiguration(
        manual_depth_offset_metres=1,
        pressure_to_depth_conversion=son.PRESSURE_TO_DEPTH_CONVERSION_FIXED_SCALING
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Sound Velocity Source

Sound Velocity Source Configuration

Example code: sound_velocity_source_configuration.py

import sonardyne_api as son
import time
import random

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.SoundVelocitySourceConfiguration(
        input_sound_velocity_data_source_reference=son.SoundVelocityDataSourceReference(set_salinity_parts_per_thousand=100)))
    )
    print(wrapper.set_configuration(son.SoundVelocitySourceConfiguration(
        input_sound_velocity_data_source_reference=son.SoundVelocityDataSourceReference(set_manual_sound_velocity_metres_per_second=150)))
    )
    print(wrapper.set_configuration(son.SoundVelocitySourceConfiguration(
        input_sound_velocity_data_source_reference=son.SoundVelocityDataSourceReference(set_any_ethernet_port=True)))
    )
    print(wrapper.set_configuration(son.SoundVelocitySourceConfiguration(
        input_sound_velocity_data_source_reference=son.SoundVelocityDataSourceReference(set_any_external_control_port=True)))
    )

import sonardyne_api as son
import time
import random

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.SoundVelocitySourceConfiguration(
        input_sound_velocity_data_source_reference=son.SoundVelocityDataSourceReference(set_salinity_parts_per_thousand=100)))
    )
    print(wrapper.set_configuration(son.SoundVelocitySourceConfiguration(
        input_sound_velocity_data_source_reference=son.SoundVelocityDataSourceReference(set_manual_sound_velocity_metres_per_second=150)))
    )
    print(wrapper.set_configuration(son.SoundVelocitySourceConfiguration(
        input_sound_velocity_data_source_reference=son.SoundVelocityDataSourceReference(set_any_ethernet_port=True)))
    )
    print(wrapper.set_configuration(son.SoundVelocitySourceConfiguration(
        input_sound_velocity_data_source_reference=son.SoundVelocityDataSourceReference(set_any_external_control_port=True)))
    )

import sonardyne_api as son
import time
import random

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.SoundVelocitySourceConfiguration(
        input_sound_velocity_data_source_reference=son.SoundVelocityDataSourceReference(set_salinity_parts_per_thousand=100)))
    )
    print(wrapper.set_configuration(son.SoundVelocitySourceConfiguration(
        input_sound_velocity_data_source_reference=son.SoundVelocityDataSourceReference(set_manual_sound_velocity_metres_per_second=150)))
    )
    print(wrapper.set_configuration(son.SoundVelocitySourceConfiguration(
        input_sound_velocity_data_source_reference=son.SoundVelocityDataSourceReference(set_any_ethernet_port=True)))
    )
    print(wrapper.set_configuration(son.SoundVelocitySourceConfiguration(
        input_sound_velocity_data_source_reference=son.SoundVelocityDataSourceReference(set_any_external_control_port=True)))
    )

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Sound Velocity Source Observation

Example code: sound_velocity_source_observation.py

import sonardyne_api as son
import time
import random

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation = son.SoundVelocitySourceObservation(sound_velocity_data=son.SoundVelocityData(
                time_of_validity=son.Timestamp(common_time_seconds=time.time()),
                sound_velocity_meters_per_second=1500.0 + random.randrange(-500,500) * 0.01
            ))
            observation_stream.send(observation)
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="SoundVelocitySourceObservation"))
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son
import time
import random

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation = son.SoundVelocitySourceObservation(sound_velocity_data=son.SoundVelocityData(
                time_of_validity=son.Timestamp(common_time_seconds=time.time()),
                sound_velocity_meters_per_second=1500.0 + random.randrange(-500,500) * 0.01
            ))
            observation_stream.send(observation)
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="SoundVelocitySourceObservation"))
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son
import time
import random

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation = son.SoundVelocitySourceObservation(sound_velocity_data=son.SoundVelocityData(
                time_of_validity=son.Timestamp(common_time_seconds=time.time()),
                sound_velocity_meters_per_second=1500.0 + random.randrange(-500,500) * 0.01
            ))
            observation_stream.send(observation)
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="SoundVelocitySourceObservation"))
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))
while True:
    if data := socket.recv(10240):
        print(data.decode())

SUSBL Source

SUSBL Source Configuration

Example code: susbl_source_configuration.py

import sonardyne_api as son
import time

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.SusblSourceConfiguration(input_data_port=son.DataPortReference(set_uid=son.UniqueID(name="External Control")))))

import sonardyne_api as son
import time

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.SusblSourceConfiguration(input_data_port=son.DataPortReference(set_uid=son.UniqueID(name="External Control")))))

import sonardyne_api as son
import time

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.SusblSourceConfiguration(input_data_port=son.DataPortReference(set_uid=son.UniqueID(name="External Control")))))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

SUSBL Source Observation

Example code: susbl_source_observation.py

import sonardyne_api as son
import time

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation_stream.send(son.SusblSourceObservation(psimssb_telegram=son.PsimssbTelegram(
                time_of_validity=son.Timestamp(common_time_seconds=time.time()),
                coordinate_type=son.PsimssbTelegram.PSIMSSB_COORDINATE_SYSTEM_RADIANS,
                x_coordinate=0,
                y_coordinate=0,
                transponder_code=0,
                is_valid=True,
                orientation_type=son.PsimssbTelegram.PSIMSSB_ORIENTATION_NORTH,
                error_code="",
                software_filter=son.PsimssbTelegram.PSIMSSB_SOFTWARE_FILTER_UNKNOWN,
                depth_metres=0,
                position_uncertainty_metres=1
            )))
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="SusblSourceObservation"))
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son
import time

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation_stream.send(son.SusblSourceObservation(psimssb_telegram=son.PsimssbTelegram(
                time_of_validity=son.Timestamp(common_time_seconds=time.time()),
                coordinate_type=son.PsimssbTelegram.PSIMSSB_COORDINATE_SYSTEM_RADIANS,
                x_coordinate=0,
                y_coordinate=0,
                transponder_code=0,
                is_valid=True,
                orientation_type=son.PsimssbTelegram.PSIMSSB_ORIENTATION_NORTH,
                error_code="",
                software_filter=son.PsimssbTelegram.PSIMSSB_SOFTWARE_FILTER_UNKNOWN,
                depth_metres=0,
                position_uncertainty_metres=1
            )))
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="SusblSourceObservation"))
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son
import time

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation_stream.send(son.SusblSourceObservation(psimssb_telegram=son.PsimssbTelegram(
                time_of_validity=son.Timestamp(common_time_seconds=time.time()),
                coordinate_type=son.PsimssbTelegram.PSIMSSB_COORDINATE_SYSTEM_RADIANS,
                x_coordinate=0,
                y_coordinate=0,
                transponder_code=0,
                is_valid=True,
                orientation_type=son.PsimssbTelegram.PSIMSSB_ORIENTATION_NORTH,
                error_code="",
                software_filter=son.PsimssbTelegram.PSIMSSB_SOFTWARE_FILTER_UNKNOWN,
                depth_metres=0,
                position_uncertainty_metres=1
            )))
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="SusblSourceObservation"))
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))
while True:
    if data := socket.recv(10240):
        print(data.decode())

Example code: time_module_configuration.py

import sonardyne_api as son
import time

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.TimeModuleConfiguration(
        input_time_sources=son.TimeSourceInputReferences(
            input_ntp_time_source=son.NtpTimeSourceReference(set_none=True),
            input_zda_time_source=son.ZdaTimeSourceReference(set_any=True),
            input_pps_time_source=son.PpsTimeSourceReference(set_none=True)
        )
    )))

import sonardyne_api as son
import time

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.TimeModuleConfiguration(
        input_time_sources=son.TimeSourceInputReferences(
            input_ntp_time_source=son.NtpTimeSourceReference(set_none=True),
            input_zda_time_source=son.ZdaTimeSourceReference(set_any=True),
            input_pps_time_source=son.PpsTimeSourceReference(set_none=True)
        )
    )))

import sonardyne_api as son
import time

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.TimeModuleConfiguration(
        input_time_sources=son.TimeSourceInputReferences(
            input_ntp_time_source=son.NtpTimeSourceReference(set_none=True),
            input_zda_time_source=son.ZdaTimeSourceReference(set_any=True),
            input_pps_time_source=son.PpsTimeSourceReference(set_none=True)
        )
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

Time Source

NTP Time Source Configuration

Example code: ntp_time_source_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.NtpTimeSourceConfiguration(ntp_server_ip_address="230.229.228.227")))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.NtpTimeSourceConfiguration(ntp_server_ip_address="230.229.228.227")))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.NtpTimeSourceConfiguration(ntp_server_ip_address="230.229.228.227")))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

PPS Time Source Configuration

Example code: pps_time_source_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.PpsTimeSourceConfiguration(input_trigger_port=son.TriggerPortReference(set_any=True))))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.PpsTimeSourceConfiguration(input_trigger_port=son.TriggerPortReference(set_any=True))))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.PpsTimeSourceConfiguration(input_trigger_port=son.TriggerPortReference(set_any=True))))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

ZDA Time Source Configuration

Example code: zda_time_source_configuration.py

import sonardyne_api as son

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.ZdaTimeSourceConfiguration(input_data_port=son.DataPortReference(set_any_ethernet_port=True))))

import sonardyne_api as son

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.ZdaTimeSourceConfiguration(input_data_port=son.DataPortReference(set_any_ethernet_port=True))))

import sonardyne_api as son

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.ZdaTimeSourceConfiguration(input_data_port=son.DataPortReference(set_any_ethernet_port=True))))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

ZDA Time Source Observation

Example code: zda_time_source_observation.py

import sonardyne_api as son
import time

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation_stream.send(son.ZdaTimeSourceObservation(zda_data=son.ZdaData(current_utc_time_seconds = time.time())))
            time.sleep(1)
    except KeyboardInterrupt:
        pass

import sonardyne_api as son
import time

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation_stream.send(son.ZdaTimeSourceObservation(zda_data=son.ZdaData(current_utc_time_seconds = time.time())))
            time.sleep(1)
    except KeyboardInterrupt:
        pass

import sonardyne_api as son
import time

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation_stream.send(son.ZdaTimeSourceObservation(zda_data=son.ZdaData(current_utc_time_seconds = time.time())))
            time.sleep(1)
    except KeyboardInterrupt:
        pass

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

XPOS Source

XPOS Source Configuration

Example code: xpos_source_configuration.py

import sonardyne_api as son
import time

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.XposSourceConfiguration(
        input_data_port=son.DataPortReference(
            set_uid=son.UniqueID(name="External Control")
        )
    )))

import sonardyne_api as son
import time

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.XposSourceConfiguration(
        input_data_port=son.DataPortReference(
            set_uid=son.UniqueID(name="External Control")
        )
    )))

import sonardyne_api as son
import time

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    print(wrapper.set_configuration(son.XposSourceConfiguration(
        input_data_port=son.DataPortReference(
            set_uid=son.UniqueID(name="External Control")
        )
    )))

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))

XPOS Source Observation

Example code: xpos_source_observation.py

import sonardyne_api as son
import time

with son.WrapperGrpc('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation = son.XposSourceObservation(xpos_data=son.XposData(
                time_of_validity=son.Timestamp(common_time_seconds=time.time()),
                latitude_radians=0.0,
                longitude_radians=0.0,
                depth_metres=0.0,
                depth_uncertainty_metres=1.0,
                horizontal_position_uncertainty_metres=0.5,
                source=son.XPOS_SOURCE_GNSS
            ))
            observation_stream.send(observation)
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="XposSourceObservation"))
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son
import time

with son.WrapperFramedTcp('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation = son.XposSourceObservation(xpos_data=son.XposData(
                time_of_validity=son.Timestamp(common_time_seconds=time.time()),
                latitude_radians=0.0,
                longitude_radians=0.0,
                depth_metres=0.0,
                depth_uncertainty_metres=1.0,
                horizontal_position_uncertainty_metres=0.5,
                source=son.XPOS_SOURCE_GNSS
            ))
            observation_stream.send(observation)
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="XposSourceObservation"))
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import sonardyne_api as son
import time

with son.WrapperJsonTcp('0.0.0.0', 8103) as wrapper:  
    # write to stream
    observation_stream = wrapper.open_observation_stream()
    print("\n(writing to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            observation = son.XposSourceObservation(xpos_data=son.XposData(
                time_of_validity=son.Timestamp(common_time_seconds=time.time()),
                latitude_radians=0.0,
                longitude_radians=0.0,
                depth_metres=0.0,
                depth_uncertainty_metres=1.0,
                horizontal_position_uncertainty_metres=0.5,
                source=son.XPOS_SOURCE_GNSS
            ))
            observation_stream.send(observation)
            time.sleep(1)
    except KeyboardInterrupt:
        pass

    # read from stream
    observation_stream = wrapper.open_observation_stream(
        son.ObservationSubscriptionRequest(observation_subscriptions=[
            son.ObservationSubscription(matching_criteria=son.MatchingCriteria(match_type_name_suffix="XposSourceObservation"))
        ])
    )
    print("\n(listening to observation stream) press ctrl-c to stop stream")
    try:
        while True:
            for message in observation_stream.recv():
                print(type(message), message)
                print("(listening to observation stream) press ctrl-c to stop stream")
    except KeyboardInterrupt:
        pass

import json, socket

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.connect(('0.0.0.0', 8103))
while True:
    if data := socket.recv(10240):
        print(data.decode())