diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..e5aa97a --- /dev/null +++ b/.env.example @@ -0,0 +1,18 @@ +# Copy this file to .env and fill in values for your host/container. +# .env is ignored by git. + +# Existing docker-compose volume vars +M3_VOL=/Volumes/M3 +AUVCTD_VOL=/Volumes/AUVCTD +LRAUV_VOL=/Volumes/LRAUV +CALIBRATION_VOL=/Volumes/DMO +WORK_VOL=/opt/docker_auv-python_vols/data +HOST_NAME=example.shore.mbari.org +GMT_LIBRARY_PATH=/usr/lib/x86_64-linux-gnu/ + +# SSDS API endpoint (set to your dev server as needed) +SSDS_API_BASE=https://mooring-ssds.shore.mbari.org/api + +# SSDS provenance auth stubs (API key) +SSDS_API_KEY= +SSDS_API_KEY_HEADER=X-API-Key diff --git a/.vscode/launch.json b/.vscode/launch.json index c3c725b..98a94ff 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -394,6 +394,7 @@ "request": "launch", "program": "${workspaceFolder}/src/data/process_lrauv.py", "console": "integratedTerminal", + "envFile": "${workspaceFolder}/.env", // Lots bad time values in brizo 20250914T080941 due to memory corruption on the vehicle //"args": ["-v", "1", "--log_file", "brizo/missionlogs/2025/20250909_20250915/20250914T080941/202509140809_202509150109.nc4"] //"args": ["-v", "2", "--log_file", "brizo/missionlogs/2025/20250909_20250915/20250914T080941/202509140809_202509150109.nc4", "--clobber"] @@ -459,8 +460,12 @@ //"args": ["-v", "1", "--log_file", "pontus/missionlogs/2025/20250107_20250123/20250111T233141/202501112331_202501131744.nc4", "--no_cleanup"] // Test sipper_odv() with a log_file that has an ESP Sample //"args": ["-v", "1", "--log_file", "daphne/missionlogs/2026/20260316_20260318/20260317T191958/202603171919_202603181628.nc4", "--no_cleanup"] - // Test sipper_odv() with a log_file that has an ESP Sample - "args": ["-v", "1", "--log_file", "ahi/missionlogs/2025/20250414_20250418/20250415T040019/202504150400_202504152346.nc4", "--no_cleanup"] + // Test ahi mission that has Backseat Planktivore data + //"args": ["-v", "1", "--log_file", "ahi/missionlogs/2025/20250414_20250418/20250415T040019/202504150400_202504152346.nc4", "--no_cleanup"] + // Test not writing _2column_cmocean.png file if there is no data to plot + //"args": ["-v", "1", "--log_file", "ahi/missionlogs/2025/20250128_20250131/20250131T051404/202501310514_202501310535.nc4", "--clobber"] + // Test --update_ssds_provenance with a short log_file + "args": ["-v", "1", "--log_file", "ahi/missionlogs/2025/20250128_20250131/20250131T051404/202501310514_202501310535.nc4", "--update_ssds_provenance"] }, ] diff --git a/README.md b/README.md index 9106816..b0a435a 100644 --- a/README.md +++ b/README.md @@ -103,28 +103,38 @@ First time use with Docker on a server using a service account: * cd /opt # There should be an `auv-python` directory here that is writable by docker_user * git clone git@github.com:mbari-org/auv-python.git * cd auv-python -* Create a .env file in `/opt/auv-python` with the following contents: - `M3_VOL=` - `AUVCTD_VOL=` - `LRAUV_VOL=` - `CALIBRATION_VOL=` - `WORK_VOL=/data` - `HOST_NAME=` - `GMT_LIBRARY_PATH=/usr/lib/x86_64-linux-gnu/` +* Create a .env file in `/opt/auv-python` using .env.example as a guide + After installation and when logging into the server again mission data can be processed thusly: -* Setting up environment and printing help message: - `sudo -u docker_user -i` - `cd /opt/auv-python` - `git pull` # To get new changes, e.g. mission added to src/data/dorado_info.py - `export DOCKER_USER_ID=$(id -u)` - `docker compose build` - `docker compose run --rm auvpython src/data/process_i2map.py --help` -* To actually process a mission and have the processed data copied to the archive use the `-v` and `--clobber` options, e.g.: - `docker compose run --rm auvpython src/data/process_dorado.py --mission 2025.139.04 -v --clobber --noinput` -* To process LRAUV data for a specific vehicle and time range: - `docker compose run --rm auvpython src/data/process_lrauv.py --auv_name tethys --start 20250401T000000 --end 20250502T000000 -v --noinput` -* To process a specific LRAUV log file: - `docker compose run --rm auvpython src/data/process_lrauv.py --log_file tethys/missionlogs/2012/20120908_20120920/20120917T025522/201209170255_201209171110.nc4 -v --noinput` +* Setting up environment and printing help message: +``` + sudo -u docker_user -i + cd /opt/auv-python + git pull + export DOCKER_USER_ID=$(id -u) + docker compose build + docker compose run --rm auvpython src/data/process_i2map.py --help +``` + +* To actually process a mission and have the processed data copied to the archive use the `-v` and `--clobber` options, e.g.: +``` + docker compose run --rm auvpython src/data/process_dorado.py --mission 2025.139.04 -v --clobber --noinput +``` + +* To process LRAUV data for a specific vehicle and time range: +``` + docker compose run --rm auvpython src/data/process_lrauv.py --auv_name tethys --start 20250401T000000 --end 20250502T000000 -v --clobber --noinput +``` + +* For missions/log_files to be processed sequentially add the `--num_cores 1` option: +``` + docker compose run --rm auvpython src/data/process_lrauv.py --start 20250101T000000 --end 20260101T000000 -v --clobber --noinput --num_cores 1 +``` + +* To process a specific LRAUV log file: +``` + docker compose run --rm auvpython src/data/process_lrauv.py --log_file tethys/missionlogs/2012/20120908_20120920/20120917T025522/201209170255_201209171110.nc4 -v --clobber --noinput +``` -- diff --git a/docker-compose.yml b/docker-compose.yml index 746348b..7b08125 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,3 +1,4 @@ +# Create .env from .env.example, then set values for your environment. # Env variables required in .env, e.g. for Mac: # M3_VOL=/Volumes/M3 # AUVCTD_VOL=/Volumes/AUVCTD diff --git a/pyproject.toml b/pyproject.toml index 04dd4ba..aec52fc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,7 @@ dependencies = [ "pygmt==0.16", "pyproj>=3.7.1", "pysolar>=0.13", + "requests>=2.31.0", "rolling>=0.5.0", "seawater>=3.3.5", "statsmodels>=0.14.4", diff --git a/src/data/common_args.py b/src/data/common_args.py index dcfd242..12cc853 100644 --- a/src/data/common_args.py +++ b/src/data/common_args.py @@ -87,6 +87,11 @@ def get_processing_parser(): action="store_true", help="Don't re-process existing output files", ) + parser.add_argument( + "--update_ssds_provenance", + action="store_true", + help="Submit/update provenance records in the SSDS_Metadata database", + ) return parser diff --git a/src/data/create_products.py b/src/data/create_products.py index 6887817..d1c5e4f 100755 --- a/src/data/create_products.py +++ b/src/data/create_products.py @@ -1638,6 +1638,15 @@ def plot_2column(self) -> str: # noqa: C901, PLR0912, PLR0915 self._open_ds() + # Early return if no plot variables present in dataset + # Use a quick pre-check with LRAUV or Dorado variables (excluding computed 'density') + plot_variables = self._get_plot_variables(None if self._is_lrauv() else "ctd1") + if not any(var in self.ds for var, _ in plot_variables if var != "density"): + self.logger.warning( + "No plot variables found in dataset, skipping plot_2column", + ) + return None + idist, iz, distnav = self._grid_dims() if idist.size == 0 or iz.size == 0 or distnav.size == 0: self.logger.warning("Skipping plot_2column due to missing gridding dimensions") diff --git a/src/data/process.py b/src/data/process.py index 572d61c..33e881e 100755 --- a/src/data/process.py +++ b/src/data/process.py @@ -75,6 +75,7 @@ class data are: download_process and calibrate, while for LRAUV class data from logs2netcdfs import BASE_PATH, MISSIONLOGS, MISSIONNETCDFS, AUV_NetCDF from lopcToNetCDF import LOPC_Processor, UnexpectedAreaOfCode from nc42netcdfs import BASE_LRAUV_PATH, BASE_LRAUV_WEB, Extract +from provenance import get_dods_url, submit_process_run from resample import ( AUVCTD_OPENDAP_BASE, FLASH_THRESHOLD, @@ -177,6 +178,7 @@ def __init__(self, auv_name, vehicle_dir, mount_dir, calibration_dir, config=Non "no_cleanup": False, "skip_download_process": False, "archive_only_products": False, + "update_ssds_provenance": False, "num_cores": None, # Filtering/processing params (only used in from_args, not common_config) "start_year": None, @@ -708,6 +710,47 @@ def create_products(self, mission: str = None, log_file: str = None) -> None: cp.sipper_odv() cp.logger.removeHandler(self.log_handler) + def _submit_provenance( # noqa: PLR0913 + self, + output_nc: str, + base_path: str, + input_files: list[str], + pr_start: str, + pr_end: str, + script_name: str = "src/data/process.py", + log_file: str | None = None, + ) -> None: + """Submit a provenance record — failures are logged, never raised. + + *output_nc* is a path relative to *base_path* (e.g. + ``ahi/missionlogs/.../file_1S.nc``), matching the ``--log_file`` + convention. *base_path* is used internally to locate the file and + build the OPeNDAP URL. + """ + try: + full_nc = str(Path(base_path, output_nc)) + if not Path(full_nc).exists(): + self.logger.debug("Output %s not found, skipping provenance", full_nc) + return + log_url = get_dods_url(log_file) if log_file else None + submit_process_run( + producer_name=( + f"auv-python - Execution of {Path(script_name).name}" + f" to produce {Path(output_nc)}" + ), + producer_description=self.commandline, + nc_file_path=full_nc, + input_uris=input_files, + pr_start=pr_start, + pr_end=pr_end, + script_name=script_name, + cmd_line_args=self.commandline, + log_file_url=log_url, + log=self.logger, + ) + except Exception: # noqa: BLE001 + self.logger.warning("Provenance submission failed", exc_info=True) + def email(self, mission: str) -> None: self.logger.info("Sending notification email for %s", mission) email = Emailer() @@ -771,6 +814,7 @@ def cleanup(self, mission: str = None, log_file: str = None) -> None: self.logger.error("Either mission or log_file must be provided for cleanup.") def process_mission(self, mission: str, src_dir: str = "") -> None: # noqa: C901, PLR0912, PLR0915 + _pr_start = datetime.now(tz=UTC).isoformat() netcdfs_dir = Path( self.config["base_path"], self.auv_name, @@ -854,6 +898,40 @@ def process_mission(self, mission: str, src_dir: str = "") -> None: # noqa: C90 self.align(mission) self.resample(mission) self.create_products(mission) + if self.config["update_ssds_provenance"]: + self._submit_provenance( + output_nc=str( + Path( + self.auv_name, + MISSIONNETCDFS, + mission, + f"{self.auv_name}_{mission}_{FREQ}.nc", + ) + ), + base_path=self.config["base_path"], + input_files=[ + str( + Path( + self.config["base_path"], + self.auv_name, + MISSIONLOGS, + mission, + ) + ) + ], + pr_start=_pr_start, + pr_end=datetime.now(tz=UTC).isoformat(), + script_name="src/data/process_dorado.py", + log_file=str( + Path( + self.config["base_path"], + self.auv_name, + MISSIONNETCDFS, + mission, + f"{self.auv_name}_{mission}_{LOG_NAME}", + ) + ), + ) # self.archive() is called in finally: blocks in process_missions() def process_mission_job(self, mission: str, src_dir: str = "") -> None: @@ -1041,6 +1119,7 @@ def combine(self, log_file: str) -> None: @log_file_processor def process_log_file(self, log_file: str) -> None: + _pr_start = datetime.now(tz=UTC).isoformat() netcdfs_dir = Path(BASE_LRAUV_PATH, Path(log_file).parent) Path(netcdfs_dir).mkdir(parents=True, exist_ok=True) self.log_handler = logging.FileHandler( @@ -1074,6 +1153,22 @@ def process_log_file(self, log_file: str) -> None: ) if resampled_file.exists(): self.create_products(log_file=log_file) + if self.config["update_ssds_provenance"]: + self._submit_provenance( + output_nc=str(Path(log_file).parent / f"{Path(log_file).stem}_{FREQ}.nc"), + base_path=str(BASE_LRAUV_PATH), + input_files=[get_dods_url(str(Path(BASE_LRAUV_PATH, log_file)))], + pr_start=_pr_start, + pr_end=datetime.now(tz=UTC).isoformat(), + script_name="src/data/process_lrauv.py", + log_file=str( + Path( + BASE_LRAUV_PATH, + Path(log_file).parent, + f"{Path(log_file).stem}_processing.log", + ) + ), + ) else: self.logger.warning( "Resampled file %s not found, skipping create_products step", @@ -1340,6 +1435,11 @@ def process_command_line(self): "and append to the netCDF file name" ), ) + parser.add_argument( + "--update_ssds_provenance", + action="store_true", + help="Submit/update provenance records in the SSDS_Metadata database", + ) parser.add_argument( "--num_cores", action="store", diff --git a/src/data/provenance.py b/src/data/provenance.py new file mode 100644 index 0000000..fd8df76 --- /dev/null +++ b/src/data/provenance.py @@ -0,0 +1,256 @@ +"""Record data processing provenance in the SSDS_Metadata database. + +Submits a single POST to ``/api/process-runs/``, which atomically resolves +or creates all related entities (Person, Software, DataContainers, Resources) +server-side. Can be called from ``process.py`` after a successful processing +run, or standalone via the CLI at the bottom of this file. + +See: https://github.com/mbari-org/auv-python/issues/144 +""" + +import argparse +import logging +import os +import sys +from datetime import UTC, datetime +from pathlib import Path +from socket import gethostname + +import git +import requests + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- +ENV_SSDS_API_BASE = "SSDS_API_BASE" +SSDS_API_BASE = os.environ.get(ENV_SSDS_API_BASE, "https://mooring-ssds.shore.mbari.org/api") +GIT_WEB_BASE = "https://github.com/mbari-org/auv-python/blob" +REQUEST_TIMEOUT = 30 + +# Environment variable names used for authentication stubs. +ENV_SSDS_API_KEY = "SSDS_API_KEY" # noqa: S105 +ENV_SSDS_API_KEY_HEADER = "SSDS_API_KEY_HEADER" # noqa: S105 + +# Maps local absolute path prefixes to OPeNDAP URL prefixes. +# Mirrors the Perl ``%web_lookup`` hash in ssds_util.pl. +# Computed from the project layout, matching BASE_LRAUV_PATH / BASE_PATH. +_PROJECT_DATA = Path(__file__).resolve().parent.parent.parent / "data" +PATH_TO_URL_MAP: dict[str, str] = { + str(_PROJECT_DATA / "auv_data"): "http://dods.mbari.org/opendap/data/auvctd", + str(_PROJECT_DATA / "lrauv_data"): "http://dods.mbari.org/opendap/data/lrauv", +} + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- +def _get_git_version() -> str: + """Return the current git commit hash, or ``'unknown'``.""" + try: + repo = git.Repo(Path(__file__).resolve().parent, search_parent_directories=True) + except Exception: # noqa: BLE001 + logger.debug("Could not determine git version", exc_info=True) + return "unknown" + else: + return repo.head.commit.hexsha + + +def build_authenticated_session( + api_key: str | None = None, + api_key_header: str | None = None, +) -> requests.Session: + """Return a requests.Session with auth headers from args or env. + + Resolution order: + 1) Explicit function args + 2) Environment variables + + Supported auth stubs: + - API key via configurable header (default: X-API-Key) + """ + session = requests.Session() + + resolved_api_key = api_key or os.environ.get(ENV_SSDS_API_KEY) + resolved_api_key_header = ( + api_key_header or os.environ.get(ENV_SSDS_API_KEY_HEADER) or "X-API-Key" + ) + + if resolved_api_key: + session.headers[resolved_api_key_header] = resolved_api_key + + return session + + +def get_dods_url(nc_file_path: str) -> str: + """Translate a local NetCDF path to its OPeNDAP URL. + + Walks ``PATH_TO_URL_MAP`` looking for a matching prefix in the + *resolved* path string. Returns the original path unchanged if no + match is found. + """ + resolved = str(Path(nc_file_path).resolve()) + for local_prefix, url_prefix in PATH_TO_URL_MAP.items(): + if resolved.startswith(local_prefix): + return resolved.replace(local_prefix, url_prefix, 1) + return resolved + + +def get_git_url(script_name: str, version: str) -> str: + """Return a GitHub web URL for *script_name* at *version*.""" + return f"{GIT_WEB_BASE}/{version}/{script_name}" + + +# --------------------------------------------------------------------------- +# Core function +# --------------------------------------------------------------------------- +def submit_process_run( # noqa: PLR0913 + nc_file_path: str, + input_uris: list[str], + *, + producer_name: str | None = None, + producer_description: str | None = None, + poc_email: str = "mccann@mbari.org", + pr_start: str | None = None, + pr_end: str | None = None, + software_name: str = "auv-python", + software_version: str | None = None, + script_name: str = "src/data/process.py", + cmd_line_args: str = "", + log_file_url: str | None = None, + additional_resources: list[dict] | None = None, + api_key: str | None = None, + api_key_header: str | None = None, + api_base: str = SSDS_API_BASE, + session: requests.Session | None = None, + log: logging.Logger | None = None, +) -> dict | None: + """Submit a ProcessRun to the SSDS_Metadata database via POST /api/process-runs/. + + The server resolves or creates all related entities (Person, Software, + DataContainers, Resources) from the supplied natural keys. + + Returns the created ProcessRun dict on success, or raises on HTTP error. + """ + log = log or logger + session = session or build_authenticated_session( + api_key=api_key, + api_key_header=api_key_header, + ) + resolved_api_key_header = ( + api_key_header or os.environ.get(ENV_SSDS_API_KEY_HEADER) or "X-API-Key" + ) + if resolved_api_key_header not in session.headers: + log.warning( + "No SSDS API key header detected on session; set %s " + "(or pass a pre-authenticated session). " + "If the server uses cookie/SSO auth, this warning can be ignored.", + ENV_SSDS_API_KEY, + ) + software_version = software_version or _get_git_version() + + resources: list[dict] = [ + { + "name": "processing_script", + "uristring": get_git_url(script_name, software_version), + "description": f"Processing script: {script_name}", + }, + ] + if cmd_line_args: + resources.append( + { + "name": "command_line", + "uristring": f"urn:cmdline:{gethostname()}", + "description": cmd_line_args, + } + ) + if log_file_url: + resources.append( + { + "name": "processing_log", + "uristring": log_file_url, + "description": "Processing log file", + } + ) + if additional_resources: + resources.extend(additional_resources) + + output_uri = get_dods_url(nc_file_path) + payload = { + "output_uri": output_uri, + "output_dodsurlstring": f"{output_uri}.html", + "producer_name": producer_name, + "producer_description": producer_description, + "input_uris": input_uris, + "input_dodsurlstrings": [f"{uri}.html" for uri in input_uris], + "software_name": software_name, + "software_version": software_version, + "software_uristring": f"https://github.com/mbari-org/auv-python/tree/{software_version}", + "person_email": poc_email, + "startdate": pr_start, + "enddate": pr_end, + "resources": resources, + } + + url = f"{api_base}/process-runs/" + resp = session.post(url, json=payload, timeout=REQUEST_TIMEOUT) + if not resp.ok: + err_txt = ( + f"{resp.status_code} {resp.reason} for url: {url}" + f"\n payload: {payload}\n response: {resp.text}" + ) + raise requests.HTTPError(err_txt, response=resp) + process_run = resp.json() + log.info("Provenance recorded: %s -> id=%s", url, process_run.get("id", "?")) + return process_run + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- +def process_command_line() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("nc_file", help="Path to the output NetCDF file") + parser.add_argument( + "--input", + dest="inputs", + action="append", + default=[], + help="Input URI (may be repeated)", + ) + parser.add_argument("--email", default="mccann@mbari.org", help="Point-of-contact email") + parser.add_argument("--script", default="src/data/process.py", help="Processing script path") + parser.add_argument("--cmd", default=" ".join(sys.argv), help="Command line string") + parser.add_argument("--log-url", default=None, help="URL to processing log") + parser.add_argument( + "--api-base", + default=SSDS_API_BASE, + help="SSDS REST API base URL", + ) + parser.add_argument("-v", "--verbose", action="count", default=0) + return parser.parse_args() + + +if __name__ == "__main__": + args = process_command_line() + logging.basicConfig( + level=logging.DEBUG if args.verbose else logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + ) + now = datetime.now(tz=UTC).isoformat() + result = submit_process_run( + nc_file_path=args.nc_file, + input_uris=args.inputs, + poc_email=args.email, + pr_start=now, + pr_end=now, + script_name=args.script, + cmd_line_args=args.cmd, + log_file_url=args.log_url, + api_base=args.api_base, + ) + sys.exit(0 if result else 1) diff --git a/src/data/test_process_i2map.py b/src/data/test_process_i2map.py index 88d0026..89eb695 100644 --- a/src/data/test_process_i2map.py +++ b/src/data/test_process_i2map.py @@ -30,7 +30,7 @@ def test_process_i2map(complete_i2map_processing): # but it will alert us if a code change unexpectedly changes the file size. # If code changes are expected to change the file size then we should # update the expected size here. - EXPECTED_SIZE_GITHUB = 63137 + EXPECTED_SIZE_GITHUB = 63130 EXPECTED_SIZE_ACT = 63106 EXPECTED_SIZE_LOCAL = 64650 if str(proc.args.base_path).startswith("/home/runner"):