BatchRead¶

class HSTB.kluster.xarray_conversion.BatchRead(filfolder=None, dest=None, address=None, client=None, minchunksize=40000000, max_chunks=20, filtype='zarr', skip_dask=False, dashboard=False, show_progress=True, parallel_write=True)¶

Bases: ZarrBackend

BatchRead - multibeam data converter using dask infrastructure and xarray data types Pass in multibeam files, call read(), and gain access to xarray Datasets for each data type

NOTE: CURRENTLY ONLY ZARR BASED PROCESSING OF kluster_variables.supported_sonar FILES IS SUPPORTED

BatchRead is stored internally using the following conventions:
X = + Forward, Y = + Starboard, Z = + Down
roll = + Port Up, pitch = + Bow Up, gyro = + Clockwise

>> from xarray_conversion import BatchRead
>> converted = BatchRead(r’C:/data_dir/0009_20170523_181119_FA2806.all’)
Started local cluster client…
<Client: ‘tcp://127.0.0.1:62327’ processes=4 threads=16, memory=34.27 GB>
>> converted.read()

Running Kongsberg .all converter
C:/data_dir/0009_20170523_181119_FA2806.all: Using 20 chunks of size 1962957
Operating on sector 0, s/n 40111, freq 265000
Rebalancing 108 total ping records across 1 blocks of size 1000
Operating on sector 0, s/n 40111, freq 275000
Rebalancing 108 total ping records across 1 blocks of size 1000
Operating on sector 1, s/n 40111, freq 285000
Rebalancing 108 total ping records across 1 blocks of size 1000
Operating on sector 1, s/n 40111, freq 290000
Rebalancing 108 total ping records across 1 blocks of size 1000
Operating on sector 2, s/n 40111, freq 270000
Rebalancing 108 total ping records across 1 blocks of size 1000
Operating on sector 2, s/n 40111, freq 280000
Rebalancing 108 total ping records across 1 blocks of size 1000
Rebalancing 5302 total attitude records across 1 blocks of size 20000
Distributed conversion complete: 5.3s****
Constructed offsets successfully
read successful

# examine the serial number/sector/frequency combinations

>> [cnv.sector_identifier for cnv in converted.raw_ping]

[‘40111_0_265000’,

‘40111_0_275000’,
‘40111_1_285000’,
‘40111_1_290000’,
‘40111_2_270000’,
‘40111_2_280000’]

# display the first ping dataset (serial number 40111, sector 0, frequency 265khz)

>> converted.raw_ping[0]

<xarray.Dataset>

Dimensions: (beam: 182, time: 108)

Coordinates:

* beam (beam) int32 0 1 2 3 4 5 6 … 176 177 178 179 180 181

* time (time) float64 1.496e+09 1.496e+09 … 1.496e+09

Data variables:

beampointingangle  (time, beam) float32 dask.array<chunksize=(108, 182), meta=np.ndarray>
counter            (time) uint16 dask.array<chunksize=(108,), meta=np.ndarray>
detectioninfo      (time, beam) int32 dask.array<chunksize=(108, 182), meta=np.ndarray>
mode               (time) <U2 dask.array<chunksize=(108,), meta=np.ndarray>
modetwo            (time) <U4 dask.array<chunksize=(108,), meta=np.ndarray>
ntx                (time) uint16 dask.array<chunksize=(108,), meta=np.ndarray>
qualityfactor      (time, beam) int32 dask.array<chunksize=(108, 182), meta=np.ndarray>
soundspeed         (time) float32 dask.array<chunksize=(108,), meta=np.ndarray>
tiltangle          (time) float32 dask.array<chunksize=(108,), meta=np.ndarray>
traveltime         (time, beam) float32 dask.array<chunksize=(108, 182), meta=np.ndarray>
yawpitchstab       (time) <U2 dask.array<chunksize=(108,), meta=np.ndarray>

Attributes:

