Structuring a Data-Acquisition App

import os
import time


class CSVFile:
    """Creates (or appends to) a CSV file and writes comment lines and rows."""

    def __init__(self, path: str, name: str):
        os.makedirs(path, exist_ok=True)
        if not name.endswith(".csv"):
            name += ".csv"
        self.filename = os.path.join(path, name)
        self.is_new = not os.path.exists(self.filename)
        if self.is_new:
            self.write_comment(f"Created {time.ctime()}")

    def write_comment(self, text: str):
        """Write a metadata line, prefixed with # so analysis code can skip it."""
        with open(self.filename, "a") as f:
            f.write(f"# {text}\n")

    def write_row(self, values: list):
        """Append one comma-delimited row."""
        with open(self.filename, "a") as f:
            f.write(",".join(str(v) for v in values) + "\n")

from instruments import TemperatureController, VoltageSupply   # your wrappers


class ResistanceVsTemp(CSVFile):
    """Example: log resistance against temperature until told to stop."""

    columns = ["time_s", "temperature_K", "resistance_ohm"]

    def __init__(self, path, name, tc_resource, vs_resource):
        super().__init__(path, name)
        self.tc = TemperatureController(tc_resource)
        self.vs = VoltageSupply(vs_resource)
        self.running = False
        if self.is_new:
            self.write_row(self.columns)          # header row for a fresh file

    def take_point(self) -> list:
        """Read every instrument once and return a row."""
        temperature = self.tc.temperature("A")
        resistance = self.vs.measure_resistance()
        return [time.time(), temperature, resistance]

    def run(self, interval_s: float = 1.0):
        """Acquire continuously until self.running is set False."""
        self.running = True
        while self.running:
            self.write_row(self.take_point())
            time.sleep(interval_s)

import numpy as np


class MockDiode:
    """Stand-in for a real instrument: same methods, simulated readings.

    Returns current through an ideal diode (the Shockley equation) so the
    curve has the shape the real measurement will have.
    """

    I0 = 1e-12          # reverse saturation current, A
    N = 1.8             # ideality factor
    VT = 0.02585        # thermal voltage at room temperature, V

    def __init__(self):
        self._voltage = 0.0
        self._rng = np.random.default_rng()

    def set_voltage(self, volts: float):
        self._voltage = volts

    def measure_current(self) -> float:
        current = self.I0 * (np.exp(self._voltage / (self.N * self.VT)) - 1)
        return float(current + self._rng.normal(0, max(abs(current), 1e-12) * 0.01))

class DiodeSweep(CSVFile):
    columns = ["voltage_V", "current_A"]

    def __init__(self, path, name, resource, offline=False):
        super().__init__(path, name)
        self.smu = MockDiode() if offline else SourceMeasureUnit(resource)
        self.running = False
        if self.is_new:
            self.write_row(self.columns)

    def take_point(self, volts):
        self.smu.set_voltage(volts)
        return [volts, self.smu.measure_current()]


# At a desk, with no instrument attached:
sweep = DiodeSweep("data", "test", resource=None, offline=True)

import threading

experiment = ResistanceVsTemp("data", "run1", "GPIB0::12::INSTR", "GPIB0::4::INSTR")

worker = threading.Thread(target=experiment.run, daemon=True)
worker.start()        # acquisition now runs in the background

# ... main program keeps going; later, to stop cleanly:
experiment.running = False
worker.join()         # wait for the loop to finish its current pass

import sys

def main():
    # Name the file from a command-line argument, or default to a timestamp.
    name = sys.argv[1] if len(sys.argv) > 1 else f"run_{int(time.time())}"
    experiment = ResistanceVsTemp("data", name, "GPIB0::12::INSTR", "GPIB0::4::INSTR")

    worker = threading.Thread(target=experiment.run, daemon=True)
    worker.start()
    print("Acquiring. Commands: 'setpoint <K>', 'stop'.")

    while True:
        command = input("> ").strip().lower()
        if command == "stop":
            experiment.running = False
            worker.join()
            break
        elif command.startswith("setpoint"):
            experiment.tc.set_setpoint(float(command.split()[1]))
        else:
            print("Unknown command.")


if __name__ == "__main__":
    main()