MiPi_TEST/mipi_test.py

#!/usr/bin/env python3
"""
MIPI TEST APPLICATION - MIPI_TEST.PY
- ENTRY POINT OF APPLICATION

VERSION: 0.2
AUTHOR: D. RICE 25/03/2026
© 2026 ARRIVE
"""
import vxi11
import time
import sys
import requests
import threading

# --- Configuration ---
URL        = "http://192.168.45.8:5000/display"
SCOPE_IP   = "192.168.45.4"
PSU_IP     = "192.168.45.3"

test_running = False  # Global flag to control the background thread

# --- Instrument Connection ---
try:
    psu   = vxi11.Instrument(PSU_IP)
    scope = vxi11.Instrument(SCOPE_IP)
    scope.timeout = 30   # seconds — needs to cover *RST and image saves
    psu.timeout   = 5
except Exception as e:
    print(f"ERROR: CANNOT CONNECT TO INSTRUMENTS: {e}")
    sys.exit(1)

# ---------------------------------------------------------------------------

def setup_scope():
    """Initialises DSO80204B for MIPI DSI signals (~210 MHz)."""
    print("CONFIGURING SCOPE...")

    cmds = [
        # ── Reset & stop ──────────────────────────────────────────────────
        "*RST",
        ":RUN",
        ":STOP",

        # ── Channel 1 — Clock D+ ─────────────────────────────────────────
        ":CHANnel1:DISPlay ON",
        ":CHANnel1:INPut DC50",
        ":CHANnel1:PROBe 19.2",          # 910Ω + 50Ω divider = 19.2:1
        ":CHANnel1:SCALe 0.1",           # 100 mV/div
        ":CHANnel1:OFFSet 0.0",
        ":CHANnel1:LABel 'CLK+'",

        # ── Channel 2 — Clock D- ─────────────────────────────────────────
        ":CHANnel2:DISPlay ON",
        ":CHANnel2:INPut DC50",
        ":CHANnel2:PROBe 19.2",
        ":CHANnel2:SCALe 0.1",
        ":CHANnel2:OFFSet 0.0",
        ":CHANnel2:LABel 'CLK-'",

        # ── Channel 3 — Data Lane 0 D+ ───────────────────────────────────
        ":CHANnel3:DISPlay ON",
        ":CHANnel3:INPut DC50",
        ":CHANnel3:PROBe 19.2",
        ":CHANnel3:SCALe 0.1",
        ":CHANnel3:OFFSet 0.0",
        ":CHANnel3:LABel 'DAT0+'",

        # ── Channel 4 — Data Lane 0 D- ───────────────────────────────────
        ":CHANnel4:DISPlay ON",
        ":CHANnel4:INPut DC50",
        ":CHANnel4:PROBe 19.2",
        ":CHANnel4:SCALe 0.1",
        ":CHANnel4:OFFSet 0.0",
        ":CHANnel4:LABel 'DAT0-'",

        # ── Timebase — 5 ns/div shows ~10 cycles of 200 MHz clock ────────
        ":TIMebase:SCALe 5E-9",
        ":TIMebase:POSition 0",
        ":TIMebase:REFerence CENTer",

        # ── Trigger — rising edge on Ch1 (Clock D+) ──────────────────────
        ":TRIGger:MODE EDGE",
        ":TRIGger:EDGE:SOURce CHANnel1",
        ":TRIGger:EDGE:SLOPe POSitive",
        ":TRIGger:EDGE:LEVel 0.05",      # 50 mV post-attenuation
        ":TRIGger:SWEep NORMal",

        # ── Acquisition ───────────────────────────────────────────────────
        ":ACQuire:MODE RTIMe",
        ":ACQuire:INTerpolate ON",
        ":ACQuire:POINts 500000",

        # ── Display ───────────────────────────────────────────────────────
        ":DISPlay:LAYout STACKED",

        ":RUN",
    ]

    for cmd in cmds:
        scope.write(cmd)
        time.sleep(0.05)

    print("CHANNEL SETUP COMPLETE.")
    setup_math_channels()


