Detect negative axis limits from data type

src/python_spectrometer/_plot_manager.py

@@ -7,8 +7,9 @@ from itertools import compress
 from typing import (Dict, Any, Optional, Mapping, Tuple, ContextManager, Literal, Iterable,
                     Union, List)
 
+import matplotlib.pyplot as plt
 import numpy as np
-from matplotlib import pyplot as plt, gridspec
+from matplotlib import gridspec, scale
 from qutil.misc import filter_warnings
 from qutil.plotting import assert_interactive_figure
 from qutil.typecheck import check_literals
@@ -31,9 +32,9 @@ class PlotManager:
                  plot_cumulative_normalized: bool = False, plot_style: _styleT = 'fast',
                  plot_update_mode: str = 'never', plot_dB_scale: bool = False, prop_cycle=None,
                  raw_unit: str = 'V', processed_unit: Optional[str] = None,
-                 uses_windowed_estimator: bool = True, figure_kw: Optional[Mapping] = None,
-                 subplot_kw: Optional[Mapping] = None, gridspec_kw: Optional[Mapping] = None,
-                 legend_kw: Optional[Mapping] = None):
+                 uses_windowed_estimator: bool = True, complex_data: bool = False,
+                 figure_kw: Optional[Mapping] = None, subplot_kw: Optional[Mapping] = None,
+                 gridspec_kw: Optional[Mapping] = None, legend_kw: Optional[Mapping] = None):
         """A helper class that manages plotting spectrometer data."""
         self._data = data
 
@@ -57,6 +58,7 @@ class PlotManager:
         self.prop_cycle = prop_cycle or plt.rcParams['axes.prop_cycle']
         self.raw_unit = raw_unit
         self.uses_windowed_estimator = uses_windowed_estimator
+        self.complex_data = complex_data
 
         self._leg = None
         self.axes = {key: dict.fromkeys(self.LINE_TYPES) for key in self.PLOT_TYPES}
@@ -617,22 +619,17 @@ class PlotManager:
 
     def set_xscales(self):
         if (
-                self.plot_negative_frequencies
-                and any(d['f_processed'][0] < 0 for d in self._data.values() if 'f_processed' in d)
+                # If daq returns complex data, the spectrum will have negative freqs
+                self.complex_data
+                and self.plot_negative_frequencies
                 or self.plot_raw
-                and any(d['f_raw'][0] < 0 for d in self._data.values() if 'f_raw' in d)
                 and self.axes['main']['processed'].get_xscale() == 'log'
         ):
             if self.axes['main']['processed'].get_xscale() == 'log':
                 # matplotlib>=3.6 has asinh scale for log plots with negative values
-                try:
-                    self.axes['main']['processed'].set_xscale('asinh')
-                    if self.plot_cumulative:
-                        self.axes['cumulative']['processed'].set_xscale('asinh')
-                except ValueError:
-                    self.axes['main']['processed'].set_xscale('linear')
+                self.axes['main']['processed'].set_xscale(_asinh_scale_maybe())
                 if self.plot_cumulative:
-                        self.axes['cumulative']['processed'].set_xscale('linear')
+                    self.axes['cumulative']['processed'].set_xscale(_asinh_scale_maybe())
         else:
             if self.axes['main']['processed'].get_xscale() != 'log':
                 self.axes['main']['processed'].set_xscale('log')
@@ -739,3 +736,12 @@ def _ax_label(amplitude: bool, integrated: bool, dB: bool, reference: _keyT) ->
         b='amplitude' if amplitude else 'power',
         c=reference[0]
     ).capitalize()
+
+
+def _asinh_scale_maybe() -> Literal['asinh', 'linear']:
+    try:
+        getattr(scale, 'AsinhScale')
+    except AttributeError:
+        return 'linear'
+    else:
+        return 'asinh'

src/python_spectrometer/core.py

@@ -28,7 +28,7 @@ from qutil.typecheck import check_literals
 from qutil.ui import progressbar
 
 from ._audio_manager import WaveformPlaybackManager
