Scientific Data Analysis with LabPlot in Python: Signal Processing, Spectral Peak Fitting, Visualization, and Batch Automation

THEMES = {
"BlackOnWhite": dict(bg="#ffffff", fg="#000000", grid="#c8c8c8",
  cycle=["#3465a4", "#cc0000", "#4e9a06", "#f57900", "#75507b", "#06989a"]),
"Dracula": dict(bg="#282a36", fg="#f8f8f2", grid="#44475a",
  cycle=["#8be9fd", "#ff79c6", "#50fa7b", "#ffb86c", "#bd93f9", "#f1fa8c"]),
"SolarizedDark": dict(bg="#002b36", fg="#93a1a1", grid="#0f4b57",
  cycle=["#268bd2", "#dc322f", "#859900", "#b58900", "#6c71c4", "#2aa198"])}
class XYCurve(AbstractAspect):
   def __init__(self, name, x=None, y=None, lineStyle="-", lineWidth=1.6,
                symbolStyle=None, symbolSize=4., color=None, alpha=1., zorder=2):
       super().__init__(name)
       self.xColumn, self.yColumn, self.color, self.alpha = x, y, color, alpha
       self.lineStyle, self.lineWidth = lineStyle, lineWidth
       self.symbolStyle, self.symbolSize, self.zorder = symbolStyle, symbolSize, zorder
       self.yErrorColumn = self.fillBetween = None
   def setXColumn(self, c): self.xColumn = c; return self
   def setYColumn(self, c): self.yColumn = c; return self
   @staticmethod
   def _v(c): return c.values() if isinstance(c, Column) else np.asarray(c, float)
   def draw(self, ax, color):
       c = self.color or color; X, Y = self._v(self.xColumn), self._v(self.yColumn)
       if self.fillBetween is not None:
           ax.fill_between(X, *self.fillBetween, color=c, alpha=.2, lw=0, zorder=self.zorder-1)
       if self.yErrorColumn is not None:
           ax.errorbar(X, Y, yerr=self._v(self.yErrorColumn), fmt="none", ecolor=c,
                       elinewidth=.8, capsize=2, alpha=.7, zorder=self.zorder)
       ax.plot(X, Y, linestyle=self.lineStyle or "none", marker=self.symbolStyle or "none",
               markersize=self.symbolSize, linewidth=self.lineWidth, color=c, alpha=self.alpha,
               label=self._name, zorder=self.zorder, markeredgewidth=0)
class Histogram(AbstractAspect):
   """normalization: 'Count' | 'Probability' | 'CountDensity' | 'ProbabilityDensity'."""
   def __init__(self, name, dataColumn=None, bins="auto", normalization="ProbabilityDensity"):
       super().__init__(name)
       self.dataColumn, self.bins, self.normalization = dataColumn, bins, normalization
   def draw(self, ax, color):
       d = (self.dataColumn.clean() if isinstance(self.dataColumn, Column)
            else np.asarray(self.dataColumn, float))
       ax.hist(d, bins=self.bins, color=color, alpha=.55, edgecolor=color, lw=.8,
               label=self._name, zorder=1, density="Density" in self.normalization
               or self.normalization == "Probability")
class CartesianPlot(AbstractAspect):
   class Type(Enum):
       FourAxes = 0; TwoAxes = 1
   def __init__(self, name, title=None, xLabel="x", yLabel="y", logX=False, logY=False):
       super().__init__(name); self.type = CartesianPlot.Type.FourAxes
       self.title, self.xLabel, self.yLabel = title or name, xLabel, yLabel
       self.logX, self.logY, self.legend = logX, logY, None
       self.xRange, self.yRange, self.labels = None, None, []
   def setType(self, t): self.type = t; return self
   def addLegend(self, loc="best"): self.legend = loc; return self
   def setRange(self, x=None, y=None): self.xRange, self.yRange = x, y; return self
   def addTextLabel(self, txt, x, y): self.labels.append((txt, x, y)); return self
   def _render(self, ax, th):
       ax.set_facecolor(th["bg"])
       for i, ch in enumerate(self.children): ch.draw(ax, th["cycle"][i % len(th["cycle"])])
       ax.set_title(self.title, color=th["fg"], fontsize=10.5, pad=7)
       ax.set_xlabel(self.xLabel, color=th["fg"], fontsize=9.5)
       ax.set_ylabel(self.yLabel, color=th["fg"], fontsize=9.5)
       for lg, sc, axis in ((self.logX, ax.set_xscale, ax.xaxis), (self.logY, ax.set_yscale, ax.yaxis)):
           sc("log") if lg else axis.set_minor_locator(AutoMinorLocator(2))
       if self.xRange: ax.set_xlim(*self.xRange)
       if self.yRange: ax.set_ylim(*self.yRange)
       four = self.type is CartesianPlot.Type.FourAxes
       for s in ("top", "right"): ax.spines[s].set_visible(four)
       for s in ax.spines.values(): s.set_color(th["fg"]); s.set_linewidth(.9)
       ax.tick_params(which="both", direction="in", colors=th["fg"], top=four,
                      right=four, labelsize=8.5)
       ax.grid(True, color=th["grid"], lw=.6, alpha=.7, zorder=0)
       for t, x, y in self.labels:
           ax.annotate(t, (x, y), color=th["fg"], fontsize=7.5, ha="center")
       if self.legend:
           for t in ax.legend(loc=self.legend, fontsize=8, framealpha=.85, facecolor=th["bg"],
                              edgecolor=th["grid"]).get_texts(): t.set_color(th["fg"])
