updated azure processor

oceanstream/cli/main.py

@@ -39,6 +39,7 @@
  'distributed.comm.retry.count': 0
 })
 
+
 def initialize(settings, file_path, log_level=None):
  config_data = load_config(settings["config"])
  config_data["raw_path"] = file_path
@@ -103,6 +104,8 @@ def convert(
  output: str = typer.Option(None,
  help="Destination path for saving Zarr converted data. Defaults to a predefined "
  "directory if not specified."),
+ sonar_model: str = typer.Option(None, help="Sonar model used to collect the data",
+ show_choices=["AZFP", "EK60", "ES70", "EK80", "ES80", "EA640", "AD2CP"]),
  workers_count: int = typer.Option(os.cpu_count(), help="Number of CPU workers to use for processing"),
  config: str = typer.Option(None, help="Path to a configuration file"),
  log_level: str = typer.Option("WARNING", help="Set the logging level",
@@ -113,6 +116,7 @@ def convert(
  """
  settings = {
  "config": config,
+ "sonar_model": sonar_model,
  "output_folder": output or DEFAULT_OUTPUT_FOLDER
  }
 
@@ -122,6 +126,7 @@ def convert(
  try:
  if filePath.is_file():
  from oceanstream.process import convert_raw_file
+ print(f"[blue]Converting raw file {source} to Zarr...[/blue]")
  convert_raw_file(filePath, configData)
  print(
  f"[blue]✅ Converted raw file {source} to Zarr and wrote output to: {configData['output_folder']} [/blue]")

oceanstream/exports/csv/csv_export_from_Sv.py

@@ -99,6 +99,7 @@ def create_location(data: xr.Dataset, epsilon=0.00001, min_distance=0.01) -> pd.
 
  # Apply Ramer-Douglas-Peucker algorithm for thinning coordinates
  points = df[["lat", "lon"]].values
+
  thinned_points = ramer_douglas_peucker(points, epsilon)
 
  # Create a thinned DataFrame

oceanstream/process/azure/blob_storage.py

@@ -1,8 +1,18 @@
+import os
+
 import echopype as ep
+from adlfs import AzureBlobFileSystem
 
 
-def list_zarr_files(azfs, path):
+def list_zarr_files(path, azfs=None):
  """List all Zarr files in the Azure Blob Storage container along with their metadata."""
+
+ if azfs is None:
+ azfs = get_azfs()
+
+ if azfs is None:
+ raise ValueError("Azure Blob Storage connection string not found and no azfs instance was specified.")
+
  zarr_files = []
  for blob in azfs.ls(path, detail=True):
  if blob['type'] == 'directory' and not blob['name'].endswith('.zarr'):
@@ -18,6 +28,18 @@ def list_zarr_files(azfs, path):
  return zarr_files
 
 
+def get_azfs():
+ """Get the Azure Blob Storage filesystem object using the connection string from environment variables."""
+ connection_string = os.getenv('AZURE_STORAGE_CONNECTION_STRING')
+
+ if not connection_string:
+ return None
+
+ azfs = AzureBlobFileSystem(connection_string=connection_string)
+
+ return azfs
+
+
 def open_zarr_store(azfs, store_name, chunks=None):
  """Open a Zarr store from Azure Blob Storage."""
  mapper = azfs.get_mapper(store_name)

oceanstream/process/file_processor.py

@@ -229,21 +229,34 @@ def convert_raw_file(file_path, config_data, base_path=None, progress_queue=None
  relative_path = file_path_obj.relative_to(base_path)
  relative_path = relative_path.parent
  else:
- relative_path = file_path_obj.name
+ relative_path = None
 
  echodata, encode_mode = read_file(file_config_data, use_swap=True, skip_integrity_check=True)
+ file_name = file_path_obj.stem + ".zarr"
 
  if 'cloud_storage' in config_data:
- file_name = file_path_obj.stem + ".zarr"
- store = _get_chunk_store(config_data['cloud_storage'], Path(relative_path) / file_name)
+ if relative_path:
+ file_location = Path(relative_path) / file_name
+ else:
+ file_location = file_name
+ store = _get_chunk_store(config_data['cloud_storage'], file_location)
  echodata.to_zarr(save_path=store, overwrite=True, parallel=False)
+ output_zarr_path = store
  else:
- output_path = Path(config_data["output_folder"]) / relative_path
- output_path.mkdir(parents=True, exist_ok=True)
- echodata.to_zarr(save_path=output_path, overwrite=True, parallel=False)
+ if relative_path:
+ output_path = Path(config_data["output_folder"]) / relative_path
+ output_path.mkdir(parents=True, exist_ok=True)
+ else:
+ output_path = Path(config_data["output_folder"])
+
+ output_zarr_path = output_path / file_name
+ echodata.to_zarr(save_path=output_zarr_path, overwrite=True, parallel=False)
 
  if progress_queue:
  progress_queue.put(file_path)
+
+ return output_zarr_path
+
  except Exception as e:
  logging.error("Error processing file %s", file_path)
  print(Traceback())
@@ -264,7 +277,6 @@ def _get_chunk_store(storage_config, path):
  azfs = AzureBlobFileSystem(**storage_config['storage_options'])
 
  return azfs.get_mapper(f"{storage_config['container_name']}/{path}")
-
  else:
  raise ValueError(f"Unsupported storage type: {storage_config['storage_type']}")
 

requirements.txt

@@ -4,4 +4,7 @@ numpy
 pydantic
 rich
 tqdm
+matplotlib
+haversine
+IPython
 git+https://github.com/OceanStreamIO/echopype-dev.git@oceanstream#egg=echopype