def setup_math_channels():
    """
    F1 = Ch1 - Ch2  (clock differential)
    F2 = Ch3 - Ch4  (lane 0 differential)

    DSO80204B firmware 05.30.0005 confirmed syntax:
      :FUNCtion<n>:DISPlay ON
      :FUNCtion<n>:SUBTract CHANnel<a>,CHANnel<b>
    """
    print("SETTING UP MATH CHANNELS...")

    scope.write("*CLS")
    time.sleep(0.2)

    math_cmds = [
        # ── F1 = Ch1 - Ch2 (clock differential) ─────────────────────────
        ":FUNCtion1:DISPlay ON",
        ":FUNCtion1:SUBTract CHANnel1,CHANnel2",
        ":FUNCtion1:RANGe 0.8",          # 0.8V range = 100mV/div (range = 8 × scale)
        ":FUNCtion1:OFFSet 0.0",

        # ── F2 = Ch3 - Ch4 (lane 0 differential) ─────────────────────────
        ":FUNCtion2:DISPlay ON",
        ":FUNCtion2:SUBTract CHANnel3,CHANnel4",
        ":FUNCtion2:RANGe 0.8",          # 0.8V range = 100mV/div
        ":FUNCtion2:OFFSet 0.0",
    ]

    for cmd in math_cmds:
        scope.write(cmd)
        time.sleep(0.2)

    try:
        time.sleep(1.0)
        opc = scope.ask("*OPC?")
        print(f"  SCOPE SYNC OK (OPC={opc.strip()})")
    except Exception as e:
        print(f"  WARNING: OPC SYNC FAILED ({e})")

    try:
        err = scope.ask(":SYSTem:ERRor?")
        if err.strip().startswith("0"):
            print("  MATH COMMANDS ACCEPTED — NO SCPI ERRORS.")
            print("  F1 = CLK DIFF (CH1-CH2),  F2 = DAT DIFF (CH3-CH4)")
        else:
            print(f"  SCPI ERROR: {err.strip()}")
    except Exception as e:
        print(f"  COULD NOT READ ERROR QUEUE ({e})")


def save_screenshot(iteration):
    """Save a PNG screenshot to C:\\TEMP on the scope's local disk."""
    try:
        filename = f"C:\\TEMP\\cap{iteration:04d}.png"
        print(f"SAVING SCREENSHOT: {filename}")
        scope.write(f':DISK:SAVE:IMAGe "{filename}",PNG')
        time.sleep(5.0)   # wait for scope to write file — do not query during this
        print("SCREENSHOT SAVED.")
    except Exception as e:
        print(f"SCREENSHOT ERROR: {e}")


def test_worker(on_time):
    """
    Background loop:
      - Arms scope in single trigger mode
      - Turns display ON for on_time seconds
      - Saves screenshot
      - Turns display OFF for 1 second
      - Repeats until test_running = False
    """
    global test_running
    count = 1

    while test_running:
        scope.write(":SINGle")
        requests.put(URL, json={"state": "on"}, timeout=2)
        time.sleep(on_time + 0.5)   # +0.5s to ensure acquisition completes
        save_screenshot(count)
        count += 1
        requests.put(URL, json={"state": "off"}, timeout=2)
        time.sleep(1.0)


def main_menu():
    global test_running

    while True:
        print("\n===== MIPI TEST CONTROL =====")
        print("1. RUN IDN CHECK (PSU & SCOPE)")
        print("2. SETUP SCOPE (RUN FIRST)")
        print("3. CONFIGURE PSU (DEFAULT 24V / 1.5A)")
        print("4. PSU OUTPUT ON/OFF (CH1)")
        print("5. START TEST & CAPTURE")
        print("6. STOP TEST")
        print("7. EXIT")

        choice = input("\nSELECT OPTION (1-7): ").strip()

        if choice == '1':
            print(f"PSU  : {psu.ask('*IDN?').strip()}")
            print(f"SCOPE: {scope.ask('*IDN?').strip()}")

        elif choice == '2':
            setup_scope()

        elif choice == '3':
            psu.write('CH1:VOLT 24.0')
            psu.write('CH1:CURR 1.5')
            print("PSU CONFIGURED: 24V / 1.5A")

        elif choice == '4':
            state = input("TYPE 'ON' OR 'OFF': ").strip().upper()
            if state in ('ON', 'OFF'):
                psu.write(f'OUTP CH1,{state}')
                print(f"PSU OUTPUT {state}.")
            else:
                print("INVALID — TYPE 'ON' OR 'OFF'.")

        elif choice == '5':
            if not test_running:
                try:
                    sec = float(input("ON DURATION (seconds): "))
                    test_running = True
                    t = threading.Thread(target=test_worker, args=(sec,), daemon=True)
                    t.start()
                    print(f"TEST STARTED — ON TIME: {sec}s")
                except ValueError:
                    print("INVALID TIME ENTERED.")
            else:
                print("TEST IS ALREADY RUNNING!")

        elif choice == '6':
            print("STOPPING TEST...")
            test_running = False

        elif choice == '7':
            test_running = False
            psu.close()
            scope.close()
            print("INSTRUMENTS CLOSED. BYE.")
            break

        else:
            print("INVALID ENTRY. PLEASE CHOOSE 1-7.")


if __name__ == "__main__":
    main_menu()