class Worksheet(AbstractAspect):
   class ExportFormat(Enum):
       PDF = 0; SVG = 1; PNG = 2
   def __init__(self, name, cols=None, figsize=(15, 8.5), dpi=110):
       super().__init__(name); self.themeName = "BlackOnWhite"
       self.cols, self.figsize, self.dpi, self._fig = cols, figsize, dpi, None
   def setTheme(self, n):
       if n not in THEMES: raise KeyError(f"themes: {list(THEMES)}")
       self.themeName = n; return self
   def render(self):
       th = THEMES[self.themeName]
       ps = [c for c in self.children if isinstance(c, CartesianPlot)]
       cols = self.cols or min(len(ps), 2)
       fig, axes = plt.subplots(math.ceil(len(ps)/cols), cols, figsize=self.figsize, dpi=self.dpi)
       fig.patch.set_facecolor(th["bg"]); axes = np.atleast_1d(axes).ravel()
       for ax, p in zip(axes, ps): p._render(ax, th)
       for ax in axes[len(ps):]: ax.axis("off")
       fig.suptitle(self._name, color=th["fg"], fontsize=13, y=.995)
       fig.tight_layout(rect=(0, 0, 1, .98)); self._fig = fig; return fig
   def show(self):
       (self.render() if self._fig is None else None); plt.show()
   def exportToFile(self, path, format=None):
       if self._fig is None: self.render()
       fmt = (format.name.lower() if isinstance(format, Worksheet.ExportFormat)
              else format or os.path.splitext(path)[1].lstrip("."))
       self._fig.savefig(path, format=fmt, dpi=self.dpi, bbox_inches="tight",
                         facecolor=self._fig.get_facecolor()); return path
def _reduce(x, y, tolerance=None):
   i = nsl_geom.douglas_peucker(x, y, tolerance if tolerance is not None else .02*np.ptp(y))
   return x[i], y[i], {"in": len(x), "out": len(i), "compression": 1 - len(i)/len(x)}
class XYAnalysisCurve(XYCurve):
   OPS = {
    "smooth": lambda x, y, points=11, order=3:
       (x, nsl_smooth.savitzky_golay(y, points, order), {}),
    "differentiate": lambda x, y, derivOrder=1, smoothPoints=0:
       (x, nsl_diff.derive(x, y, derivOrder, smoothPoints), {}),
    "integrate": lambda x, y, method="trapezoid", absolute=False:
       (lambda c: (x, c, {"total": float(c[-1])}))(nsl_int.integrate(x, y, method, absolute)),
    "dft": lambda x, y, output="amplitude", window="rectangular":
       nsl_dft.transform(x, y, output, window) + ({},),
    "filter": lambda x, y, type="lowpass", form="butterworth", cutoff=.1, cutoff2=.3, order=3:
       (x, nsl_filter.apply(x, y, type, form, cutoff, cutoff2, order), {}),
    "hilbert": lambda x, y, output="envelope": (x, nsl_hilbert.transform(y, output), {}),
    "reduce": _reduce}
   def __init__(self, name, xData, yData, op, style=None, **opts):
       super().__init__(name, **(style or {}))
       self._xin, self._yin = XYCurve._v(xData), XYCurve._v(yData)
       self.op, self.opts, self.result = op, opts, None
       self.recalculate()
   def recalculate(self):
       self.xColumn, self.yColumn, self.result = 
           XYAnalysisCurve.OPS[self.op](self._xin, self._yin, **self.opts)
       return self