_conversion_complete:            Tue Oct 20 15:53:34 2020
installsettings_1495563079:      {“waterline_vertical_location”: “-0.640”…
multibeam_files:                 {‘0009_20170523_181119_FA2806.all’: [149…
output_path:                     C:collabdasktestdata_dirEM2040_small…
profile_1495563079:              [[0.0, 1489.2000732421875], [0.32, 1489….
reference:                       {‘beampointingangle’: ‘receiver’, ‘tilta…
runtimesettings_1495563080:      {“Counter”: “61968”, “SystemSerial#”: “4…
secondary_system_serial_number:  [0]
sector_identifier:               40111_0_265000
survey_number:                   [‘01_Patchtest_2806’]
system_serial_number:            [40111]
units:                           {‘beampointingangle’: ‘degrees’, ‘tiltan…
xyzrph:                          {‘antenna_x’: {‘1495563079’: ‘0.000’}, ‘…

Attributes Summary

chunk_size

Return the chunk size of the dataset for (time, beam)

Methods Summary

`batch_read`([output_mode])	General converter for multibeam files leveraging xarray and dask.distributed See batch_read, same process but working on memory efficiency
`build_additional_line_metadata`([save_pths])	After conversion, we run this additional step to build the line specific values to store as metadata.
`build_offsets`([save_pths])	Form sorteddict for unique entries in installation parameters across all lines, retaining the xyzrph for each transducer/receiver.
`correct_for_counter_reset`()	Ping counter (at least with the Kongsberg systems) is a 16 bit unsigned integer that will just reset once it reaches 65536.
`get_nearest_install_parameters`(query_time)	Return the install parameters dict object that is nearest in time to query_time
`get_nearest_runtime_parameters`(query_time)	Return the runtime parameters dict object that is nearest in time to query_time
`initialize_log`()	Initialize the logger, which writes to logfile, that is made at the root folder housing the converted data
`is_dual_head`()	Use the xyzrph keys to determine if sonar is dual head.
`read`([build_offsets])	Run the batch_read method on all available lines, writes to datastore (netcdf/zarr depending on self.filtype), and loads the data back into the class as self.raw_ping, self.raw_att.
`read_from_netcdf_fils`(ping_pths, attitude_pths)	Read from the generated netCDF files constructed with read()
`read_from_zarr_fils`(ping_pth, attitude_pth, ...)	Read from the generated zarr datastores constructed with read()
`reload_attituderecords`([skip_dask])
`reload_pingrecords`([skip_dask])	After writing new data to the zarr data store, you need to refresh the xarray Dataset object so that it sees the changes.
`return_all_profiles`()	Return dict of attribute_name/data for each sv profile in the ping dataset
`return_nearest_soundspeed_profile`(time_idx)	Using the settings_xxxxx attribute in the xarray dataset and a given timestamp, return the waterline offset (relative to the tx) nearest in time to the timestamp.
`return_ping_counters_at_time`(tme)	Accepts times as float or a numpy array of times
`return_prefix_for_rp`()	Determine the correct prefix index based on the sonar reference point of this converted data.
`return_raw_navigation`([start_time, end_time])	Return just the navigation side of the first ping record.
`return_runtime_and_installation_settings_dicts`()	installation and runtime parameters are saved as string (json.dumps) as attributes in each raw_ping dataset.
`return_rx_xyzrph`(time_idx)	Using the constructed xyzrph attribute (see build_offsets) and a given timestamp, return the receiver offsets and angles nearest in time to the timestamp
`return_system_time_indexed_array`([subset_time])	Most of the processing involves matching static, timestamped offsets or angles to time series data.
`return_tpu_parameters`(timestamp)	Pull out the tpu parameters from the xyzrph installation parameters.
`return_tx_xyzrph`(time_idx)	Using the constructed xyzrph attribute (see build_offsets) and a given timestamp, return the transmitter offsets and angles nearest in time to the timestamp
`return_utm_zone_number`()	Get the minimum/maximum longitude values and return the utm zone number
`return_waterline`(time_idx)	Using the settings_xxxxx attribute in the xarray dataset and a given timestamp, return the waterline offset (relative to the tx) nearest in time to the timestamp.
`return_xyz_prefixes_for_systems`()	self.raw_ping contains Datasets broken up by system.
`return_xyzrph_sorted_timestamps`(ky)	Takes in key name and outputs a list of sorted timestamps that are valid for that key.

Attributes Documentation

chunk_size¶: Return the chunk size of the dataset for (time, beam)

Methods Documentation

batch_read(output_mode='zarr')¶

General converter for multibeam files leveraging xarray and dask.distributed See batch_read, same process but working on memory efficiency

Parameters: output_mode (str) – ‘zarr’ or ‘netcdf’, zarr is the only currently supported mode, alters the output datastore
Returns: nested dictionary for each type (ping, attitude, navigation) with path to written data and metadata
Return type: dict

build_additional_line_metadata(save_pths=None)¶

After conversion, we run this additional step to build the line specific values to store as metadata. The end result is a ‘multibeam_files’ attribute that stores [mintime, maxtime, start_latitude, start_longitude, end_latitude, end_longitude, azimuth, distance]

Parameters: save_pths (Optional[str]) – a list of paths to zarr datastores for writing the multibeam_files attribute to if provided

build_offsets(save_pths=None)¶

Form sorteddict for unique entries in installation parameters across all lines, retaining the xyzrph for each transducer/receiver. key values depend on type of sonar, see sonar_translator

Modifies the xyzrph attribute with timestamp dictionary of entries

Parameters: save_pths (Optional[str]) – a list of paths to zarr datastores for writing the xyzrph attribute to if provided

correct_for_counter_reset()¶

Ping counter (at least with the Kongsberg systems) is a 16 bit unsigned integer that will just reset once it reaches 65536. This zero crossing can happen multiple times in a kluster dataset, as it comprises multiple survey lines. We need to handle this by reading it as a larger datatype (int64) and add the int16 limit whenever it is reached in the counter record. This should transform a sawtooth record into a smooth, unique array of values.

fqpr_generation.reform_2d_vars_across_sectors_at_time, reform_1d_vars_across_sectors_at_time will use this method automatically. Having duplicate ping counters will mess up the logic we use to reform pings from these sector based datasets.

get_nearest_install_parameters(query_time)¶

Return the install parameters dict object that is nearest in time to query_time

Parameters: query_time (float) – time in UTC seconds that you need the nearest install parameters to
Returns: install parameters dict object that is nearest in time to the query time
Return type: dict

get_nearest_runtime_parameters(query_time)¶

Return the runtime parameters dict object that is nearest in time to query_time

Parameters: query_time (float) – time in UTC seconds that you need the nearest runtime parameters to
Returns: runtime parameters dict object that is nearest in time to the query time
Return type: dict

initialize_log()¶: Initialize the logger, which writes to logfile, that is made at the root folder housing the converted data

is_dual_head()¶

Use the xyzrph keys to determine if sonar is dual head. Port/Starboard identifiers will exist if dual. Kongsberg writes both heads to one file, only identifiable by serial number (each head will have a different serial number)

Returns: True if dual head, False if not
Return type: bool

read(build_offsets=True)¶

Run the batch_read method on all available lines, writes to datastore (netcdf/zarr depending on self.filtype), and loads the data back into the class as self.raw_ping, self.raw_att.

If data loads correctly, builds out the self.xyzrph attribute and translates the runtime parameters to a usable form.

Parameters: build_offsets (bool) – if this is set, also build the xyzrph attribute, which is mandatory for processing later in Kluster. Make it optional so that when processing chunks of files, we can just run it once at the end after read()

read_from_netcdf_fils(ping_pths, attitude_pths)¶

Read from the generated netCDF files constructed with read()

Currently some issues with open_mfdataset that I’ve not resolved. Using it with the dask distributed cluster active results in worker errors/hdf errors. Using it without the distributed cluster works fine. So annoying. I’m sticking to the zarr stuff for now, distributed parallel read/writes appear to work there after I built my own writer.

Parameters

ping_pths (list) – paths to the ping netcdf files
attitude_pths (list) – path to the attitude netcdf files

read_from_zarr_fils(ping_pth, attitude_pth, logfile_pth)¶

Read from the generated zarr datastores constructed with read()

Parameters

ping_pth (list) – list of paths to each ping zarr group (by system)
attitude_pth (str) – path to the attitude zarr group
logfile_pth (str) – path to the text log file used by logging

reload_attituderecords(skip_dask=False)¶

reload_pingrecords(skip_dask=False)¶

After writing new data to the zarr data store, you need to refresh the xarray Dataset object so that it sees the changes. We do that here by just re-running open_zarr.

Parameters: skip_dask (bool) – if True will skip the dask distributd client stuff when reloading

return_all_profiles()¶

Return dict of attribute_name/data for each sv profile in the ping dataset

attribute name is always ‘profile_timestamp’ format, ex: ‘profile_1503411780’

Returns

list – list of profile names
list – list of [depth values, sv values] for each profile
list – list of times in utc seconds for each profile
list – list of [latitude, longitude] for each profile

return_nearest_soundspeed_profile(time_idx)¶

Using the settings_xxxxx attribute in the xarray dataset and a given timestamp, return the waterline offset (relative to the tx) nearest in time to the timestamp.

Parameters: time_idx (Union[int, str, float]) – UTC timestamp (accepts int/str/float)
Returns: key = closest timestamp and value = waterline offset
Return type: dict

return_ping_counters_at_time(tme)¶

Accepts times as float or a numpy array of times

To rebuild the full ping at a specific time, you need to get the ping counter(s) at that time. EM2040c have multiple pings at a specific time, so this will return a list of counters that is usually only one element long.

Parameters: tme (Union[float, array]) – float or numpy array, time to find ping counters for
Returns: list of ints for ping counter numbers at that time
Return type: cntrs

return_prefix_for_rp()¶: Determine the correct prefix index based on the sonar reference point of this converted data. For instance, if the sonar reference point is [‘tx_x’, ‘tx_y’, ‘rx_z’], the returned prefix indices would be [0,0,1], which will allow you to pull the correct lever arms from the xyzrph indices. See return_system_time_indexed_array.

return_raw_navigation(start_time=None, end_time=None)¶

Return just the navigation side of the first ping record. If a start time and end time are provided, will subset to just those times.

If this is a dual head sonar, it only returns the nav for the first head!

Parameters

start_time (Optional[float]) – if provided will allow you to only return navigation after this time. Selects the nearest time value to the one provided.
end_time (Optional[float]) – if provided will allow you to only return navigation before this time. Selects the nearest time value to the one provided.

Returns

latitude/longitude/altitude pulled from the raw navigation part of the ping record

Return type

xr.Dataset

return_runtime_and_installation_settings_dicts()¶: installation and runtime parameters are saved as string (json.dumps) as attributes in each raw_ping dataset. Use this method to return the dicts that encompass each installation and runtime entry.

return_rx_xyzrph(time_idx)¶

Using the constructed xyzrph attribute (see build_offsets) and a given timestamp, return the receiver offsets and angles nearest in time to the timestamp

Parameters: time_idx (Union[int, str, float]) – UTC timestamp (accepts int/str/float)
Returns: key = closest timestamp and values = mounting angle/offsets for receiver
Return type: dict

return_system_time_indexed_array(subset_time=None)¶

Most of the processing involves matching static, timestamped offsets or angles to time series data. Given that we might have a list of systems for dual head sonar and a list of timestamped offsets, need to iterate through all of this in each processing loop. Systems/timestamps length should be minimal, so we just loop in python.

Parameters: subset_time (Optional[list]) – List of unix timestamps in seconds, used as ranges for times that you want to process
Returns: list of indices for each system/timestamped offsets that are within the provided subset. length of the list is the number of heads for this sonar.
Return type: list

return_tpu_parameters(timestamp)¶

Pull out the tpu parameters from the xyzrph installation parameters. We need these parameters to compute tpu. Only pulls the values for a single timestamped entry, using the provided timestamp.

Parameters: timestamp (str) – utc time in seconds for the entry
Returns: dict of tpu parameters for the timestamped entry
Return type: dict

return_tx_xyzrph(time_idx)¶

Using the constructed xyzrph attribute (see build_offsets) and a given timestamp, return the transmitter offsets and angles nearest in time to the timestamp

Parameters: time_idx (Union[int, str, float]) – UTC timestamp (accepts int/str/float)
Returns: key = closest timestamp and values = mounting angle/offsets for receiver
Return type: dict

return_utm_zone_number()¶

Get the minimum/maximum longitude values and return the utm zone number

Returns: zone number, e.g. ‘19N’ for UTM Zone 19 N
Return type: str

return_waterline(time_idx)¶

Using the settings_xxxxx attribute in the xarray dataset and a given timestamp, return the waterline offset (relative to the tx) nearest in time to the timestamp.

Parameters: time_idx (Union[int, str, float]) – UTC timestamp (accepts int/str/float)
Returns: key = closest timestamp and value = waterline offset
Return type: dict

return_xyz_prefixes_for_systems()¶

self.raw_ping contains Datasets broken up by system. This method will return the prefixes you need to get the offsets/angles from self.xyzrph depending on dual-head

Returns: list of two element lists containing the prefixes needed for tx/rx offsets and angles
Return type: List

return_xyzrph_sorted_timestamps(ky)¶

Takes in key name and outputs a list of sorted timestamps that are valid for that key.

Parameters: ky (str) – key name that you want the timestamps from (i.e. ‘tx_x’)
Returns: sorted timestamps of type str, in increasing order
Return type: list

BatchRead¶

kluster

Navigation

Related Topics