Chnages
This commit is contained in:
137
mipi_test.py
137
mipi_test.py
@@ -12,78 +12,131 @@ import time
|
|||||||
import sys
|
import sys
|
||||||
import requests
|
import requests
|
||||||
import threading
|
import threading
|
||||||
|
import os
|
||||||
|
|
||||||
|
# --- Configuration ---
|
||||||
|
URL = "http://192.168.45.8:5000/display"
|
||||||
|
SCOPE_IP = "192.168.45.4"
|
||||||
|
PSU_IP = "192.168.45.3"
|
||||||
|
|
||||||
url = f"http://192.168.45.8:5000/display"
|
|
||||||
test_running = False # Global flag to control the background thread
|
test_running = False # Global flag to control the background thread
|
||||||
|
|
||||||
# --- Hardware Initialisation ---
|
# --- Instrument Connection ---
|
||||||
try:
|
try:
|
||||||
psu = vxi11.Instrument("192.168.45.3")
|
psu = vxi11.Instrument("192.168.45.3")
|
||||||
scope = vxi11.Instrument("192.168.45.4")
|
scope = vxi11.Instrument("192.168.45.4")
|
||||||
setup_del = 0.2
|
# Increase timeout for large image data transfers
|
||||||
|
scope.timeout = 5
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
print(f"ERROR: CANNOT CONNECT TO INSTRUMENTS: {e}")
|
print(f"ERROR: CANNOT CONNECT TO INSTRUMENTS: {e}")
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
|
|
||||||
def set_display(state):
|
def setup_scope():
|
||||||
""" Helper to handle HTTP PUT with error catching """
|
"""Initializes DSO80204B for 210MHz MIPI Signals"""
|
||||||
try:
|
print("Configuring Scope...")
|
||||||
r = requests.put(url, json={"state": state}, timeout=2)
|
scope.write("*RST")
|
||||||
return r.status_code == 200
|
time.sleep(1)
|
||||||
|
|
||||||
except requests.exceptions.RequestException:
|
# Setup Channels (1&2=Clock, 3&4=Data)
|
||||||
return False
|
for ch in range(1, 5):
|
||||||
|
scope.write(f":CHANnel{ch}:DISPlay ON")
|
||||||
def test_worker(interval):
|
scope.write(f":CHANnel{ch}:PROBe 19.2") # 910R/50R Divider
|
||||||
""" Background loop that toggles the display """
|
scope.write(f":CHANnel{ch}:INPut DC50")
|
||||||
global test_running
|
scope.write(f":CHANnel{ch}:SCALe 0.1") # 100mV/div (200mV HS swing)
|
||||||
print(f"\n[TEST STARTED] TOGGLING EVERY {interval} SECONDS. USE MENU TO STOP")
|
scope.write(f":CHANnel{ch}:OFFSet 0.0")
|
||||||
|
|
||||||
state = "on"
|
# Timebase & Trigger
|
||||||
while test_running:
|
scope.write(":TIMebase:SCALe 10e-6") # 10us/div
|
||||||
success = set_display(state)
|
scope.write(":TRIGger:EDGE:SOURce CHANnel3")
|
||||||
if not success:
|
scope.write(":TRIGger:EDGE:SLOPe FALLing") # Catch LP-11 to HS transition
|
||||||
print(f"\n[TEST ERROR] FAILED TO REACH FLASK SERVER... STOPPING TEST...")
|
scope.write(":TRIGger:LEVel CHANnel3, 0.15") # 150mV
|
||||||
test_running = False
|
scope.write(":TRIGger:SWEep SINGle")
|
||||||
break
|
print("Scope Configured.")
|
||||||
|
|
||||||
# Toggle state for next iteration
|
def save_screenshot(iteration):
|
||||||
state = "off" if state == "on" else "on"
|
try:
|
||||||
|
scope.clear()
|
||||||
|
# Use a very short, simple path. No subdirectories if possible.
|
||||||
|
# Ensure C:\TEMP exists!
|
||||||
|
filename = f"C:\\TEMP\\cap{iteration}.png"
|
||||||
|
|
||||||
# Sleep in small increments so we can exit the thread quickly
|
print(f"ATTEMPTING SIMPLEST XP SYNTAX: {filename}")
|
||||||
for _ in range(int(interval * 10)):
|
|
||||||
if not test_running: break
|
# Try ONLY the filename and the format.
|
||||||
time.sleep(0.1)
|
# Many XP versions hate the SCR,COL extra parameters via SCPI.
|
||||||
|
cmd = f':DISK:SAVE:IMAGe "{filename}",PNG'
|
||||||
|
|
||||||
|
scope.write(cmd)
|
||||||
|
|
||||||
|
# DO NOT use scope.ask() or any queries yet.
|
||||||
|
# Just sleep and check the folder manually.
|
||||||
|
time.sleep(5.0)
|
||||||
|
print("COMMAND SENT. CHECK C:\\TEMP")
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
print(f"XP COMM ERROR: {e}")
|
||||||
|
|
||||||
|
def test_worker(on_time):
|
||||||
|
""" Background loop:
|
||||||
|
- Sets display ON for user-defined duration.
|
||||||
|
- Sets display OFF for exactly 1 second.
|
||||||
|
"""
|
||||||
|
global test_running
|
||||||
|
count = 1
|
||||||
|
|
||||||
|
while test_running:
|
||||||
|
# Arm scope
|
||||||
|
scope.write(":SINGle")
|
||||||
|
|
||||||
|
# Start display
|
||||||
|
requests.put(URL, json={"state": "on"}, timeout=2)
|
||||||
|
|
||||||
|
# Wait for the scope to trigger and fill memory
|
||||||
|
# We wait slightly longer than the 'on_time' to ensure capture is done
|
||||||
|
time.sleep(on_time + 0.5)
|
||||||
|
|
||||||
|
# Pull the image
|
||||||
|
save_screenshot(count)
|
||||||
|
count += 1
|
||||||
|
|
||||||
|
# Set Display OFF
|
||||||
|
requests.put(URL, json={"state": "off"}, timeout=2)
|
||||||
|
|
||||||
|
# Fixed 1s OFF period
|
||||||
|
time.sleep(1.0)
|
||||||
|
|
||||||
def main_menu():
|
def main_menu():
|
||||||
global test_running
|
global test_running
|
||||||
|
|
||||||
"""Main Application Loop."""
|
"""Main Application Loop."""
|
||||||
while True:
|
while True:
|
||||||
print("\n=== MIPI TEST CONTROL ===")
|
print("\n===== MIPI TEST CONTROL =====")
|
||||||
print("1. RUN IDN CHECK (PSU & SCOPE)")
|
print("1. RUN IDN CHECK (PSU & SCOPE)")
|
||||||
print("2. CONFIGURE PSU (DEFAULT 24V/1.5A)")
|
print("2. SETUP SCOPE (RUN FIRST)")
|
||||||
print("3. PSU OUTPUT ON/OFF (CH1)")
|
print("3. CONFIGURE PSU (DEFAULT 24V/1.5A)")
|
||||||
print("4. START FLICKER TEST")
|
print("4. PSU OUTPUT ON/OFF (CH1)")
|
||||||
print("5. STOP FLICKER TEST")
|
print("5. START TEST & CAPTURE")
|
||||||
print("6. EXIT")
|
print("6. STOP TEST")
|
||||||
|
print("7. EXIT")
|
||||||
|
|
||||||
choice = input("\nSELECT OPTION (1-6): ")
|
choice = input("\nSELECT OPTION (1-7): ")
|
||||||
|
|
||||||
if choice == '1':
|
if choice == '1':
|
||||||
print(f"PSU: {psu.ask('*IDN?')}")
|
print(f"PSU: {psu.ask('*IDN?')}")
|
||||||
print(f"SCOPE: {scope.ask('*IDN?')}")
|
print(f"SCOPE: {scope.ask('*IDN?')}")
|
||||||
elif choice == '2':
|
elif choice == '2':
|
||||||
|
setup_scope()
|
||||||
|
elif choice == '3':
|
||||||
psu.write('CH1:VOLT 24.0')
|
psu.write('CH1:VOLT 24.0')
|
||||||
psu.write('CH1:CURR 1.5')
|
psu.write('CH1:CURR 1.5')
|
||||||
print("PSU CONFIGURED...")
|
print("PSU CONFIGURED...")
|
||||||
elif choice == '3':
|
elif choice == '4':
|
||||||
state = input("TYPE 'ON' OR 'OFF': ").upper()
|
state = input("TYPE 'ON' OR 'OFF': ").upper()
|
||||||
psu.write(f'OUTP CH1,{state}')
|
psu.write(f'OUTP CH1,{state}')
|
||||||
elif choice == '4':
|
elif choice == '5':
|
||||||
if not test_running:
|
if not test_running:
|
||||||
try:
|
try:
|
||||||
sec = float(input("ENTER TOGGLE INTERVAL IN SECONDS: "))
|
sec = float(input("ON DURATION (sec): "))
|
||||||
test_running = True
|
test_running = True
|
||||||
t = threading.Thread(target=test_worker, args=(sec,), daemon=True)
|
t = threading.Thread(target=test_worker, args=(sec,), daemon=True)
|
||||||
t.start()
|
t.start()
|
||||||
@@ -91,10 +144,10 @@ def main_menu():
|
|||||||
print("INVALID TIME ENTERED.")
|
print("INVALID TIME ENTERED.")
|
||||||
else:
|
else:
|
||||||
print("TEST IS ALREADY RUNNING!")
|
print("TEST IS ALREADY RUNNING!")
|
||||||
elif choice == '5':
|
elif choice == '6':
|
||||||
print("STOPPING TEST...")
|
print("STOPPING TEST...")
|
||||||
test_running = False
|
test_running = False
|
||||||
elif choice == '6':
|
elif choice == '7':
|
||||||
test_running = False # Ensure thread stops
|
test_running = False # Ensure thread stops
|
||||||
psu.close()
|
psu.close()
|
||||||
scope.close()
|
scope.close()
|
||||||
|
|||||||
Reference in New Issue
Block a user