diff --git a/chem_spectra/lib/composer/ni.py b/chem_spectra/lib/composer/ni.py index 5875da9..50bd325 100644 --- a/chem_spectra/lib/composer/ni.py +++ b/chem_spectra/lib/composer/ni.py @@ -405,45 +405,75 @@ def __fakto(self): return -1 return 1 - def tf_img(self): - plt.rcParams['figure.figsize'] = [16, 9] - plt.rcParams['figure.dpi'] = 200 - plt.rcParams['font.size'] = 14 - - cv_info = self.__cv_display_info() if self.core.is_cyclic_volta else None - self._cv_density_scale = 1.0 - if cv_info: - cv_display = cv_info.get('cv_display') or {} - scale_val = cv_display.get('yScaleFactor', 1.0) if isinstance(cv_display, dict) else 1.0 - try: - self._cv_density_scale = float(scale_val) - except Exception: - self._cv_density_scale = 1.0 - - # PLOT data - y_values = self.core.ys - if self.core.is_cyclic_volta and self._cv_density_scale != 1.0: - y_values = self.core.ys * self._cv_density_scale - plt.plot(self.core.xs, y_values) - x_max, x_min = self.core.boundary['x']['max'], self.core.boundary['x']['min'] # noqa: E501 + def _ensure_itg_mpy_from_core_tables(self): + refShift = self.refShift + if (len(self.all_itgs) == 0 and len(self.core.itg_table) > 0 + and not self.core.params['integration'].get('edited') + and ('originStack' not in self.core.params['integration'])): + core_itg_table = self.core.itg_table[0] + for itg in core_itg_table.split('\n'): + clear_itg = itg.replace('(', '').replace(')', '') + split_itg = clear_itg.split(',') + if len(split_itg) > 2: + self.all_itgs.append({ + 'xL': float(split_itg[0].strip()), + 'xU': float(split_itg[1].strip()), + 'area': float(split_itg[2].strip()), + }) + + if (not self.core.non_nmr and len(self.mpys) == 0 + and len(self.core.mpy_itg_table) > 0 + and len(self.core.mpy_pks_table) > 0 + and not self.core.params['integration'].get('edited') + and ('originStack' not in self.core.params['integration'])): + tmp_dic_mpy_peaks = {} + core_mpy_pks_table = self.core.mpy_pks_table[0] + for peak in core_mpy_pks_table.split('\n'): + clear_peak = peak.replace('(', '').replace(')', '') + split_peak = clear_peak.split(',') + if len(split_peak) < 3: + continue + idx_peakStr = split_peak[0].strip() + if idx_peakStr not in tmp_dic_mpy_peaks: + tmp_dic_mpy_peaks[idx_peakStr] = [] + tmp_dic_mpy_peaks[idx_peakStr].append({ + 'x': float(split_peak[1].strip()), + 'y': float(split_peak[2].strip()), + }) - xlim_left, xlim_right = [x_min, x_max] if (self.core.is_tga or self.core.is_gc or self.core.is_uv_vis or self.core.is_hplc_uv_vis or self.core.is_xrd or self.core.is_cyclic_volta or self.core.is_sec or self.core.is_cds or self.core.is_aif or self.core.is_emissions or self.core.is_dls_acf or self.core.is_dls_intensity) else [x_max, x_min] # noqa: E501 - plt.xlim(xlim_left, xlim_right) + for mpy in self.core.mpy_itg_table[0].split('\n'): + clear_mpy = mpy.replace('(', '').replace(')', '') + split_mpy = clear_mpy.split(',') + if len(split_mpy) <= 7: + continue + idxStr = split_mpy[0].strip() + peaks = tmp_dic_mpy_peaks.get(idxStr, []) + if not peaks: + # calc_mpy_center divides by the peak count, so a + # multiplet without matching peaks cannot be drawn + continue + mpy_item = { + 'mpyType': split_mpy[6].strip(), + 'xExtent': { + 'xL': float(split_mpy[1].strip()) + refShift, + 'xU': float(split_mpy[2].strip()) + refShift, + }, + 'yExtent': { + 'yL': float(split_mpy[3].strip()) + refShift, + 'yU': float(split_mpy[4].strip()) + refShift, + }, + 'peaks': peaks, + 'area': 1.0, + } + self.mpys.append(mpy_item) + + def plot_overlays(self, plt, y_values=None, adjust_xlim=True): + y_values = self.core.ys if y_values is None else y_values y_max, y_min = np.max(y_values), np.min(y_values) h = y_max - y_min - w = x_max - x_min - y_boundary_min = y_min - h * 0.2 + w = self.core.boundary['x']['max'] - self.core.boundary['x']['min'] y_boundary_max = y_max + h * 0.5 - if self.core.is_cyclic_volta: - ymax_abs = max(abs(y_min), abs(y_max)) - if ymax_abs > 0: - self._cv_axis_exp = int(np.floor(np.log10(ymax_abs))) - else: - self._cv_axis_exp = 0 - self._cv_axis_base = (10.0 ** self._cv_axis_exp) if self._cv_axis_exp != 0 else 1.0 - - # PLOT peaks faktor = self.__fakto() path_data = [ (mpath.Path.MOVETO, (0, faktor * 5)), @@ -457,8 +487,7 @@ def tf_img(self): cirle_codes = np.concatenate([circle.codes, codes]) cut_star_marker = mpath.Path(cirle_verts, cirle_codes) - x_peaks = [] - y_peaks = [] + x_peaks, y_peaks = [], [] if self.core.edit_peaks: x_peaks = self.core.edit_peaks['x'] y_peaks = self.core.edit_peaks['y'] @@ -466,52 +495,38 @@ def tf_img(self): x_peaks = self.core.auto_peaks['x'] y_peaks = self.core.auto_peaks['y'] - x_peckers = [] - y_peckers = [] + x_peckers, y_peckers = [], [] x_peaks_ref, y_peaks_ref = [], [] if self.core.is_cyclic_volta: display_scale = getattr(self, '_cv_density_scale', 1.0) - x_peaks = [] - y_peaks = [] - listMaxMinPeaks = [] - if self.core.params['list_max_min_peaks'] is not None: - listMaxMinPeaks = self.core.params['list_max_min_peaks'] - + x_peaks, y_peaks = [], [] + listMaxMinPeaks = self.core.params['list_max_min_peaks'] or [] for peak in listMaxMinPeaks: - max_peak, min_peak = None, None - if 'max' in peak: - max_peak = peak['max'] - if 'min' in peak: - min_peak = peak['min'] + max_peak = peak.get('max') + min_peak = peak.get('min') x_max_peak, y_max_peak = self.__get_xy_of_peak(max_peak) x_min_peak, y_min_peak = self.__get_xy_of_peak(min_peak) - - if (x_max_peak == '' and x_min_peak == ''): - # ignore if missing both max and min peak + if x_max_peak == '' and x_min_peak == '': continue - - if (x_max_peak == '' and y_max_peak == ''): + if x_max_peak == '' and y_max_peak == '': x_peaks.extend([x_min_peak]) y_peaks.extend([y_min_peak]) - elif (x_min_peak == '' and y_min_peak == ''): + elif x_min_peak == '' and y_min_peak == '': x_peaks.extend([x_max_peak]) y_peaks.extend([y_max_peak]) else: - is_ref = peak.get('isRef', False) if 'isRef' in peak else False + is_ref = peak.get('isRef', False) if is_ref: x_peaks_ref.extend([x_max_peak, x_min_peak]) y_peaks_ref.extend([y_max_peak, y_min_peak]) else: x_peaks.extend([x_max_peak, x_min_peak]) y_peaks.extend([y_max_peak, y_min_peak]) - - if 'pecker' in peak and peak['pecker'] is not None: + if peak.get('pecker') is not None: pecker = peak['pecker'] - x_pecker, y_pecker = pecker['x'], pecker['y'] - x_peckers.append(x_pecker) - y_peckers.append(y_pecker) + x_peckers.append(pecker['x']) + y_peckers.append(pecker['y']) - # display x value of peak for cyclic voltammetry if display_scale != 1.0: y_peaks = [y * display_scale for y in y_peaks] y_peckers = [y * display_scale for y in y_peckers] @@ -520,123 +535,99 @@ def tf_img(self): for i in range(len(x_peaks)): x_pos = x_peaks[i] y_pos = y_peaks[i] + h * 0.1 - x_float = '{:.2e}'.format(x_pos) - y_float = '{:.2e}'.format(y_peaks[i]) - peak_label = 'x: {x}\ny: {y}'.format(x=x_float, y=y_float) + peak_label = 'x: {x}\ny: {y}'.format( + x='{:.2e}'.format(x_pos), y='{:.2e}'.format(y_peaks[i]), + ) plt.text(x_pos, y_pos, peak_label) - # display x value of ref peak for cyclic voltammetry for i in range(len(x_peaks_ref)): x_pos = x_peaks_ref[i] y_pos = y_peaks_ref[i] + h * 0.1 - x_float = '{:.2e}'.format(x_pos) - y_float = '{:.2e}'.format(y_peaks_ref[i]) - peak_label = 'x: {x}\ny: {y}'.format(x=x_float, y=y_float) + peak_label = 'x: {x}\ny: {y}'.format( + x='{:.2e}'.format(x_pos), y='{:.2e}'.format(y_peaks_ref[i]), + ) plt.text(x_pos, y_pos, peak_label) - plt.plot( - x_peaks, - y_peaks, - 'r', - ls='', - marker=marker, - markersize=50, - ) - - plt.plot( - x_peckers, - y_peckers, - 'g', - ls='', - marker=marker, - markersize=50, - ) + if x_peaks: + plt.plot(x_peaks, y_peaks, 'r', ls='', marker=marker, markersize=50) + if x_peckers: + plt.plot(x_peckers, y_peckers, 'g', ls='', marker=marker, markersize=50) + if x_peaks_ref: + plt.plot(x_peaks_ref, y_peaks_ref, 'r', ls='', marker=cut_star_marker, markersize=50) - plt.plot( - x_peaks_ref, - y_peaks_ref, - 'r', - ls='', - marker=cut_star_marker, - markersize=50, - ) - - # ----- Calculate integration ----- refShift, refArea = self.refShift, self.refArea - if (len(self.all_itgs) == 0 and len(self.core.itg_table) > 0 and not self.core.params['integration'].get('edited') and ('originStack' not in self.core.params['integration'])): - core_itg_table = self.core.itg_table[0] - itg_table = core_itg_table.split('\n') - for itg in itg_table: - clear_itg = itg.replace('(', '') - clear_itg = clear_itg.replace(')', '') - split_itg = clear_itg.split(',') - if (len(split_itg) > 2): - xLStr = split_itg[0].strip() - xUStr = split_itg[1].strip() - areaStr = split_itg[2].strip() - self.all_itgs.append({'xL': float(xLStr), 'xU': float(xUStr), 'area': float(areaStr)}) # noqa: E501 - + self._ensure_itg_mpy_from_core_tables() - # ----- Calculate multiplicity (NMR only) ----- - if (not self.core.non_nmr and len(self.mpys) == 0 and len(self.core.mpy_itg_table) > 0 and not self.core.params['integration'].get('edited') and ('originStack' not in self.core.params['integration'])): - core_mpy_pks_table = self.core.mpy_pks_table[0] - mpy_pks_table = core_mpy_pks_table.split('\n') - tmp_dic_mpy_peaks = {} - for peak in mpy_pks_table: - clear_peak = peak.replace('(', '') - clear_peak = clear_peak.replace(')', '') - split_peak = clear_peak.split(',') - idx_peakStr = split_peak[0].strip() - xStr = split_peak[1].strip() - yStr = split_peak[2].strip() - if idx_peakStr not in tmp_dic_mpy_peaks: - tmp_dic_mpy_peaks[idx_peakStr] = [] - - tmp_dic_mpy_peaks[idx_peakStr].append({'x': float(xStr), 'y': float(yStr)}) - - core_mpy_itg_table = self.core.mpy_itg_table[0] - mpy_itg_table = core_mpy_itg_table.split('\n') - for mpy in mpy_itg_table: - clear_mpy = mpy.replace('(', '') - clear_mpy = clear_mpy.replace(')', '') - split_mpy = clear_mpy.split(',') - mpy_item = { 'mpyType': '', 'xExtent': {'xL': 0.0, 'xU': 0.0}, 'yExtent': {'yL': 0.0, 'yU': 0.0}, 'peaks': [], 'area': 1.0 } - if (len(split_mpy) > 7): - idxStr = split_mpy[0].strip() - xLStr = split_mpy[1].strip() - xUStr = split_mpy[2].strip() - mpy_item['xExtent']['xL'] = float(xLStr) + refShift - mpy_item['xExtent']['xU'] = float(xUStr) + refShift - yLStr = split_mpy[3].strip() - yUStr = split_mpy[4].strip() - mpy_item['yExtent']['yL'] = float(yLStr) + refShift - mpy_item['yExtent']['yU'] = float(yUStr) + refShift - typeStr = split_mpy[6].strip() - mpy_item['mpyType'] = typeStr - mpy_item['peaks'] = tmp_dic_mpy_peaks[idxStr] - self.mpys.append(mpy_item) - - # ----- PLOT integration ----- - itg_h = y_max + h * 0.6 - itg_h = itg_h + itg_h * 0.1 itg_value_position_y = y_min - h * 0.25 - if (len(self.mpys) == 0): - itg_value_position_y = y_min - h * 0.05 - y_boundary_max = self.__draw_integrals(plt, refShift, refArea, y_max, h, itg_value_position_y) + if len(self.mpys) == 0: + itg_value_position_y = y_min - h * 0.05 + y_boundary_max = self.__draw_integrals( + plt, refShift, refArea, y_max, h, itg_value_position_y, + ) y_boundary_min = itg_value_position_y - h * 0.1 - - # ----- PLOT multiplicity (NMR only) ----- + if not self.core.non_nmr: mpy_h = y_min - h * 0.03 for mpy in self.mpys: - xL, xU, area, typ, peaks = mpy['xExtent']['xL'] - refShift, mpy['xExtent']['xU'] - refShift, mpy['area'] * refArea, mpy['mpyType'], mpy['peaks'] # noqa: E501 + xL = mpy['xExtent']['xL'] - refShift + xU = mpy['xExtent']['xU'] - refShift + typ = mpy['mpyType'] plt.plot([xL, xU], [mpy_h, mpy_h], color='#DA70D6') - plt.plot([xL, xL], [mpy_h + h * 0.01, mpy_h - h * 0.01], color='#DA70D6') # noqa: E501 - plt.plot([xU, xU], [mpy_h + h * 0.01, mpy_h - h * 0.01], color='#DA70D6') # noqa: E501 - plt.text((xL + xU) / 2, mpy_h - h * 0.01, '{:0.3f} ({})'.format(calc_mpy_center(mpy['peaks'], refShift, mpy['mpyType']), typ), color='#DA70D6', size=7, rotation=90., ha='right', va='top', rotation_mode='anchor') # noqa: E501 - for p in peaks: - x = p['x'] - plt.plot([x - refShift, x - refShift], [mpy_h, mpy_h + h * 0.02], color='#DA70D6') # noqa: E501 + plt.plot([xL, xL], [mpy_h + h * 0.01, mpy_h - h * 0.01], color='#DA70D6') + plt.plot([xU, xU], [mpy_h + h * 0.01, mpy_h - h * 0.01], color='#DA70D6') + plt.text( + (xL + xU) / 2, mpy_h - h * 0.01, + '{:0.3f} ({})'.format(calc_mpy_center(mpy['peaks'], refShift, typ), typ), + color='#DA70D6', size=7, rotation=90., ha='right', va='top', + rotation_mode='anchor', + ) + for p in mpy['peaks']: + x = p['x'] - refShift + plt.plot([x, x], [mpy_h, mpy_h + h * 0.02], color='#DA70D6') + + self.__generate_info_box(plt) + y_boundary_max = self.__draw_peaks( + plt, x_peaks, y_peaks, h, w, + y_boundary_max * (1.1 if self.core.is_ir else 1.5), + adjust_xlim=adjust_xlim, + ) + return y_boundary_min, y_boundary_max + + def tf_img(self): + plt.rcParams['figure.figsize'] = [16, 9] + plt.rcParams['figure.dpi'] = 200 + plt.rcParams['font.size'] = 14 + + cv_info = self.__cv_display_info() if self.core.is_cyclic_volta else None + self._cv_density_scale = 1.0 + if cv_info: + cv_display = cv_info.get('cv_display') or {} + scale_val = cv_display.get('yScaleFactor', 1.0) if isinstance(cv_display, dict) else 1.0 + try: + self._cv_density_scale = float(scale_val) + except Exception: + self._cv_density_scale = 1.0 + + # PLOT data + y_values = self.core.ys + if self.core.is_cyclic_volta and self._cv_density_scale != 1.0: + y_values = self.core.ys * self._cv_density_scale + plt.plot(self.core.xs, y_values) + x_max, x_min = self.core.boundary['x']['max'], self.core.boundary['x']['min'] # noqa: E501 + + xlim_left, xlim_right = [x_min, x_max] if (self.core.is_tga or self.core.is_gc or self.core.is_uv_vis or self.core.is_hplc_uv_vis or self.core.is_xrd or self.core.is_cyclic_volta or self.core.is_sec or self.core.is_cds or self.core.is_aif or self.core.is_emissions or self.core.is_dls_acf or self.core.is_dls_intensity) else [x_max, x_min] # noqa: E501 + plt.xlim(xlim_left, xlim_right) + y_max, y_min = np.max(y_values), np.min(y_values) + + if self.core.is_cyclic_volta: + ymax_abs = max(abs(y_min), abs(y_max)) + if ymax_abs > 0: + self._cv_axis_exp = int(np.floor(np.log10(ymax_abs))) + else: + self._cv_axis_exp = 0 + self._cv_axis_base = (10.0 ** self._cv_axis_exp) if self._cv_axis_exp != 0 else 1.0 + + y_boundary_min, y_boundary_max = self.plot_overlays(plt, y_values, adjust_xlim=True) # PLOT label if (self.core.is_xrd): @@ -659,11 +650,6 @@ def tf_img(self): plt.locator_params(nbins=self.__plt_nbins()) plt.grid(False) - self.__generate_info_box(plt) - - y_boundary_max = self.__draw_peaks(plt, x_peaks, y_peaks, h, w, y_boundary_max * (1.1 if self.core.is_ir else 1.5)) - - plt.ylim( y_boundary_min, y_boundary_max, @@ -676,7 +662,6 @@ def tf_img(self): self._cv_axis_exp = int(np.floor(np.log10(ymax_abs))) else: self._cv_axis_exp = 0 - print(f"[tf_img] ymax_abs={ymax_abs}, exp={self._cv_axis_exp}") self._cv_axis_base = (10.0 ** self._cv_axis_exp) if self._cv_axis_exp != 0 else 1.0 ax.yaxis.set_major_formatter(ticker.FuncFormatter(lambda y, _: @@ -697,7 +682,7 @@ def tf_img(self): fmt.set_scientific(True) fmt.set_powerlimits((-1, 1)) ax.yaxis.set_major_formatter(fmt) - + # Save tf_img = tempfile.NamedTemporaryFile(suffix='.png') plt.savefig(tf_img, format='png') @@ -705,7 +690,7 @@ def tf_img(self): plt.clf() plt.cla() return tf_img - + def __uses_auc_drawing(self): return self.core.is_hplc_uv_vis or self.core.is_uv_vis @@ -788,7 +773,7 @@ def __draw_integrals(self, plt, refShift, refArea, y_max, h, itg_value_position_ return y_boundary_max - def __draw_peaks(self, plt, x_peaks, y_peaks, h, w, y_boundary_max): + def __draw_peaks(self, plt, x_peaks, y_peaks, h, w, y_boundary_max, adjust_xlim=True): if self.core.non_nmr == True or len(x_peaks) == 0: return y_boundary_max @@ -878,10 +863,11 @@ def __draw_peaks(self, plt, x_peaks, y_peaks, h, w, y_boundary_max): elif self.core.ncl == '1H': x_boundary_min = min(x_boundary_min, -0.2) x_boundary_max = max(x_boundary_max, 9.0) - plt.xlim( - x_boundary_max, - x_boundary_min, - ) + if adjust_xlim: + plt.xlim( + x_boundary_max, + x_boundary_min, + ) y_boundary_max = min(y_boundary_max, highest_peak_anotation) return y_boundary_max diff --git a/chem_spectra/lib/converter/bagit/base.py b/chem_spectra/lib/converter/bagit/base.py index f5b955d..7ced2f5 100644 --- a/chem_spectra/lib/converter/bagit/base.py +++ b/chem_spectra/lib/converter/bagit/base.py @@ -2,6 +2,7 @@ import base64 import tempfile import json +import logging import math from chem_spectra.lib.converter.jcamp.base import JcampBaseConverter @@ -16,6 +17,8 @@ import matplotlib.pyplot as plt # noqa: E402 from matplotlib import ticker # noqa: E402 +logger = logging.getLogger(__name__) + class BagItBaseConverter: def __init__(self, target_dir, params=False, fname=''): @@ -161,10 +164,17 @@ def __combine_images(self, list_composer, list_file_names=None): list_composer = non_lcms_ni plt.rcParams['figure.figsize'] = [16, 9] + plt.rcParams['figure.dpi'] = 200 plt.rcParams['font.size'] = 14 cv_mode = False cv_abs_max = 0.0 + try: + active_idx = int(self.params.get('jcamp_idx', 0) or 0) + except (TypeError, ValueError): + active_idx = 0 + active_composer = None + active_y_values = None for idx, composer in enumerate(list_composer): filename = str(idx) if (list_file_names is not None) and idx < len(list_file_names): @@ -211,6 +221,14 @@ def __combine_images(self, list_composer, list_file_names=None): marker = 'v' plt.plot(xs, y_values, label=filename, marker=marker) + + is_active = idx == active_idx + if is_active: + active_composer = composer + active_y_values = y_values + if composer.core.is_cyclic_volta: + composer._cv_density_scale = scale + # PLOT label if (composer.core.is_xrd): waveLength = composer.core.params['waveLength'] @@ -226,6 +244,18 @@ def __combine_images(self, list_composer, list_file_names=None): else: plt.ylabel("Y ({})".format(composer.core.label['y']), fontsize=18) + if active_composer is not None: + try: + y_boundary_min, y_boundary_max = active_composer.plot_overlays( + plt, active_y_values, adjust_xlim=False, + ) + ymin, ymax = plt.gca().get_ylim() + plt.ylim(min(ymin, y_boundary_min), max(ymax, y_boundary_max)) + except Exception: + logger.exception( + 'combine_images: failed to draw overlays for active spectrum' + ) + if cv_mode and cv_abs_max > 0: exp = int(math.floor(math.log10(cv_abs_max))) if cv_abs_max > 0 else 0 base = (10.0 ** exp) if exp != 0 else 1.0 diff --git a/chem_spectra/lib/converter/share.py b/chem_spectra/lib/converter/share.py index 69aadcc..0d0b7b5 100644 --- a/chem_spectra/lib/converter/share.py +++ b/chem_spectra/lib/converter/share.py @@ -55,7 +55,10 @@ def parse_params(params): waveLength = json.loads(waveLength) if waveLength else default_wavelength jcamp_idx = params.get('jcamp_idx', 0) - jcamp_idx = jcamp_idx if jcamp_idx else 0 + try: + jcamp_idx = int(jcamp_idx) if jcamp_idx else 0 + except (TypeError, ValueError): + jcamp_idx = 0 axesUnitsJson = params.get('axesUnits') axesUnitsDic = json.loads(axesUnitsJson) if axesUnitsJson else None axesUnits = None diff --git a/chem_spectra/model/transformer.py b/chem_spectra/model/transformer.py index e9746a8..c158e2d 100644 --- a/chem_spectra/model/transformer.py +++ b/chem_spectra/model/transformer.py @@ -1,4 +1,5 @@ import json +import logging import zipfile import tempfile import glob # noqa: F401 @@ -20,12 +21,13 @@ from chem_spectra.lib.composer.base import BaseComposer # noqa: F401 from chem_spectra.lib.converter.nmrium.base import NMRiumDataConverter import matplotlib.pyplot as plt # noqa: E402 -import matplotlib.path as mpath # noqa: E402 import numpy as np # noqa: E402 from matplotlib import ticker # noqa: E402 from chem_spectra.model.concern.property import decorate_sim_property +logger = logging.getLogger(__name__) + def find_dir(path, name): for root, _, files in os.walk(path): @@ -319,49 +321,24 @@ def tf_nmrium(self): tf_jcamp = nicp.tf_jcamp() return tf_jcamp - def __get_cyclic_volta_ref_peaks(self, curve_idx, extraParams): - x_peaks, y_peaks = [], [] - try: - extras_dict = json.loads(extraParams) if extraParams else None - cyclic_volta_str = extras_dict['cyclicvolta'] if extras_dict else None - cyclic_volta = json.loads(cyclic_volta_str) if cyclic_volta_str else None - spectra_list = cyclic_volta['spectraList'] if cyclic_volta else None - spectra_extra = spectra_list[curve_idx] if spectra_list and curve_idx < len(spectra_list) else None - list_peaks = spectra_extra['list'] if spectra_extra else [] - x_peaks, y_peaks = [], [] - for peak in list_peaks: - min_peak, max_peak, isRef = peak['min'], peak['max'], peak['isRef'] - if isRef == True: - x_peaks.extend([min_peak['x'], max_peak['x']]) - y_peaks.extend([min_peak['y'], max_peak['y']]) - except: - pass - - return x_peaks, y_peaks - def tf_combine(self, list_file_names=None, extraParams=None): if not self.multiple_files: return False - path_data = [ - (mpath.Path.MOVETO, (0, 5)), - (mpath.Path.LINETO, (0, 20)), - ] - codes, verts = zip(*path_data) - - circle = mpath.Path.unit_circle() - cirle_verts = np.concatenate([circle.vertices, verts]) - cirle_codes = np.concatenate([circle.codes, codes]) - cut_star_marker = mpath.Path(cirle_verts, cirle_codes) - plt.rcParams['figure.figsize'] = [16, 9] + plt.rcParams['figure.dpi'] = 200 plt.rcParams['font.size'] = 14 plt.rcParams['legend.loc'] = 'upper left' - curve_idx = self.params.get('jcamp_idx', 0) + try: + curve_idx = int(self.params.get('jcamp_idx', 0) or 0) + except (TypeError, ValueError): + curve_idx = 0 xlabel, ylabel = '', '' xlabel_set, ylabel_set = [], [] dic_x_label, dic_y_label = {}, {} + global_x_min, global_x_max = None, None + any_forward_orientation = False cv_mode = False cv_abs_max = 0.0 @@ -370,8 +347,15 @@ def tf_combine(self, list_file_names=None, extraParams=None): file.name = list_file_names[idx] self.multiple_files[idx] = file + active_name = None + if 0 <= curve_idx < len(self.multiple_files): + active_name = self.multiple_files[curve_idx].name + self.multiple_files.sort(key=lambda file: file.name) - + + active_nicp = None + active_y_values = None + for idx, file in enumerate(self.multiple_files): tf = store_str_in_tmp(file.core) jbcv = JcampBaseConverter(tf.name, self.params) @@ -385,6 +369,7 @@ def tf_combine(self, list_file_names=None, extraParams=None): nicp = NIComposer(nicv) xs, ys = nicp.core.xs, nicp.core.ys y_values = ys + is_active = active_name is not None and file.name == active_name if nicp.core.is_cyclic_volta: cv_state = {} if extraParams: @@ -418,19 +403,8 @@ def tf_combine(self, list_file_names=None, extraParams=None): scale = float(cv_display.get('yScaleFactor', 1.0)) except Exception: scale = 1.0 - print( - "[combined:tf_combine] file=", filename, - "cvDisplay=", cv_display, - "scale=", scale, - "y_max=", float(np.max(ys)) if len(ys) else None, - "cv_state_source=", "extras" if extraParams else "params", - ) if scale != 1.0: y_values = ys * scale - print( - "[combined:tf_combine] file=", filename, - "scaled_y_max=", float(np.max(y_values)) if len(y_values) else None, - ) cv_mode = True try: cv_abs_max = max(cv_abs_max, float(np.max(np.abs(y_values)))) @@ -447,23 +421,37 @@ def tf_combine(self, list_file_names=None, extraParams=None): marker = 'v' plt.plot(xs, y_values, label=filename, marker=marker) + if is_active: + active_nicp = nicp + active_y_values = y_values + if nicp.core.is_cyclic_volta: + nicp._cv_density_scale = scale + # the combine endpoint may deliver the CV state only + # via extraParams, in which case parse_params left + # list_max_min_peaks empty + if not nicp.core.params.get('list_max_min_peaks'): + spectra_list = cv_state.get('spectraList') or [] + if 0 <= curve_idx < len(spectra_list): + nicp.core.params['list_max_min_peaks'] = \ + (spectra_list[curve_idx] or {}).get('list') + + try: + x_max = np.max(xs) + x_min = np.min(xs) + global_x_min = x_min if global_x_min is None else min(global_x_min, x_min) + global_x_max = x_max if global_x_max is None else max(global_x_max, x_max) + except Exception: + pass + if (nicp.core.is_tga or nicp.core.is_gc or nicp.core.is_uv_vis or nicp.core.is_hplc_uv_vis + or nicp.core.is_xrd or nicp.core.is_cyclic_volta or nicp.core.is_sec + or nicp.core.is_cds or nicp.core.is_aif or nicp.core.is_emissions + or nicp.core.is_dls_acf or nicp.core.is_dls_intensity): + any_forward_orientation = True + # PLOT label core_label_x = nicp.core.label['x'] core_label_y = nicp.core.label['y'] if nicp.core.is_cyclic_volta: - x_peaks, y_peaks = self.__get_cyclic_volta_ref_peaks(curve_idx, extraParams) - if y_peaks and y_values is not ys: - y_peaks = [y * scale for y in y_peaks] - - plt.plot( - x_peaks, - y_peaks, - 'r', - ls='', - marker=cut_star_marker, - markersize=50, - ) - if core_label_x not in dic_x_label: xlabel_set.append(core_label_x) dic_x_label[core_label_x] = 1 @@ -482,16 +470,36 @@ def tf_combine(self, list_file_names=None, extraParams=None): tf.close() + try: + if global_x_min is not None and global_x_max is not None: + if any_forward_orientation: + plt.xlim(global_x_min, global_x_max) + else: + plt.xlim(global_x_max, global_x_min) + except Exception: + logger.exception('tf_combine: failed to set global x-limits') + + if active_nicp is not None: + try: + y_boundary_min, y_boundary_max = active_nicp.plot_overlays( + plt, active_y_values, adjust_xlim=False, + ) + ymin, ymax = plt.gca().get_ylim() + plt.ylim(min(ymin, y_boundary_min), max(ymax, y_boundary_max)) + except Exception: + logger.exception( + 'tf_combine: failed to draw overlays for active spectrum' + ) + plt.xlabel(xlabel, fontsize=18) plt.ylabel(ylabel, fontsize=18) ax = plt.gca() if cv_mode: ymin, ymax = ax.get_ylim() cv_abs_max = max(abs(ymin), abs(ymax)) - + if cv_mode and cv_abs_max > 0: exp = int(np.floor(np.log10(cv_abs_max))) if cv_abs_max > 0 else 0 - print(f"[tf_combine] cv_abs_max={cv_abs_max}, exp={exp}") base = (10.0 ** exp) if exp != 0 else 1.0 ax.yaxis.set_major_formatter(ticker.FuncFormatter(lambda y, _: f"{(y / base):.3g}" @@ -506,6 +514,7 @@ def tf_combine(self, list_file_names=None, extraParams=None): fontsize=14, clip_on=False ) + plt.grid(False) plt.legend() tf_img = tempfile.NamedTemporaryFile(suffix='.png') plt.savefig(tf_img, format='png') diff --git a/tests/lib/composer/test_ni_composer.py b/tests/lib/composer/test_ni_composer.py new file mode 100644 index 0000000..96a8176 --- /dev/null +++ b/tests/lib/composer/test_ni_composer.py @@ -0,0 +1,94 @@ +import pytest + +from chem_spectra.lib.converter.jcamp.base import JcampBaseConverter +from chem_spectra.lib.converter.jcamp.ni import JcampNIConverter +# NIComposer selects the Agg backend on import, so import pyplot after it +from chem_spectra.lib.composer.ni import NIComposer +import matplotlib.pyplot as plt # noqa: E402 + +source_nmr = './tests/fixtures/source/1H.dx' + + +@pytest.fixture +def ni_composer(): + base_converter = JcampBaseConverter(source_nmr) + ni_converter = JcampNIConverter(base_converter) + return NIComposer(ni_converter) + + +def test_ensure_itg_mpy_hydrates_from_core_tables(ni_composer): + ni_composer.all_itgs = [] + ni_composer.mpys = [] + ni_composer.refShift = 0 + ni_composer.core.itg_table = ['(1.0, 2.0, 3.0)\n(4.0, 5.0, 6.0)\n'] + ni_composer.core.mpy_itg_table = ['(1, 1.0, 2.0, 0.0, 1.0, 0.0, m, 1.0)\n'] + ni_composer.core.mpy_pks_table = ['(1, 1.5, 100.0)\n(1, 1.7, 90.0)\n'] + + ni_composer._ensure_itg_mpy_from_core_tables() + + assert ni_composer.all_itgs == [ + {'xL': 1.0, 'xU': 2.0, 'area': 3.0}, + {'xL': 4.0, 'xU': 5.0, 'area': 6.0}, + ] + assert len(ni_composer.mpys) == 1 + mpy = ni_composer.mpys[0] + assert mpy['mpyType'] == 'm' + assert mpy['xExtent'] == {'xL': 1.0, 'xU': 2.0} + assert mpy['peaks'] == [{'x': 1.5, 'y': 100.0}, {'x': 1.7, 'y': 90.0}] + + +def test_ensure_itg_mpy_skips_malformed_lines(ni_composer): + ni_composer.all_itgs = [] + ni_composer.mpys = [] + ni_composer.core.itg_table = ['\n(1.0)\n(1.0, 2.0, 3.0)\n\n'] + ni_composer.core.mpy_itg_table = ['\n(1, 1.0, 2.0)\n(1, 1.0, 2.0, 0.0, 1.0, 0.0, s, 1.0)\n'] + ni_composer.core.mpy_pks_table = ['\n(1, 1.5, 100.0)\n(2)\n'] + + ni_composer._ensure_itg_mpy_from_core_tables() + + assert ni_composer.all_itgs == [{'xL': 1.0, 'xU': 2.0, 'area': 3.0}] + assert len(ni_composer.mpys) == 1 + assert ni_composer.mpys[0]['peaks'] == [{'x': 1.5, 'y': 100.0}] + + +def test_ensure_itg_mpy_skips_multiplet_without_peaks(ni_composer): + # a multiplet whose index has no matching peaks must be dropped: + # calc_mpy_center divides by the peak count when it is drawn later + ni_composer.all_itgs = [] + ni_composer.mpys = [] + ni_composer.core.itg_table = [] + ni_composer.core.mpy_itg_table = ['(2, 1.0, 2.0, 0.0, 1.0, 0.0, t, 1.0)\n'] + ni_composer.core.mpy_pks_table = ['(1, 1.5, 100.0)\n'] + + ni_composer._ensure_itg_mpy_from_core_tables() + + assert ni_composer.mpys == [] + + +def test_ensure_itg_mpy_handles_missing_peaks_table(ni_composer): + # $OBSERVEDMULTIPLETS present but $OBSERVEDMULTIPLETSPEAKS missing + ni_composer.all_itgs = [] + ni_composer.mpys = [] + ni_composer.core.itg_table = [] + ni_composer.core.mpy_itg_table = ['(1, 1.0, 2.0, 0.0, 1.0, 0.0, m, 1.0)\n'] + ni_composer.core.mpy_pks_table = [] + + ni_composer._ensure_itg_mpy_from_core_tables() + + assert ni_composer.mpys == [] + + +def test_plot_overlays_returns_boundaries(ni_composer): + ni_composer.core.itg_table = ['(1.0, 2.0, 3.0)\n'] + ni_composer.core.mpy_itg_table = ['(1, 1.0, 2.0, 0.0, 1.0, 0.0, m, 1.0)\n'] + ni_composer.core.mpy_pks_table = ['(1, 1.5, 100.0)\n'] + + fig = plt.figure() + try: + y_boundary_min, y_boundary_max = ni_composer.plot_overlays( + plt, adjust_xlim=False, + ) + finally: + plt.close(fig) + + assert y_boundary_min < y_boundary_max diff --git a/tests/model/test_transformer.py b/tests/model/test_transformer.py index 3135c80..d76b1ce 100644 --- a/tests/model/test_transformer.py +++ b/tests/model/test_transformer.py @@ -61,6 +61,25 @@ def test_combine(): assert tf is not None +def test_combine_with_active_spectrum_overlays(): + with open(source_dir_1h_jcamp, 'rb') as f: + file_1 = FileContainer(FileStorage(f)) + + with open(source_dir_1h_jcamp, 'rb') as f: + file_2 = FileContainer(FileStorage(f)) + + # jcamp_idx picks the active spectrum whose overlays are drawn; a string + # value must be tolerated (form params arrive untyped) + params = {**params_1h_jcamp, 'jcamp_idx': '1'} + with open(source_dir_molfile, "r") as molfile: + tranform_model = TransformerModel(None, molfile=molfile, params=params, multiple_files=[file_1, file_2]) + + tf = tranform_model.tf_combine(list_file_names=['a.dx', 'b.dx']) + assert tf is not None + tf.seek(0) + assert tf.read(8) == b'\x89PNG\r\n\x1a\n' + + def test_zip2cv_with_processed_file(): with open(source_dir_molfile, 'rb') as f: molfile = FileContainer(FileStorage(f))