-from ._plot_manager import PlotManager
+from ._plot_manager import PlotManager, _asinh_scale_maybe
 from .daq import settings as daq_settings
 from .daq.base import DAQ
 
@@ -412,6 +412,7 @@ class Spectrometer:
         else:
             raise TypeError('psd_estimator should be callable or kwarg dict for welch().')
         uses_windowed_estimator = 'window' in inspect.signature(self.psd_estimator).parameters
+        complex_data = np.issubdtype(self.daq.DTYPE, np.complexfloating)
 
         self._plot_manager = PlotManager(self._data, plot_raw, plot_timetrace,
                                          plot_cumulative, plot_negative_frequencies,
@@ -419,8 +420,8 @@ class Spectrometer:
                                          plot_density, plot_cumulative_normalized,
                                          plot_style, plot_update_mode, plot_dB_scale,
                                          prop_cycle, raw_unit, processed_unit,
-                                         uses_windowed_estimator, figure_kw, subplot_kw,
-                                         gridspec_kw, legend_kw)
+                                         uses_windowed_estimator, complex_data, figure_kw,
+                                         subplot_kw, gridspec_kw, legend_kw)
 
         self._audio_amplitude_normalization = audio_amplitude_normalization
         self._play_sound = play_sound
@@ -1081,15 +1082,25 @@ class Spectrometer:
         settings = self.daq.setup(**settings)
         T = settings['n_pts'] / settings['fs']
 
+        if np.issubdtype(self.daq.DTYPE, np.complexfloating):
+            xscale = _asinh_scale_maybe()
+            xlim = np.array([-settings['f_max'], settings['f_max']])
+        else:
+            xscale = 'log'
+            xlim = np.array([settings['f_min'], settings['f_max']])
+        if self.plot_absolute_frequencies:
+            xlim += settings.get('freq', 0)
+
         if live_view_kw is None:
             live_view_kw = {}
 
         live_view_kw.setdefault('autoscale', 'c')
         live_view_kw.setdefault('autoscale_interval', None)
+        live_view_kw.setdefault('xscale', xscale)
 
         fixed_kw = dict(
-            plot_line=True, xscale='log', xlim=(settings['f_min'], settings['f_max']),
-            ylim=(0, (max_rows - 1) * T), xlabel='$f$', ylabel='$t$', clabel='$S(f)$',
+            plot_line=True, xlim=xlim, ylim=(0, (max_rows - 1) * T),
+            xlabel='$f$', ylabel='$t$', clabel='$S(f)$',
             units={'x': 'Hz', 'y': 's', 'c': self.processed_unit + r'$/\sqrt{{Hz}}$'},
             img_kw=dict(norm=colors.LogNorm(vmin=0.1, vmax=10), cmap='Blues')
         )

src/python_spectrometer/daq/base.py

@@ -5,6 +5,7 @@ import warnings
 from abc import ABC
 from typing import Any, Dict, Iterator, Type
 
+import numpy as np
 from qutil.functools import cached_property
 from qutil.misc import filter_warnings
 
@@ -29,6 +30,8 @@ class DAQ(ABC):
     implement hardware constraints by subclassing the
     :class:`.DAQSettings` class.
     """
+    DTYPE = np.floating
+    """The data type yielded by :meth:`acquire`."""
 
     def __post_init__(self):
         """Run import checks here and other setup steps here."""
@@ -73,7 +76,7 @@ class DAQ(ABC):
         return self.DAQSettings(**settings).to_consistent_dict()
 
     @abc.abstractmethod
-    def acquire(self, *, n_avg: int, **settings) -> Iterator[NDArray]:
+    def acquire(self, *, n_avg: int, **settings) -> Iterator[NDArray[DTYPE]]:
         """Returns an iterator that yields data n_avg times.
 
         This method should execute the measurement and yield a

src/python_spectrometer/daq/zurich_instruments.py

@@ -143,6 +143,8 @@ class ZurichInstrumentsMFLIDAQ(_ZurichInstrumentsDevice):
         directly from the device's ADC (before being demodulated).
 
     """
+    DTYPE = np.complexfloating
+
     demod: int = 0
     """The demodulator to use. The default is 0."""
     osc: int = 0