diff --git a/pyproject.toml b/pyproject.toml index 3d8e3c4..05518ef 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,6 +27,7 @@ dependencies = { file = ["requirements.txt"] } [project.scripts] lung_utils = "lung_utils.main:main" +hamilton-waveform-export = "lung_utils.hamilton_ventilator.waveform_exporter:main" ## Tools diff --git a/src/lung_utils/hamilton_ventilator/waveform_exporter.py b/src/lung_utils/hamilton_ventilator/waveform_exporter.py new file mode 100644 index 0000000..7289ee3 --- /dev/null +++ b/src/lung_utils/hamilton_ventilator/waveform_exporter.py @@ -0,0 +1,431 @@ +"""Export Hamilton waveform data to downstream formats.""" + +import argparse +import re +import shutil +import subprocess + +import numpy as np +import pandas as pd +from lung_utils.hamilton_ventilator.waveform_plotter import load_waveform_txt + +EXCLUDED_WAVEFORM_COLUMNS = { + "Date_Time", + "Time (s)", + "Absolute_Time", + "Breath Number", + "Status", +} + + +def get_waveform_fields(df: pd.DataFrame) -> list[str]: + """Return data columns that can be exported as waveforms.""" + return [ + column + for column in df.columns + if column not in EXCLUDED_WAVEFORM_COLUMNS + ] + + +def make_variable_name(field_name: str) -> str: + """Create a valid simple variable name from a waveform field name.""" + variable_name = re.sub(r"\W+", "_", field_name.strip().lower()).strip("_") + if not variable_name: + return "waveform" + if variable_name[0].isdigit(): + return f"waveform_{variable_name}" + return variable_name + + +def extract_waveforms( + df: pd.DataFrame, + fields: list[str], + start: float, + end: float, + sampling_rate: float | None = None, + preserve_time: bool = False, +) -> dict[str, np.ndarray]: + """Extract one or more waveform fields into a common time basis.""" + if not fields: + raise ValueError("At least one waveform field must be selected.") + missing_fields = [field for field in fields if field not in df.columns] + if missing_fields: + raise ValueError( + "Waveform field(s) not found: " + ", ".join(missing_fields) + ) + if start < 0: + raise ValueError("Start time must be non-negative.") + if end <= start: + raise ValueError("End time must be larger than start time.") + if sampling_rate is not None and sampling_rate <= 0: + raise ValueError("Sampling rate must be positive.") + + time = pd.to_numeric(df["Time (s)"], errors="coerce") + values_by_field = { + field: pd.to_numeric(df[field], errors="coerce") for field in fields + } + valid = time.notna() + for values in values_by_field.values(): + valid &= values.notna() + + time_values = time[valid].to_numpy(dtype=float) + if len(time_values) < 2: + raise ValueError("At least two valid waveform samples are required.") + + waveform_values_by_field = { + field: values[valid].to_numpy(dtype=float) + for field, values in values_by_field.items() + } + + tolerance = 1.0e-6 + if abs(start - time_values[0]) <= tolerance: + start = time_values[0] + if abs(end - time_values[-1]) <= tolerance: + end = time_values[-1] + + if start < time_values[0] or end > time_values[-1]: + raise ValueError( + "Requested interval must be within the available waveform time " + f"range [{time_values[0]:.12g}, {time_values[-1]:.12g}]." + ) + + if sampling_rate is None: + in_interval = (time_values >= start) & (time_values <= end) + sample_times = time_values[in_interval] + if len(sample_times) == 0: + raise ValueError( + "No original waveform samples found in the requested interval." + ) + output = { + field: values[in_interval] + for field, values in waveform_values_by_field.items() + } + else: + step = 1.0 / sampling_rate + sample_times = np.arange(start, end + step * 0.5, step) + sample_times = sample_times[sample_times <= end] + if sample_times[-1] < end: + sample_times = np.append(sample_times, end) + output = { + field: np.interp(sample_times, time_values, values) + for field, values in waveform_values_by_field.items() + } + + output_time = sample_times if preserve_time else sample_times - start + return {"time": output_time, **output} + + +def save_waveforms_npy(waveforms: dict[str, np.ndarray], output: str) -> None: + """Save extracted waveform arrays to a numpy file.""" + np.save(output, waveforms, allow_pickle=True) + + +def format_fourc_linearinterpolation( + times: np.ndarray, + values: np.ndarray, + funct_number: int, + variable_name: str, +) -> str: + """Format sampled points as a 4C linearinterpolation function block.""" + if len(times) != len(values): + raise ValueError("Times and values must have the same length.") + if len(times) == 0: + raise ValueError("At least one interpolation point is required.") + if funct_number <= 0: + raise ValueError("Function number must be positive.") + + lines = [ + f"FUNCT{funct_number}:", + f"- SYMBOLIC_FUNCTION_OF_TIME: {variable_name}", + "- VARIABLE: 0", + f" NAME: {variable_name}", + " TYPE: linearinterpolation", + f" NUMPOINTS: {len(times)}", + " TIMES:", + ] + lines.extend(f" - {_format_number(time)}" for time in times) + lines.append(" VALUES:") + lines.extend(f" - {_format_number(value)}" for value in values) + return "\n".join(lines) + + +def copy_to_clipboard(text: str) -> None: + """Copy text to the system clipboard without blocking the CLI.""" + if shutil.which("wl-copy"): + subprocess.run( + ["wl-copy"], + input=text, + text=True, + check=True, + timeout=5, + ) + return + + if shutil.which("xclip"): + process = subprocess.Popen( + ["xclip", "-selection", "clipboard"], + stdin=subprocess.PIPE, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + text=True, + start_new_session=True, + ) + if process.stdin is not None: + process.stdin.write(text) + process.stdin.close() + return + + if shutil.which("xsel"): + process = subprocess.Popen( + ["xsel", "--clipboard", "--input"], + stdin=subprocess.PIPE, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + text=True, + start_new_session=True, + ) + if process.stdin is not None: + process.stdin.write(text) + process.stdin.close() + return + + raise RuntimeError( + "No supported clipboard backend found. Install " + "wl-copy, xclip, or xsel." + ) + + +def choose_waveform_fields(fields: list[str]) -> list[str]: + """Ask the user to choose waveform fields interactively.""" + if not fields: + raise ValueError("No waveform fields found in input file.") + + print("Available waveform fields:") + for index, field in enumerate(fields, start=1): + print(f" {index}: {field}") + + while True: + selected = input( + "Choose waveform field numbers separated by commas, or 'all': " + ).strip() + if selected.lower() == "all": + return fields + + try: + selected_indices = [ + int(part.strip()) + for part in selected.split(",") + if part.strip() + ] + except ValueError: + print("Please enter valid numbers or 'all'.") + continue + + if selected_indices and all( + 1 <= selected_index <= len(fields) + for selected_index in selected_indices + ): + return [ + fields[selected_index - 1] + for selected_index in selected_indices + ] + print(f"Please enter numbers between 1 and {len(fields)}.") + + +def prompt_float(label: str, default: float | None = None) -> float: + """Ask the user to enter a floating point value.""" + default_text = ( + f" [{_format_number(default)}]" if default is not None else "" + ) + while True: + entered = input(f"{label}{default_text}: ").strip() + if not entered and default is not None: + return default + try: + return float(entered) + except ValueError: + print("Please enter a valid number.") + + +def build_parser() -> argparse.ArgumentParser: + """Create the command line parser.""" + parser = argparse.ArgumentParser( + description="Export Hamilton waveform.txt fields to npy or 4C formats." + ) + parser.add_argument("waveform_file", help="Path to Hamilton waveform.txt") + parser.add_argument( + "--list-fields", + action="store_true", + help="List waveform fields in the input file and exit.", + ) + parser.add_argument( + "--fields", + nargs="+", + help="Waveform fields to export. Quote names that contain spaces.", + ) + parser.add_argument( + "--all-fields", + action="store_true", + help="Export all available waveform fields.", + ) + parser.add_argument( + "--start", + type=float, + help="Start time in seconds relative to the waveform recording start.", + ) + parser.add_argument( + "--end", + type=float, + help="End time in seconds relative to the waveform recording start.", + ) + parser.add_argument( + "--sampling-rate", + type=float, + help=( + "Sampling rate in Hz for resampling. If omitted, exact original " + "waveform samples are used." + ), + ) + parser.add_argument( + "--format", + choices=("npy", "fourc"), + default="npy", + help="Output format. Defaults to npy.", + ) + parser.add_argument( + "--output", + help="Output file path. Required for --format npy.", + ) + parser.add_argument( + "--funct", + type=int, + default=1, + help="4C function number for --format fourc.", + ) + parser.add_argument( + "--variable-name", + help="Variable name used inside the 4C function block.", + ) + parser.add_argument( + "--preserve-time", + action="store_true", + help="Keep original recording times instead of resetting to t=0.", + ) + parser.add_argument( + "--print", + action="store_true", + help="Print generated output summary or 4C function string to stdout.", + ) + return parser + + +def main() -> None: + """Run the Hamilton waveform export CLI.""" + parser = build_parser() + args = parser.parse_args() + + df = load_waveform_txt(args.waveform_file) + available_fields = get_waveform_fields(df) + + if args.list_fields: + for field in available_fields: + print(field) + return + + selected_fields = _get_selected_fields(parser, args, available_fields) + if args.format == "fourc" and len(selected_fields) != 1: + parser.error( + "--format fourc requires exactly one selected waveform field." + ) + if args.format == "npy" and not args.output: + parser.error("--output is required for --format npy.") + + available_start = float(df["Time (s)"].min()) + available_end = float(df["Time (s)"].max()) + print( + "Available time range: " + f"{_format_number(available_start)} s to " + f"{_format_number(available_end)} s" + ) + + start = args.start + if start is None: + start = prompt_float("Start time in seconds", default=available_start) + + end = args.end + if end is None: + end = prompt_float("End time in seconds", default=available_end) + + if args.sampling_rate is None: + print( + "Using original waveform samples. Pass --sampling-rate to " + "resample." + ) + + waveforms = extract_waveforms( + df, + fields=selected_fields, + start=start, + end=end, + sampling_rate=args.sampling_rate, + preserve_time=args.preserve_time, + ) + + if args.format == "npy": + save_waveforms_npy(waveforms, args.output) + print( + f"Saved {len(selected_fields)} waveform field(s) with " + f"{len(waveforms['time'])} samples to {args.output}." + ) + return + + field = selected_fields[0] + variable_name = args.variable_name or make_variable_name(field) + function_string = format_fourc_linearinterpolation( + times=waveforms["time"], + values=waveforms[field], + funct_number=args.funct, + variable_name=variable_name, + ) + copy_to_clipboard(function_string) + print( + "Copied 4C linearinterpolation function " + f"for '{field}' with {len(waveforms['time'])} points to clipboard." + ) + if args.print: + print(function_string) + + +def _get_selected_fields( + parser: argparse.ArgumentParser, + args: argparse.Namespace, + available_fields: list[str], +) -> list[str]: + if args.fields and args.all_fields: + parser.error("Use either --fields or --all-fields, not both.") + + if args.all_fields: + selected_fields = available_fields + elif args.fields: + selected_fields = args.fields + else: + selected_fields = choose_waveform_fields(available_fields) + + unknown_fields = [ + field for field in selected_fields if field not in available_fields + ] + if unknown_fields: + parser.error( + "Unknown waveform field(s): " + f"{', '.join(unknown_fields)}. Available fields: " + f"{', '.join(available_fields)}" + ) + return selected_fields + + +def _format_number(value: float) -> str: + return f"{value:.12g}" + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/src/lung_utils/hamilton_ventilator/waveform_plotter.py b/src/lung_utils/hamilton_ventilator/waveform_plotter.py index 35144e9..370a788 100644 --- a/src/lung_utils/hamilton_ventilator/waveform_plotter.py +++ b/src/lung_utils/hamilton_ventilator/waveform_plotter.py @@ -1,4 +1,5 @@ -import sys +import argparse +from pathlib import Path import dash import pandas as pd @@ -6,6 +7,20 @@ from dash import Input, Output, dcc, html from plotly.subplots import make_subplots +METADATA_COLUMNS = { + "Date_Time", + "Time (s)", + "Absolute_Time", + "Breath Number", + "Status", +} + +PARAMETER_DEFAULT_FIELDS = [ + "Mode Name", + "PEEP/ CPAP /cmH2O", + "Tidal Volume /ml", +] + # ==== Load waveform file ==== def load_waveform_txt(filepath): @@ -35,20 +50,131 @@ def load_waveform_txt(filepath): return df +def find_parameter_file(waveform_filepath): + """Find a single Hamilton parameter file next to a waveform file.""" + waveform_path = Path(waveform_filepath) + parameter_files = sorted(waveform_path.parent.glob("P_Hamilton-C6*")) + if not parameter_files: + parameter_files = sorted(waveform_path.parent.glob("P_Hamilton*")) + + if not parameter_files: + raise FileNotFoundError( + "No Hamilton parameter file matching P_Hamilton-C6* was found in " + f"{waveform_path.parent}." + ) + if len(parameter_files) > 1: + matches = "\n".join(str(path) for path in parameter_files) + raise ValueError( + "Multiple Hamilton parameter files were found. Please pass one " + f"with --parameter-file:\n{matches}" + ) + return str(parameter_files[0]) + + +def _load_hamilton_txt(filepath, start_time_val=None): + try: + df = pd.read_csv( + filepath, sep="\t", engine="python", encoding="latin1" + ) + except UnicodeDecodeError: + raise ValueError( + f"Failed to decode {filepath}. Please check the file encoding." + ) + + df.columns = [col.strip() for col in df.columns] + df = df.dropna(how="all") + df["Date_Time"] = pd.to_numeric(df["Date_Time"], errors="coerce") + df = df.dropna(subset=["Date_Time"]) + + if start_time_val is None: + start_time_val = df["Date_Time"].iloc[0] + df["Time (s)"] = (df["Date_Time"] - start_time_val) * 24 * 3600 + df["Absolute_Time"] = pd.to_datetime( + df["Date_Time"], unit="D", origin="1899-12-30" + ) + + return df + + +def load_parameter_txt(filepath, start_time_val=None): + """Load a Hamilton parameter file and align time to waveform start.""" + return _load_hamilton_txt(filepath, start_time_val=start_time_val) + + +def get_plot_columns(df): + """Return non-metadata columns suitable for plotting.""" + return [col for col in df.columns if col not in METADATA_COLUMNS] + + +def get_default_parameter_fields(parameter_columns): + defaults = [ + field + for field in PARAMETER_DEFAULT_FIELDS + if field in parameter_columns + ] + return defaults or parameter_columns[:1] + + +def add_trace_to_subplot(fig, df, field, row, *, is_parameter=False): + y_data = pd.to_numeric(df[field], errors="coerce") + original_values = None + + if is_parameter and y_data.notna().sum() == 0: + category_values = df[field].fillna("--").astype(str) + codes, categories = pd.factorize(category_values, sort=True) + y_data = pd.Series(codes, index=df.index) + original_values = category_values + fig.update_yaxes( + tickmode="array", + tickvals=list(range(len(categories))), + ticktext=list(categories), + row=row, + col=1, + ) + + abs_time_str = df["Absolute_Time"].dt.strftime("%H:%M:%S.%f").str[:-3] + customdata = ( + original_values if original_values is not None else abs_time_str + ) + hovertemplate = ( + "%{y}
" + "Time: %{x:.2f} s
" + "Abs Time: %{customdata}" + "" + ) + if original_values is not None: + hovertemplate = ( + "%{customdata}
" "Time: %{x:.2f} s" "" + ) + + trace = go.Scatter( + x=df["Time (s)"], + y=y_data, + mode="lines+markers" if is_parameter else "lines", + name=field, + customdata=customdata, + hovertemplate=hovertemplate, + ) + fig.add_trace(trace, row=row, col=1) + fig.update_yaxes(title_text=field, row=row, col=1) + + # ==== Create Dash App ==== -def create_dash_app(df, file_path): +def create_dash_app( + df, file_path, parameter_df=None, parameter_file_path=None +): app = dash.Dash(__name__) app.title = "Hamilton Waveform Viewer" # Select waveform columns - exclude_cols = [ - "Date_Time", - "Time (s)", - "Absolute_Time", - "Breath Number", - "Status", - ] - waveform_columns = [col for col in df.columns if col not in exclude_cols] + waveform_columns = get_plot_columns(df) + parameter_columns = [] + parameter_default_fields = [] + if parameter_df is not None: + parameter_columns = get_plot_columns(parameter_df) + parameter_default_fields = get_default_parameter_fields( + parameter_columns + ) # Get absolute start time start_time_str = "" @@ -57,78 +183,90 @@ def create_dash_app(df, file_path): df["Absolute_Time"].iloc[0].strftime("%Y-%m-%d %H:%M:%S") ) - app.layout = html.Div( - [ - html.H2("Hamilton Ventilator Waveform Viewer"), - html.Div( - f"Loaded file: {file_path}", - id="file-info", - style={"marginBottom": "5px"}, - ), - html.Div( - f"Recording Start Time: {start_time_str}", - id="start-time-info", - style={"marginBottom": "10px", "fontWeight": "bold"}, - ), - html.Label("Select up to 3 waveforms:"), - dcc.Dropdown( - id="waveform-dropdown", - options=[ - {"label": col, "value": col} for col in waveform_columns - ], - value=[waveform_columns[0]] if waveform_columns else [], - multi=True, - ), - dcc.Graph(id="waveform-plot"), - ] - ) + layout_children = [ + html.H2("Hamilton Ventilator Waveform Viewer"), + html.Div( + f"Loaded file: {file_path}", + id="file-info", + style={"marginBottom": "5px"}, + ), + html.Div( + f"Recording Start Time: {start_time_str}", + id="start-time-info", + style={"marginBottom": "10px", "fontWeight": "bold"}, + ), + html.Label("Select up to 3 waveforms:"), + dcc.Dropdown( + id="waveform-dropdown", + options=[{"label": col, "value": col} for col in waveform_columns], + value=[waveform_columns[0]] if waveform_columns else [], + multi=True, + ), + ] - @app.callback( - Output("waveform-plot", "figure"), Input("waveform-dropdown", "value") - ) - def update_graph(selected_waveforms): - if not selected_waveforms or df.empty: + callback_inputs = [Input("waveform-dropdown", "value")] + if parameter_df is not None: + layout_children.extend( + [ + html.Div( + f"Loaded parameter file: {parameter_file_path}", + id="parameter-file-info", + style={"marginTop": "10px", "marginBottom": "5px"}, + ), + html.Label("Select up to 3 parameter fields:"), + dcc.Dropdown( + id="parameter-dropdown", + options=[ + {"label": col, "value": col} + for col in parameter_columns + ], + value=parameter_default_fields, + multi=True, + ), + ] + ) + callback_inputs.append(Input("parameter-dropdown", "value")) + + layout_children.append(dcc.Graph(id="waveform-plot")) + app.layout = html.Div(layout_children) + + @app.callback(Output("waveform-plot", "figure"), *callback_inputs) + def update_graph(selected_waveforms, selected_parameters=None): + if df.empty: return go.Figure() # Enforce maximum of 3 plots if isinstance(selected_waveforms, str): selected_waveforms = [selected_waveforms] + if isinstance(selected_parameters, str): + selected_parameters = [selected_parameters] - selected_waveforms = selected_waveforms[:3] - num_plots = len(selected_waveforms) + selected_waveforms = (selected_waveforms or [])[:3] + selected_parameters = (selected_parameters or [])[:3] + num_waveform_plots = len(selected_waveforms) + num_parameter_plots = len(selected_parameters) + num_plots = num_waveform_plots + num_parameter_plots + if num_plots == 0: + return go.Figure() fig = make_subplots( rows=num_plots, cols=1, shared_xaxes=True, vertical_spacing=0.05, - subplot_titles=selected_waveforms, + subplot_titles=selected_waveforms + selected_parameters, ) for i, waveform in enumerate(selected_waveforms, start=1): - y_data = pd.to_numeric(df[waveform], errors="coerce") + add_trace_to_subplot(fig, df, waveform, i) - # Format absolute time for hover (hours:minutes:seconds.ms) - abs_time_str = ( - df["Absolute_Time"].dt.strftime("%H:%M:%S.%f").str[:-3] + for i, parameter in enumerate( + selected_parameters, start=num_waveform_plots + 1 + ): + add_trace_to_subplot( + fig, parameter_df, parameter, i, is_parameter=True ) - trace = go.Scatter( - x=df["Time (s)"], - y=y_data, - mode="lines", - name=waveform, - customdata=abs_time_str, - hovertemplate=( - "%{y}
" - "Time: %{x:.2f} s
" - "Abs Time: %{customdata}" - "" - ), - ) - fig.add_trace(trace, row=i, col=1) - fig.update_yaxes(title_text=waveform, row=i, col=1) - # Calculate dynamic height (approx 300px per plot) plot_height = max(400, 300 * num_plots) @@ -147,21 +285,47 @@ def update_graph(selected_waveforms): return app -if __name__ == "__main__": - # ==== CLI Argument ==== - if len(sys.argv) != 2: - print( - "Usage: python " - "src/lung_utils/hamilton_ventilator/waveform_plotter.py " - "/path/to/hamilton_file.txt" - ) - sys.exit(1) +def parse_args(): + parser = argparse.ArgumentParser( + description="View Hamilton ventilator waveform TXT files." + ) + parser.add_argument("waveform_file", help="Path to Hamilton waveform.txt") + parser.add_argument( + "--include-parameters", + action="store_true", + help="Search for a matching P_Hamilton-C6* file and plot fields.", + ) + parser.add_argument( + "--parameter-file", + help="Explicit Hamilton P_Hamilton-C6* parameter file to include.", + ) + return parser.parse_args() + - FILE_PATH = sys.argv[1] +def main(): + args = parse_args() # Load the file - df = load_waveform_txt(FILE_PATH) + df = load_waveform_txt(args.waveform_file) + + parameter_df = None + parameter_file_path = args.parameter_file + if args.include_parameters or args.parameter_file: + if parameter_file_path is None: + parameter_file_path = find_parameter_file(args.waveform_file) + parameter_df = load_parameter_txt( + parameter_file_path, start_time_val=df["Date_Time"].iloc[0] + ) # Create and run the app - app = create_dash_app(df, FILE_PATH) + app = create_dash_app( + df, + args.waveform_file, + parameter_df=parameter_df, + parameter_file_path=parameter_file_path, + ) app.run(debug=True) + + +if __name__ == "__main__": + main() diff --git a/tests/lung_utils/hamilton_ventilator/test_waveform_exporter.py b/tests/lung_utils/hamilton_ventilator/test_waveform_exporter.py new file mode 100644 index 0000000..bb9d0bf --- /dev/null +++ b/tests/lung_utils/hamilton_ventilator/test_waveform_exporter.py @@ -0,0 +1,329 @@ +from unittest.mock import patch + +import numpy as np +import pandas as pd +import pytest +from lung_utils.hamilton_ventilator.waveform_exporter import ( + extract_waveforms, + format_fourc_linearinterpolation, + get_waveform_fields, + main, + make_variable_name, + save_waveforms_npy, +) + + +@pytest.fixture +def sample_dataframe(): + return pd.DataFrame( + { + "Date_Time": [44917.0, 44917.00001, 44917.00002], + "Time (s)": [0.0, 1.0, 2.0], + "Pressure Waveform": [10.0, 20.0, 30.0], + "Flow": [0.0, 2.0, 0.0], + "Status": [0, 0, 0], + } + ) + + +def test_get_waveform_fields_excludes_metadata(sample_dataframe): + assert get_waveform_fields(sample_dataframe) == [ + "Pressure Waveform", + "Flow", + ] + + +def test_make_variable_name_sanitizes_field_name(): + assert make_variable_name("Pressure Waveform") == "pressure_waveform" + assert make_variable_name("3 Flow [l/min]") == "waveform_3_flow_l_min" + + +def test_extract_waveforms_uses_original_samples_by_default( + sample_dataframe, +): + waveforms = extract_waveforms( + sample_dataframe, + fields=["Pressure Waveform", "Flow"], + start=0.5, + end=2.0, + ) + + np.testing.assert_allclose(waveforms["time"], [0.5, 1.5]) + np.testing.assert_allclose(waveforms["Pressure Waveform"], [20.0, 30.0]) + np.testing.assert_allclose(waveforms["Flow"], [2.0, 0.0]) + + +def test_extract_waveforms_preserves_original_sample_times(sample_dataframe): + waveforms = extract_waveforms( + sample_dataframe, + fields=["Pressure Waveform"], + start=0.5, + end=2.0, + preserve_time=True, + ) + + np.testing.assert_allclose(waveforms["time"], [1.0, 2.0]) + np.testing.assert_allclose(waveforms["Pressure Waveform"], [20.0, 30.0]) + + +def test_extract_waveforms_resamples_multiple_fields(sample_dataframe): + waveforms = extract_waveforms( + sample_dataframe, + fields=["Pressure Waveform", "Flow"], + start=0.5, + end=1.5, + sampling_rate=2.0, + ) + + np.testing.assert_allclose(waveforms["time"], [0.0, 0.5, 1.0]) + np.testing.assert_allclose( + waveforms["Pressure Waveform"], [15.0, 20.0, 25.0] + ) + np.testing.assert_allclose(waveforms["Flow"], [1.0, 2.0, 1.0]) + + +def test_extract_waveforms_rejects_invalid_interval(sample_dataframe): + with pytest.raises(ValueError, match="End time"): + extract_waveforms( + sample_dataframe, + fields=["Flow"], + start=1.0, + end=1.0, + sampling_rate=10.0, + ) + + +def test_save_waveforms_npy_uses_separate_arrays_per_field(tmp_path): + output = tmp_path / "waveforms.npy" + save_waveforms_npy( + { + "time": np.array([0.0, 1.0]), + "Flow": np.array([0.0, 2.0]), + "Pressure": np.array([10.0, 20.0]), + }, + str(output), + ) + + loaded = np.load(output, allow_pickle=True).item() + + assert set(loaded) == {"time", "Flow", "Pressure"} + np.testing.assert_allclose(loaded["time"], [0.0, 1.0]) + np.testing.assert_allclose(loaded["Flow"], [0.0, 2.0]) + np.testing.assert_allclose(loaded["Pressure"], [10.0, 20.0]) + + +def test_format_fourc_linearinterpolation(): + function_string = format_fourc_linearinterpolation( + times=np.array([0.0, 0.5, 1.0]), + values=np.array([10.0, 15.0, 20.0]), + funct_number=7, + variable_name="p", + ) + + assert function_string == "\n".join( + [ + "FUNCT7:", + "- SYMBOLIC_FUNCTION_OF_TIME: p", + "- VARIABLE: 0", + " NAME: p", + " TYPE: linearinterpolation", + " NUMPOINTS: 3", + " TIMES:", + " - 0", + " - 0.5", + " - 1", + " VALUES:", + " - 10", + " - 15", + " - 20", + ] + ) + + +def test_main_lists_fields(tmp_path, capsys): + waveform_file = tmp_path / "waveform.txt" + waveform_file.write_text( + "Date_Time\tPressure\tFlow\tStatus\n" + "44917.0\t10\t0\t0\n" + "44917.00001\t20\t2\t0\n", + encoding="latin1", + ) + + with patch( + "sys.argv", + [ + "hamilton-waveform-export", + str(waveform_file), + "--list-fields", + ], + ): + main() + + assert capsys.readouterr().out == "Pressure\nFlow\n" + + +def test_main_writes_npy_for_multiple_fields(tmp_path): + waveform_file = tmp_path / "waveform.txt" + output_file = tmp_path / "waveforms.npy" + waveform_file.write_text( + "Date_Time\tPressure\tFlow\n" + "44917.0\t10\t0\n" + "44917.000011574074\t20\t2\n" + "44917.000023148148\t30\t0\n", + encoding="latin1", + ) + + with patch( + "sys.argv", + [ + "hamilton-waveform-export", + str(waveform_file), + "--fields", + "Pressure", + "Flow", + "--start", + "0", + "--end", + "2.00000046752", + "--output", + str(output_file), + ], + ): + main() + + loaded = np.load(output_file, allow_pickle=True).item() + assert set(loaded) == {"time", "Pressure", "Flow"} + assert len(loaded["time"]) == 3 + + +def test_main_writes_npy_for_all_fields(tmp_path): + waveform_file = tmp_path / "waveform.txt" + output_file = tmp_path / "waveforms.npy" + waveform_file.write_text( + "Date_Time\tPressure\tFlow\n" + "44917.0\t10\t0\n" + "44917.000011574074\t20\t2\n", + encoding="latin1", + ) + + with patch( + "sys.argv", + [ + "hamilton-waveform-export", + str(waveform_file), + "--all-fields", + "--start", + "0", + "--end", + "1.00000023376", + "--output", + str(output_file), + ], + ): + main() + + loaded = np.load(output_file, allow_pickle=True).item() + assert set(loaded) == {"time", "Pressure", "Flow"} + + +def test_main_prompts_for_multiple_fields_and_writes_npy(tmp_path): + waveform_file = tmp_path / "waveform.txt" + output_file = tmp_path / "waveforms.npy" + waveform_file.write_text( + "Date_Time\tPressure\tFlow\n" + "44917.0\t10\t0\n" + "44917.000011574074\t20\t2\n", + encoding="latin1", + ) + + with ( + patch( + "sys.argv", + [ + "hamilton-waveform-export", + str(waveform_file), + "--output", + str(output_file), + ], + ), + patch("builtins.input", side_effect=["1,2", "0", ""]), + ): + main() + + loaded = np.load(output_file, allow_pickle=True).item() + assert set(loaded) == {"time", "Pressure", "Flow"} + + +def test_main_copies_fourc_function_to_clipboard(tmp_path, capsys): + waveform_file = tmp_path / "waveform.txt" + waveform_file.write_text( + "Date_Time\tPressure\tFlow\n" + "44917.0\t10\t0\n" + "44917.000011574074\t20\t2\n" + "44917.000023148148\t30\t0\n", + encoding="latin1", + ) + + with ( + patch( + "sys.argv", + [ + "hamilton-waveform-export", + str(waveform_file), + "--format", + "fourc", + "--fields", + "Pressure", + "--start", + "0", + "--end", + "2.00000046752", + "--funct", + "2", + "--variable-name", + "p", + ], + ), + patch( + "lung_utils.hamilton_ventilator." + "waveform_exporter.copy_to_clipboard" + ) as copy, + ): + main() + + copied = copy.call_args.args[0] + assert "FUNCT2:" in copied + assert "- SYMBOLIC_FUNCTION_OF_TIME: p" in copied + assert " NUMPOINTS: 3" in copied + assert "Copied 4C linearinterpolation function" in capsys.readouterr().out + + +def test_main_rejects_multiple_fields_for_fourc(tmp_path): + waveform_file = tmp_path / "waveform.txt" + waveform_file.write_text( + "Date_Time\tPressure\tFlow\n" + "44917.0\t10\t0\n" + "44917.000011574074\t20\t2\n", + encoding="latin1", + ) + + with ( + patch( + "sys.argv", + [ + "hamilton-waveform-export", + str(waveform_file), + "--format", + "fourc", + "--fields", + "Pressure", + "Flow", + "--start", + "0", + "--end", + "1.1", + ], + ), + pytest.raises(SystemExit), + ): + main() diff --git a/tests/lung_utils/hamilton_ventilator/test_waveform_plotter.py b/tests/lung_utils/hamilton_ventilator/test_waveform_plotter.py index 59a2a28..cb47eb2 100644 --- a/tests/lung_utils/hamilton_ventilator/test_waveform_plotter.py +++ b/tests/lung_utils/hamilton_ventilator/test_waveform_plotter.py @@ -3,7 +3,11 @@ import pandas as pd import pytest from dash import Dash -from lung_utils.hamilton_ventilator.waveform_plotter import create_dash_app +from lung_utils.hamilton_ventilator.waveform_plotter import ( + create_dash_app, + find_parameter_file, + load_parameter_txt, +) @pytest.fixture @@ -23,6 +27,22 @@ def sample_dataframe(): return pd.DataFrame(data) +@pytest.fixture +def sample_parameter_dataframe(sample_dataframe): + """Fixture for a sample Hamilton parameter dataframe.""" + return pd.DataFrame( + { + "Date_Time": sample_dataframe["Date_Time"], + "Time (s)": sample_dataframe["Time (s)"], + "Breath Number": [1, 2, 3, 4], + "Mode Name": ["(S)CMV", "(S)CMV", "PCV+", "PCV+"], + "PEEP/ CPAP /cmH2O": [8, 8, 15, 15], + "Tidal Volume /ml": [350, 350, 350, 350], + "Absolute_Time": sample_dataframe["Absolute_Time"], + } + ) + + def test_create_dash_app(sample_dataframe): """Test the create_dash_app function.""" file_path = "test_file.txt" @@ -79,3 +99,72 @@ def test_create_dash_app_layout(mock_graph, mock_dropdown, sample_dataframe): multi=True, ) mock_graph.assert_called_once_with(id="waveform-plot") + + +def test_create_dash_app_with_parameter_data( + sample_dataframe, sample_parameter_dataframe +): + """Test layout includes parameter controls when P data is supplied.""" + app = create_dash_app( + sample_dataframe, + "waveform.txt", + parameter_df=sample_parameter_dataframe, + parameter_file_path="parameters.txt", + ) + + assert isinstance(app, Dash) + child_ids = [getattr(child, "id", None) for child in app.layout.children] + assert "parameter-file-info" in child_ids + assert "parameter-dropdown" in child_ids + assert child_ids[-1] == "waveform-plot" + + +def test_find_parameter_file_finds_single_match(tmp_path): + """Test parameter discovery finds one matching file.""" + waveform_file = tmp_path / "W_Hamilton-C6__example_Waves_001.txt" + parameter_file = tmp_path / "P_Hamilton-C6__example_All_001.txt" + waveform_file.write_text("", encoding="utf-8") + parameter_file.write_text("", encoding="utf-8") + + assert find_parameter_file(waveform_file) == str(parameter_file) + + +def test_find_parameter_file_raises_for_missing_match(tmp_path): + """Test parameter discovery raises when no P file exists.""" + waveform_file = tmp_path / "W_Hamilton-C6__example_Waves_001.txt" + waveform_file.write_text("", encoding="utf-8") + + with pytest.raises(FileNotFoundError): + find_parameter_file(waveform_file) + + +def test_find_parameter_file_raises_for_multiple_matches(tmp_path): + """Test parameter discovery raises when more than one P file exists.""" + waveform_file = tmp_path / "W_Hamilton-C6__example_Waves_001.txt" + waveform_file.write_text("", encoding="utf-8") + (tmp_path / "P_Hamilton-C6__example_All_001.txt").write_text( + "", encoding="utf-8" + ) + (tmp_path / "P_Hamilton-C6__example_All_002.txt").write_text( + "", encoding="utf-8" + ) + + with pytest.raises(ValueError, match="Multiple Hamilton parameter files"): + find_parameter_file(waveform_file) + + +def test_load_parameter_txt_uses_waveform_start_time(tmp_path): + """Test P-file times can be aligned to waveform start.""" + parameter_file = tmp_path / "P_Hamilton-C6__example_All_001.txt" + parameter_file.write_text( + "Date_Time\tBreath Number\tMode Name\n" + "44917.000010\t1\t(S)CMV\n" + "44917.000020\t2\t(S)CMV\n", + encoding="latin1", + ) + + df = load_parameter_txt(parameter_file, start_time_val=44917.000000) + + assert list(df["Mode Name"]) == ["(S)CMV", "(S)CMV"] + assert df["Time (s)"].iloc[0] == pytest.approx(0.864) + assert df["Time (s)"].iloc[1] == pytest.approx(1.728)