_mk = lambda op: (lambda name, x, y, style=None, **kw: XYAnalysisCurve(name, x, y, op, style, **kw))
XYSmoothCurve, XYDifferentiationCurve = _mk("smooth"), _mk("differentiate")
XYIntegrationCurve = _mk("integrate")
XYFourierTransformCurve, XYFourierFilterCurve = _mk("dft"), _mk("filter")
XYHilbertTransformCurve, XYDataReductionCurve = _mk("hilbert"), _mk("reduce")
class XYFitCurve(XYCurve):
   """LabPlot's centrepiece: non-linear fitting with the full statistics table."""
   def __init__(self, name, xData, yData, model, p0, paramNames=None, yerr=None,
                bounds=None, npoints=800, **kw):
       super().__init__(name, **kw)
       self._xin, self._yin = XYCurve._v(xData), XYCurve._v(yData)
       self.model, self.p0, self.paramNames = model, p0, paramNames
       self.yerr, self.bounds, self.npoints, self.fitResult = yerr, bounds, npoints, None
   def recalculate(self, conf=.95, showConfidenceInterval=True):
       self.fitResult = nsl_fit.fit(self.model, self._xin, self._yin, self.p0,
                                    self.yerr, self.bounds, self.paramNames, conf)
       xf = np.linspace(self._xin.min(), self._xin.max(), self.npoints)
       yf = self.model(xf, *self.fitResult.values); self.xColumn, self.yColumn = xf, yf
       if showConfidenceInterval:
           d = nsl_fit.confidenceBand(self.model, xf, self.fitResult, conf)
           self.fillBetween = (yf - d, yf + d)
       return self
class ProjectFile:
   MAGIC = ((b"x1fx8b", gzip.decompress, "gzip"), (b"BZh", bz2.decompress, "bzip2"),
            (b"xfd7zXZx00", lzma.decompress, "xz"))
   @staticmethod
   def load(path):
       blob = open(path, "rb").read(); kind = "plain"
       for magic, dec, nm in ProjectFile.MAGIC:
           if blob.startswith(magic): blob, kind = dec(blob), nm; break
       root = ET.fromstring(blob.decode("utf-8", "replace"))
       root = root if root.tag == "project" else root.find(".//project")
       if root is None: raise ValueError("no project element found")
       prj = Project(os.path.basename(path), root.get("author", ""))
       prj.version = root.get("version", "?")
       print(f"  loaded .lml: compression={kind} version={prj.version} xmlVersion="
             f"{root.get('xmlVersion','?')}")
       parents = {c: p for p in root.iter() for c in p}
       def sheet_of(n):
           n = parents.get(n)
           while n is not None and n.tag != "spreadsheet": n = parents.get(n)
           return n
       buckets = {}
       for col in root.iter("column"):
           buckets.setdefault(id(sheet_of(col)), (sheet_of(col), []))[1].append(col)
       for el, cols in buckets.values():
           sp = Spreadsheet(el.get("name", "spreadsheet") if el is not None else "sheet")
           for c in cols: sp.addChild(ProjectFile._column(c))
           prj.addChild(sp)
       return prj
   @staticmethod
   def _column(el):
       name = el.get("name") or next(
           (el.find(t).get("name") for t in ("general", "comment")
            if el.find(t) is not None and el.find(t).get("name")), "Column")
       rows = el.findall("row")
       if rows:
           raw = [r.text for r in sorted(rows, key=lambda r: int(r.get("index", 0)))]
       else:
           node = next((el.find(t) for t in ("values", "data", "double")
                        if el.find(t) is not None and el.find(t).text), None)
           raw = (node.text if node is not None else el.text or "").split()
       vals = []
       for v in raw:
           try: vals.append(float(v))
           except (TypeError, ValueError): vals.append(np.nan)
       try: des = PlotDesignation(int(el.get("designation", 0)))
       except (ValueError, TypeError): des = PlotDesignation.NoDesignation
       return Column(name, vals, designation=des)
   @staticmethod
   def save(project, path, compression="gzip"):
       root = ET.Element("project", {
           "version": project.version, "xmlVersion": str(Project.XML_VERSION),
           "fileName": os.path.basename(path), "author": project.author,
           "modificationTime": time.strftime("%Y-%m-%d %H:%M:%S")})
       ET.SubElement(root, "comment").text = project.comment
       for sp in project.spreadsheets():
           e = ET.SubElement(root, "spreadsheet", {"name": sp.name()})
           ET.SubElement(e, "general", {"rowCount": str(sp.rowCount()),
                                        "columnCount": str(sp.columnCount())})
           for col in sp.columns():
               c = ET.SubElement(e, "column", {
                   "name": col.name(), "rows": str(col.rowCount()),
                   "designation": str(col.plotDesignation.value), "mode": str(col.columnMode.value)})
               for i, v in enumerate(col.values()):
                   ET.SubElement(c, "row", {"index": str(i)}).text = repr(float(v))
       xml = (b'nn'
              + ET.tostring(root, encoding="utf-8"))
       open(path, "wb").write({"gzip": gzip.compress, "bzip2": bz2.compress,
                               "xz": lzma.compress, "none": lambda b: b}[compression](xml))
       return path