LLM-Powered NI Measurement Analyzer

Use a local LLM to detect anomalies in NI PXI/VST RF measurement data.

Introduction

30 Second Summary

Every RF engineer has stared at a spreadsheet of measurement data, hunting for the one reading that drifted out of spec before a product ships.

In this project, you will build a Python application that simulates NI PXI/VST measurement data, feeds it to a local LLM running on your machine, and gets back structured anomaly reports and automated test case recommendations.

What You'll Build

A command-line pipeline called NI Measurement Intelligence that generates realistic RF measurement data, analyzes it with a local Qwen 2.5 14B model via Ollama, and outputs structured test reports and test case recommendations.

By the end of this project, you'll have:

A measurement data generator that produces realistic CSV datasets mimicking NI PXIe-5842 VST output, complete with configurable anomaly injection.
An LLM-powered analyzer that detects out-of-spec measurements against instrument specifications and produces structured JSON reports with severity ratings.
An automated test case generator that uses the LLM to propose LabVIEW VI test parameters based on observed anomaly patterns.
A trend analysis mode that compares multiple measurement runs over time and identifies degradation patterns (Secret Mission).

Do I need NI hardware or a LabVIEW license?

No. This project uses simulated measurement data that mirrors real NI instrument output formats. You do not need any NI hardware, drivers, or LabVIEW installed. You should be comfortable with Python (functions, dictionaries, file I/O). Familiarity with RF concepts like EVM and PXI is helpful but not required.

Set Up Your Environment and Pull the Model

Before you can analyze any measurement data, you need a local LLM ready to do the heavy lifting. Your analysis pipeline runs entirely on your machine with no cloud APIs or costs involved. In this step, you will pull the Qwen 2.5 14B model into Ollama, set up your Python project, and confirm everything can talk to each other.

In This Step, Get Ready To:

Pull the Qwen 2.5 14B model into Ollama.
Create a Python project with a virtual environment and install the Ollama SDK.
Verify that your Python code can communicate with the local LLM.

Pull the Qwen 2.5 14B Model

The Qwen 2.5 14B model scores 72.5% on HumanEval for code generation and produces high-quality structured JSON output. That JSON capability is critical because your analyzer and test case generator both depend on parsing structured responses from the LLM. The model is approximately 9 GB on disk and needs around 14 GB of RAM to run, which fits comfortably within your 32 GB system.

Open a terminal and pull the model. This is a one-time download.

ollama pull qwen2.5:14b

What does this command do?

The ollama pull command downloads the model weights to your local machine. Once pulled, the model runs entirely offline. No internet connection is needed for inference.

Verify the model is available by listing your local models.

ollama list

You should see qwen2.5:14b in the output along with its size and modification date.

Why Qwen 2.5 14B Instead of a Smaller Model?

You might wonder why not use a smaller model like Llama 3.3 8B, which is only 4.9 GB. The answer comes down to structured output quality. This project relies on the LLM returning well-formed JSON with correct field names and sensible values. Qwen 2.5 14B consistently produces cleaner structured output than smaller alternatives. Since your system has 32 GB of RAM, you have plenty of headroom to run the larger model.

Create the Project Structure and Install the Ollama Python Package

Your project has four Python modules that each handle a different part of the pipeline. A virtual environment keeps your project dependencies isolated from your system Python installation.

Create a project directory and set up a virtual environment.

macOS

mkdir ni-measurement-intelligence
cd ni-measurement-intelligence
python -m venv venv
source venv/bin/activate

Windows

mkdir ni-measurement-intelligence
cd ni-measurement-intelligence
python -m venv venv
venv\Scripts\activate

Install the Ollama Python SDK.

pip install ollama

What does this command do?

The ollama Python package provides a clean API for communicating with your local Ollama server. You will use its chat() function to send prompts to the Qwen 2.5 14B model and receive responses.

Create the four project files.

macOS

touch data_generator.py analyzer.py test_case_generator.py main.py

Windows

New-Item data_generator.py, analyzer.py, test_case_generator.py, main.py

Here is what each file will do as you build out the project:

data_generator.py creates realistic simulated NI measurement datasets.
analyzer.py feeds measurement data to the LLM for anomaly detection.
test_case_generator.py uses the LLM to propose LabVIEW VI test cases based on analysis results.
main.py ties the full pipeline together as a single entry point.

Verify Ollama Connectivity From Python

Before building any modules, you need to confirm that Python can reach the Ollama server and get a response from the model. A quick smoke test saves you from debugging connectivity issues later when your prompts are complex.

Open main.py in your editor and add the following connectivity test.

from ollama import chat

# Send a simple test message to verify the model responds
response = chat(model='qwen2.5:14b', messages=[
    {'role': 'user', 'content': 'Respond with just "OK" if you can read this.'}
])
print(response.message.content)

What does this code do?

The chat() function sends a list of messages to the specified model running on your local Ollama server. It returns a ChatResponse object. You access the model's reply through response.message.content.

Run the test.

python main.py

You should see "OK" printed in your terminal, or a similar short confirmation from the model.

I see OK

Your local LLM is responding correctly. Python, Ollama, and the Qwen 2.5 14B model are all connected and ready to go.

I see an error

The most common issue is that the Ollama server is not running.

Open a separate terminal window and start the Ollama server.

ollama serve

Go back to your original terminal and run python main.py again.

If you see a ConnectionError or ResponseError, make sure the Ollama server finished starting before you retry. Also confirm the model name matches exactly: qwen2.5:14b.

Still seeing errors?

Make sure your virtual environment is activated (you should see (venv) in your terminal prompt). Also verify you installed the ollama package with pip install ollama. Help me debug my Ollama connection error.

Your environment is fully set up. You have a local LLM running, a Python project ready to go, and confirmed connectivity between them. Next up, you will build the measurement data generator that creates realistic NI-style RF datasets for the LLM to analyze.

Build the Measurement Data Generator

Your LLM connection is live. Now you need data for it to analyze.

Real NI PXIe-5842 VST instruments produce measurement logs with RF metrics like EVM percentages, power levels, and frequency offsets. Since you don't need physical hardware for this project, you'll build a data generator that produces realistic simulated measurement datasets. These datasets will include configurable anomalies so the LLM has something interesting to find later.

In This Step, Get Ready To:

Define instrument specifications that mirror real NI PXIe-5842 VST parameters.
Implement data generation functions that produce normal and anomalous measurement records.
Generate a sample CSV dataset and verify it contains realistic measurement data.

Define Instrument Specifications and Measurement Schemas

Before generating any data, you need a "ground truth" that defines what good measurements look like. These specifications come from real industry standards like 3GPP and IEEE 802.11. The LLM will use these same specs later to judge whether a measurement passes or fails.

Open data_generator.py in your editor.
Add the following imports and instrument specification dictionary at the top of the file.

import csv
import random
import datetime
from dataclasses import dataclass, asdict
from typing import List

# Instrument specifications mirroring the NI PXIe-5842 VST
# These thresholds come from 3GPP and IEEE 802.11 standards
INSTRUMENT_SPECS = {
    "instrument": "PXIe-5842 VST",
    "frequency_range_ghz": {"min": 0.03, "max": 26.5},
    "max_bandwidth_mhz": 4000,
    "evm_limits_percent": {
        "QPSK": 17.5,
        "16QAM": 12.5,
        "64QAM": 8.0,
        "256QAM": 3.5
    },
    "power_level_dbm": {"nominal": -10.0, "tolerance": 2.0},
    "frequency_offset_ppm": 0.1,
    "iq_offset_dbc": -25.0,
    "sample_rate_mhz": 1000
}

What are these specs?

The EVM (Error Vector Magnitude) limits define how much a transmitted signal can deviate from its ideal constellation points. Higher-order modulations like 256QAM pack more data per symbol, so they require tighter EVM limits (3.5%) compared to simpler modulations like QPSK (17.5%).

The PXIe-5842 is a vector signal transceiver that combines a signal generator and analyzer in a four-slot PXI module. It covers 30 MHz to 26.5 GHz with up to 4 GHz instantaneous bandwidth.

Next, define a dataclass to represent a single measurement record. This gives you a structured schema that maps to what real NI instruments would output.

Add the following dataclass below the INSTRUMENT_SPECS dictionary.

@dataclass
class MeasurementRecord:
    timestamp: str
    channel_id: int
    center_frequency_ghz: float
    bandwidth_mhz: float
    modulation_type: str
    evm_percent: float
    power_level_dbm: float
    frequency_offset_hz: float
    iq_offset_db: float
    phase_error_deg: float
    pass_fail: str

Why a dataclass?

A dataclass gives you a clean schema with named fields, automatic __init__ and __repr__ methods, and easy conversion to a dictionary with asdict(). This makes writing to CSV straightforward.

Implement the Data Generation Functions

You need three functions: one to generate a normal (in-spec) measurement, one to generate an anomalous (out-of-spec) measurement, and one to combine them into a full dataset. The anomalous measurements are what the LLM will need to detect in the next step.

Add the following functions below your MeasurementRecord dataclass.

MODULATION_TYPES = ["QPSK", "16QAM", "64QAM", "256QAM"]

# Typical "good" EVM values for each modulation type
# These are well within the spec limits defined above
NORMAL_EVM_RANGES = {
    "QPSK": (1.5, 5.0),
    "16QAM": (1.5, 4.0),
    "64QAM": (1.0, 3.0),
    "256QAM": (0.5, 1.5)
}


def generate_normal_measurement(channel_id: int, timestamp: str) -> MeasurementRecord:
    """Generate a single measurement record with values within spec."""
    mod_type = random.choice(MODULATION_TYPES)
    evm_range = NORMAL_EVM_RANGES[mod_type]

    # Center frequency in the sub-6 GHz 5G band
    center_freq = round(random.uniform(3.3, 3.8), 3)

    # Normal power level: nominal -10 dBm with small random variation
    power = round(random.gauss(-10.0, 0.3), 2)

    # Frequency offset well within the +/- 0.1 PPM limit
    freq_offset = round(random.gauss(0, 5.0), 2)

    # IQ offset comfortably below the -25 dBc threshold
    iq_offset = round(random.uniform(-45.0, -28.0), 2)

    # Phase error in a healthy range
    phase_error = round(random.gauss(0, 0.5), 3)

    return MeasurementRecord(
        timestamp=timestamp,
        channel_id=channel_id,
        center_frequency_ghz=center_freq,
        bandwidth_mhz=100.0,
        modulation_type=mod_type,
        evm_percent=round(random.uniform(*evm_range), 3),
        power_level_dbm=power,
        frequency_offset_hz=freq_offset,
        iq_offset_db=iq_offset,
        phase_error_deg=phase_error,
        pass_fail="PASS"
    )

What does this function do?

The generate_normal_measurement() function creates a single measurement record that would pass all specification checks. The EVM values stay well below their modulation-specific limits. The power level hovers around the nominal -10 dBm. The frequency offset stays within a few Hz of center.

Now add the anomalous measurement function. This is where you inject different types of failures that the LLM will need to identify.

Add the following function below generate_normal_measurement().

def generate_anomalous_measurement(channel_id: int, timestamp: str) -> MeasurementRecord:
    """Generate a measurement with one or more out-of-spec values."""
    # Start with a normal measurement as the baseline
    record = generate_normal_measurement(channel_id, timestamp)

    # Pick an anomaly type to inject
    anomaly_type = random.choice(["evm_spike", "power_drift", "freq_offset", "iq_imbalance"])
    evm_limit = INSTRUMENT_SPECS["evm_limits_percent"][record.modulation_type]

    if anomaly_type == "evm_spike":
        # Push EVM above the spec limit for this modulation type
        record.evm_percent = round(random.uniform(evm_limit * 1.1, evm_limit * 2.0), 3)

    elif anomaly_type == "power_drift":
        # Shift power outside the +/- 2 dB tolerance band
        drift_direction = random.choice([-1, 1])
        record.power_level_dbm = round(-10.0 + drift_direction * random.uniform(2.5, 6.0), 2)

    elif anomaly_type == "freq_offset":
        # Push frequency offset beyond the 0.1 PPM limit
        # At 3.5 GHz, 0.1 PPM = 350 Hz, so we go well beyond that
        record.frequency_offset_hz = round(random.uniform(400, 1200) * random.choice([-1, 1]), 2)

    elif anomaly_type == "iq_imbalance":
        # Push IQ offset above the -25 dBc threshold (closer to 0 = worse)
        record.iq_offset_db = round(random.uniform(-24.0, -10.0), 2)

    record.pass_fail = "FAIL"
    return record

What does this function do?

This function starts with a normal measurement and then deliberately corrupts one metric. Each anomaly type simulates a real-world failure mode. An EVM spike could indicate a failing power amplifier. Power drift might mean a thermal issue. Frequency offset exceedance could point to a clock reference problem. IQ imbalance often indicates a mixer or calibration issue.

Finally, add the dataset generation function that combines normal and anomalous records and writes them to a CSV file.

Add the following function and __main__ block at the bottom of the file.

def generate_dataset(num_records: int, anomaly_rate: float, output_path: str = "measurement_data.csv") -> List[dict]:
    """Generate a full measurement dataset and save it to CSV."""
    records = []
    base_time = datetime.datetime(2026, 1, 15, 9, 0, 0)

    for i in range(num_records):
        # Each measurement is 2 seconds apart
        timestamp = (base_time + datetime.timedelta(seconds=i * 2)).isoformat()
        channel_id = random.randint(1, 4)

        # Decide if this record should be anomalous
        if random.random() < anomaly_rate:
            record = generate_anomalous_measurement(channel_id, timestamp)
        else:
            record = generate_normal_measurement(channel_id, timestamp)

        records.append(asdict(record))

    # Write to CSV
    if records:
        fieldnames = list(records[0].keys())
        with open(output_path, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(records)

    print(f"Generated {num_records} records ({int(anomaly_rate * 100)}% anomaly rate)")
    print(f"Saved to {output_path}")
    return records


if __name__ == "__main__":
    # Generate a sample dataset with 50 records and 15% anomaly rate
    data = generate_dataset(num_records=50, anomaly_rate=0.15)

    # Print a quick summary
    fail_count = sum(1 for r in data if r["pass_fail"] == "FAIL")
    print(f"\nSummary: {fail_count} FAIL out of {len(data)} total records")

What does this function do?

The generate_dataset() function creates a full measurement session. It generates timestamped records spaced 2 seconds apart (simulating real-time acquisition), randomly assigns channels, and uses the anomaly_rate parameter to control what percentage of records are out-of-spec. The output CSV format mirrors what you would export from NI InstrumentStudio or NI MAX.

Run the Generator and Inspect the Output

Time to see if your data generator produces realistic-looking measurement data.

Run the data generator from your terminal.

python data_generator.py

What should I see?

You should see output like:

Generated 50 records (15% anomaly rate)
Saved to measurement_data.csv

Summary: 7 FAIL out of 50 total records

The exact FAIL count will vary since anomaly injection is random. Expect somewhere between 4 and 12 failures out of 50 records.

Open measurement_data.csv in your editor or a spreadsheet application.

Scan through the data and verify three things. Most records should have in-spec values with EVM percentages well below their modulation limits. A handful of records should have obviously out-of-spec values, like an EVM of 25% for QPSK or a power level of -16 dBm. The FAIL records should have at least one metric that clearly exceeds its specification threshold.

Pro tip: spot-check your anomalies

Look at a FAIL record and check which metric is out of spec. For example, if the modulation type is 64QAM and the EVM is 12.5%, that is well above the 8.0% limit. This is exactly the kind of anomaly the LLM should flag in the next step.

Your data generator is producing realistic NI-style measurement datasets with embedded anomalies. Next up, you'll build the LLM-powered analyzer that reads this CSV and automatically detects those out-of-spec measurements.

Build the LLM-Powered Measurement Analyzer

You have a dataset of simulated RF measurements sitting in a CSV file. Some of those records contain injected anomalies. Now you need something that can read through all of that data, compare each measurement against the instrument specifications, and tell you exactly what went wrong and how severe it is.

This is where the LLM comes in. Instead of writing dozens of hardcoded threshold checks, you will feed the measurement data and instrument specifications to Qwen 2.5 14B and let it reason about what is in spec and what is not. The LLM will return structured JSON reports with anomaly types, severity ratings, and recommended actions.

In This Step, Get Ready To:

Design an analysis prompt that gives the LLM full instrument specification context.
Build a batched analysis pipeline that processes measurements through the Ollama API.
Test the analyzer against your sample data and verify it catches the injected anomalies.

Design the Analysis Prompt With Instrument Context

The quality of an LLM's analysis depends entirely on the context you provide. For the analyzer to make accurate pass/fail decisions, it needs the same specification tables that a human RF test engineer would reference. Without this context, the LLM would just be guessing at what constitutes an out-of-spec measurement.

Open analyzer.py in your editor.
Add the imports and a function that builds the analysis prompt.

import csv
import json
from ollama import chat

# Same instrument specs from data_generator.py — the LLM needs these
# as ground truth to judge each measurement
INSTRUMENT_SPECS = {
    'evm_limits': {
        'QPSK': 17.5,
        '16QAM': 12.5,
        '64QAM': 8.0,
        '256QAM': 3.5
    },
    'power_level': {
        'expected_dbm': -10.0,
        'tolerance_db': 2.0
    },
    'frequency_offset': {
        'max_ppm': 0.1
    },
    'iq_offset': {
        'max_dbc': -25.0
    }
}

def build_analysis_prompt(measurements, specs):
    """Construct system and user prompts for measurement analysis."""

    system_prompt = """You are an expert RF test engineer specializing in NI PXIe-5842 VST measurements.
You analyze measurement data against instrument specifications and identify anomalies.

INSTRUMENT SPECIFICATIONS:
- EVM Limits: QPSK <= 17.5%, 16QAM <= 12.5%, 64QAM <= 8.0%, 256QAM <= 3.5%
- Power Level: Expected -10.0 dBm, tolerance +/- 2.0 dB
- Frequency Offset: Within +/- 0.1 PPM
- IQ Offset: <= -25.0 dBc

For each measurement, determine if it passes or fails against these specs.
Respond with ONLY a valid JSON array. Each element must have these fields:
- measurement_index (int): the index of the measurement in the batch
- status (string): "PASS", "FAIL", or "WARNING"
- anomaly_type (string or null): e.g., "EVM_SPIKE", "POWER_DRIFT", "FREQ_OFFSET", "IQ_IMBALANCE", or null if PASS
- severity (string): "LOW", "MEDIUM", "HIGH", or "CRITICAL"
- explanation (string): why this measurement passed or failed, referencing the spec limit
- recommended_action (string or null): what to do if it failed, or null if PASS

Do not include any text outside the JSON array."""

    # Format measurements as a readable table for the LLM
    header = "Index | Modulation | EVM(%) | Power(dBm) | FreqOffset(Hz) | IQ_Offset(dB) | PhaseErr(deg)"
    rows = [header]
    for i, m in enumerate(measurements):
        row = f"{i} | {m['modulation_type']} | {m['evm_percent']} | {m['power_level_dbm']} | {m['frequency_offset_hz']} | {m['iq_offset_db']} | {m['phase_error_deg']}"
        rows.append(row)

    user_prompt = f"Analyze these {len(measurements)} measurements:\n\n" + "\n".join(rows)

    return system_prompt, user_prompt

Why Include the Full Spec Table in the Prompt?

The LLM does not inherently know that a QPSK EVM of 18% is out of spec. By embedding the exact thresholds in the system prompt, you give the model the same reference material a human engineer would use. This is a core prompt engineering technique: provide domain-specific ground truth so the LLM can reason accurately rather than rely on its general training data.

Implement the Analysis Pipeline

LLM context windows have practical limits. Sending all 50 measurements in a single prompt can produce lower quality results because the model loses focus on individual records. Batching the data into groups of 10-15 records per call gives the LLM a manageable amount to analyze at once.

Add the following functions below your build_analysis_prompt function in analyzer.py.

def read_measurements(csv_path):
    """Read measurement data from a CSV file into a list of dicts."""
    measurements = []
    with open(csv_path, 'r') as f:
        reader = csv.DictReader(f)
        for row in reader:
            # Convert numeric fields from strings
            row['evm_percent'] = float(row['evm_percent'])
            row['power_level_dbm'] = float(row['power_level_dbm'])
            row['frequency_offset_hz'] = float(row['frequency_offset_hz'])
            row['iq_offset_db'] = float(row['iq_offset_db'])
            row['phase_error_deg'] = float(row['phase_error_deg'])
            measurements.append(row)
    return measurements

def analyze_batch(batch, specs, batch_offset):
    """Send a batch of measurements to the LLM for analysis."""
    system_prompt, user_prompt = build_analysis_prompt(batch, specs)

    response = chat(
        model='qwen2.5:14b',
        messages=[
            {'role': 'system', 'content': system_prompt},
            {'role': 'user', 'content': user_prompt}
        ]
    )

    content = response.message.content.strip()

    # Try to parse JSON from the response
    try:
        results = json.loads(content)
    except json.JSONDecodeError:
        # If the LLM wrapped JSON in markdown code fences, extract it
        if '```' in content:
            json_str = content.split('```')[1]
            if json_str.startswith('json'):
                json_str = json_str[4:]
            results = json.loads(json_str.strip())
        else:
            print(f"Warning: Could not parse LLM response for batch at offset {batch_offset}")
            print(f"Raw response: {content[:200]}")
            results = []

    # Adjust measurement indices to reflect position in full dataset
    for result in results:
        result['measurement_index'] += batch_offset

    return results

def analyze_measurements(csv_path, batch_size=10):
    """Run the full analysis pipeline on a measurement CSV."""
    measurements = read_measurements(csv_path)
    all_results = []

    print(f"Analyzing {len(measurements)} measurements in batches of {batch_size}...")

    for i in range(0, len(measurements), batch_size):
        batch = measurements[i:i + batch_size]
        print(f"  Processing batch {i // batch_size + 1} (records {i}-{i + len(batch) - 1})...")
        batch_results = analyze_batch(batch, INSTRUMENT_SPECS, i)
        all_results.extend(batch_results)

    return all_results, measurements

What Does the JSON Extraction Fallback Do?

Even when you ask an LLM to return only JSON, it sometimes wraps the response in markdown code fences like ```json ... ```. The fallback logic strips those fences and extracts the raw JSON. This is a common pattern when working with LLM-generated structured output.

Now add the report generation function that summarizes the analysis results.

Add the generate_report and print_report functions below the analysis functions.

def generate_report(analysis_results, measurements):
    """Produce a summary report from the analysis results."""
    total = len(measurements)
    pass_count = sum(1 for r in analysis_results if r.get('status') == 'PASS')
    fail_count = sum(1 for r in analysis_results if r.get('status') == 'FAIL')
    warning_count = sum(1 for r in analysis_results if r.get('status') == 'WARNING')

    # Sort anomalies by severity
    severity_order = {'CRITICAL': 0, 'HIGH': 1, 'MEDIUM': 2, 'LOW': 3}
    anomalies = [r for r in analysis_results if r.get('status') != 'PASS']
    anomalies.sort(key=lambda x: severity_order.get(x.get('severity', 'LOW'), 4))

    # Determine overall health based on fail rate
    fail_rate = fail_count / total if total > 0 else 0
    if fail_rate > 0.3:
        health = 'CRITICAL - Systemic issues detected'
    elif fail_rate > 0.15:
        health = 'DEGRADED - Multiple anomalies detected'
    elif fail_rate > 0.05:
        health = 'FAIR - Minor issues detected'
    else:
        health = 'HEALTHY - System operating within spec'

    report = {
        'summary': {
            'total_measurements': total,
            'pass': pass_count,
            'fail': fail_count,
            'warning': warning_count,
            'fail_rate': round(fail_rate, 3),
            'overall_health': health
        },
        'anomalies': anomalies,
        'all_results': analysis_results
    }

    return report

def print_report(report):
    """Print a human-readable version of the analysis report."""
    s = report['summary']
    print("\n" + "=" * 60)
    print("MEASUREMENT ANALYSIS REPORT")
    print("=" * 60)
    print(f"Total Measurements: {s['total_measurements']}")
    print(f"Pass: {s['pass']} | Fail: {s['fail']} | Warning: {s['warning']}")
    print(f"Fail Rate: {s['fail_rate'] * 100:.1f}%")
    print(f"Overall Health: {s['overall_health']}")
    print("\n--- Anomalies (sorted by severity) ---")

    for a in report['anomalies']:
        print(f"\n  Record #{a['measurement_index']}:")
        print(f"    Status: {a['status']} | Severity: {a['severity']}")
        print(f"    Type: {a.get('anomaly_type', 'N/A')}")
        print(f"    Explanation: {a.get('explanation', 'N/A')}")
        print(f"    Action: {a.get('recommended_action', 'N/A')}")

    print("\n" + "=" * 60)

What Does This Code Do?

The generate_report function tallies up pass, fail, and warning counts from the LLM's analysis results. It calculates a fail rate and assigns an overall health status ranging from HEALTHY to CRITICAL. All anomalies are sorted by severity so the most urgent issues appear first.

The print_report function formats the report into a readable terminal output, listing each anomaly with its severity, type, explanation, and recommended action.

Test the Analyzer With Your Sample Data

Time to see if the LLM can actually catch the anomalies you injected in the previous step. The analyzer will read your measurement_data.csv, send batches to Qwen 2.5 14B, and produce a structured report.

Add the following __main__ block at the bottom of analyzer.py.

if __name__ == '__main__':
    # Run analysis on the sample dataset from the previous step
    results, measurements = analyze_measurements('measurement_data.csv', batch_size=10)
    report = generate_report(results, measurements)
    print_report(report)

    # Save the report as JSON for the test case generator
    with open('analysis_report.json', 'w') as f:
        json.dump(report, f, indent=2)
    print("\nReport saved to analysis_report.json")

Run the analyzer.

python analyzer.py

This will take a minute or two as the LLM processes each batch. You will see progress messages as each batch completes.

What Should I See?

You should see a report that identifies the anomalous records you injected in the data generator. The LLM should flag records with high EVM values as FAIL with severity HIGH or CRITICAL. Power drift anomalies should reference the +/- 2.0 dB tolerance. The recommended actions should be practical suggestions like "recalibrate", "check RF connections", or "investigate power amplifier".

If the LLM misses some anomalies or produces unexpected severity ratings, that is normal. LLM output is non-deterministic. The important thing is that it catches most of the injected issues and references the correct specification limits in its explanations.

Getting JSON Parse Errors?

If you see "Warning: Could not parse LLM response" messages, the model may be adding extra text around the JSON. Try reducing the batch size to 5 records. Smaller batches give the LLM less to process and it tends to follow the JSON-only instruction more reliably. You can also help me debug my LLM JSON parsing issue.

Ollama Not Responding?

Make sure the Ollama service is running. Open a separate terminal and run ollama serve. If you see a Connection refused error, the service has not started yet. You can also help me troubleshoot my Ollama connection.

Your analyzer is now producing structured reports that identify exactly which measurements are out of spec and why. Next up, you will use these analysis results to automatically generate test cases that target the specific weaknesses found in your data.

Build the Test Case Generator

Your analyzer can now detect anomalies in measurement data and explain what went wrong. The next question is: what should you test next? In this step, you will build a test case generator that uses the LLM to recommend LabVIEW VI input parameters based on the instrument specifications and the anomalies your analyzer found.

In This Step, Get Ready To:

Design a prompt that turns analysis results into actionable test cases with VI input parameters and expected output ranges.
Implement the test case generation pipeline with structured JSON output and priority sorting.
Run the generator and verify that test cases target the specific anomalies found in your data.

Design the Test Case Generation Prompt

The test case generator bridges the gap between anomaly detection and actionable testing. Instead of manually deciding what to test next, the LLM will reason about which instrument parameters to exercise based on the problems it found.

A LabVIEW VI has Controls (inputs) and Indicators (outputs). A typical RF test VI would have controls like center_frequency, bandwidth, power_level, and modulation_type. Indicators would include measured_evm, measured_power, and pass_fail. Your prompt needs to teach the LLM this structure so it can generate realistic test parameters.

Open test_case_generator.py.
Create a function called build_test_case_prompt(specs, analysis_results) that constructs a prompt with the following components:

The system prompt should establish the LLM as an RF test engineer and provide:

The full instrument specifications (EVM limits per modulation type, power tolerances, frequency offset limits).
The anomalies found in the analysis step, so the LLM can target specific weaknesses.
Standard RF testing best practices as context for generating thorough test cases.

The prompt should request structured JSON output where each test case includes these fields: test_case_id, test_name, description, vi_input_parameters (a dictionary of control names and values), expected_output_ranges (a dictionary of indicator names and acceptable ranges), priority (HIGH/MEDIUM/LOW), and rationale.

import json
from ollama import chat

def get_instrument_specs():
    """Return instrument specifications for the PXIe-5842 VST."""
    return {
        "instrument": "NI PXIe-5842 VST",
        "frequency_range_ghz": {"min": 0.03, "max": 26.5},
        "max_bandwidth_mhz": 4000,
        "evm_limits_percent": {
            "QPSK": 17.5,
            "16QAM": 12.5,
            "64QAM": 8.0,
            "256QAM": 3.5
        },
        "power_level_dbm": {"nominal": -10, "tolerance": 2.0},
        "frequency_offset_ppm": 0.1,
        "iq_offset_dbc": -25
    }

def build_test_case_prompt(specs, analysis_results):
    """Construct the prompt for test case generation."""
    system_prompt = f"""You are an experienced RF test engineer specializing in NI PXI instrumentation
and LabVIEW test automation. You generate test cases for LabVIEW VIs that validate
RF measurement systems.

Instrument Specifications:
{json.dumps(specs, indent=2)}

A LabVIEW VI for RF testing has:
- Controls (inputs): center_frequency (GHz), bandwidth (MHz), power_level (dBm),
  modulation_type (string), num_samples (integer)
- Indicators (outputs): measured_evm (%), measured_power (dBm),
  frequency_error (Hz), iq_offset (dB), pass_fail (boolean)

Generate test cases as a JSON array. Each test case must have:
- test_case_id: string (e.g., "TC-001")
- test_name: string
- description: string
- vi_input_parameters: dict of control names to values
- expected_output_ranges: dict of indicator names to {{"min": value, "max": value}}
- priority: "HIGH" or "MEDIUM" or "LOW"
- rationale: string explaining why this test matters

Return ONLY the JSON array, no other text."""

    user_message = f"""Based on the following anomaly analysis results, generate targeted test cases
that would help verify and diagnose the issues found:

{json.dumps(analysis_results, indent=2)}

Generate test cases that:
1. Target the specific anomaly types found in the analysis
2. Include boundary condition tests near specification limits
3. Use realistic parameter ranges for the PXIe-5842 VST
4. Cover both the failing conditions and verification of passing conditions"""

    return system_prompt, user_message

Why Feed Analysis Results Into the Prompt?

By providing the anomalies from the previous step, the LLM can generate test cases that specifically target the problems it found. If the analyzer detected EVM spikes on 64QAM signals, the test case generator will propose tests that exercise 64QAM at various power levels and frequencies. This creates a feedback loop between analysis and testing.

Implement the Test Case Generation Pipeline

The generation pipeline calls the Ollama chat API with your prompt, parses the JSON response, validates the structure, and sorts the results by priority.

Add the following functions to test_case_generator.py below the prompt function.

def generate_test_cases(specs, analysis_results, num_cases=5):
    """Generate test cases using the LLM based on specs and analysis results."""
    system_prompt, user_message = build_test_case_prompt(specs, analysis_results)
    user_message += f"\n\nGenerate exactly {num_cases} test cases."

    # Call the LLM
    response = chat(
        model='qwen2.5:14b',
        messages=[
            {'role': 'system', 'content': system_prompt},
            {'role': 'user', 'content': user_message}
        ]
    )

    response_text = response.message.content

    # Parse JSON from the response
    test_cases = parse_test_cases(response_text)

    # Validate each test case has required fields
    validated = []
    required_fields = [
        'test_case_id', 'test_name', 'description',
        'vi_input_parameters', 'expected_output_ranges',
        'priority', 'rationale'
    ]
    for tc in test_cases:
        if all(field in tc for field in required_fields):
            validated.append(tc)
        else:
            missing = [f for f in required_fields if f not in tc]
            print(f"Skipping test case with missing fields: {missing}")

    # Sort by priority: HIGH first, then MEDIUM, then LOW
    priority_order = {'HIGH': 0, 'MEDIUM': 1, 'LOW': 2}
    validated.sort(key=lambda tc: priority_order.get(tc.get('priority', 'LOW'), 3))

    return validated

def parse_test_cases(response_text):
    """Extract and parse JSON from the LLM response."""
    # Try direct JSON parsing first
    try:
        return json.loads(response_text)
    except json.JSONDecodeError:
        pass

    # Try extracting JSON from markdown code blocks
    if '```' in response_text:
        start = response_text.find('```')
        # Skip the language identifier line
        start = response_text.find('\n', start) + 1
        end = response_text.find('```', start)
        if end > start:
            try:
                return json.loads(response_text[start:end].strip())
            except json.JSONDecodeError:
                pass

    # Try finding array brackets
    start = response_text.find('[')
    end = response_text.rfind(']') + 1
    if start >= 0 and end > start:
        try:
            return json.loads(response_text[start:end])
        except json.JSONDecodeError:
            pass

    print("Warning: Could not parse LLM response as JSON.")
    print("Raw response:", response_text[:500])
    return []

def format_test_plan(test_cases):
    """Format test cases into a human-readable test plan."""
    lines = []
    lines.append("=" * 60)
    lines.append("TEST PLAN: NI PXIe-5842 VST Validation")
    lines.append("=" * 60)
    lines.append(f"Total test cases: {len(test_cases)}")
    lines.append("")

    for tc in test_cases:
        lines.append(f"--- {tc['test_case_id']}: {tc['test_name']} ---")
        lines.append(f"Priority: {tc['priority']}")
        lines.append(f"Description: {tc['description']}")
        lines.append("")

        lines.append("VI Input Parameters (Controls):")
        for param, value in tc.get('vi_input_parameters', {}).items():
            lines.append(f"  {param}: {value}")
        lines.append("")

        lines.append("Expected Output Ranges (Indicators):")
        for indicator, range_val in tc.get('expected_output_ranges', {}).items():
            if isinstance(range_val, dict):
                lines.append(f"  {indicator}: {range_val.get('min', 'N/A')} to {range_val.get('max', 'N/A')}")
            else:
                lines.append(f"  {indicator}: {range_val}")
        lines.append("")

        lines.append(f"Rationale: {tc['rationale']}")
        lines.append("")

    return "\n".join(lines)

What Does This Code Do?

The generate_test_cases() function sends your prompt to the Qwen 2.5 14B model and collects the response. It then validates that every test case has all required fields before sorting them by priority.

The parse_test_cases() function handles the reality that LLMs sometimes wrap JSON in markdown code blocks or add extra text. It tries three extraction strategies: direct parsing, markdown block extraction, and bracket-based extraction.

The format_test_plan() function converts the structured JSON into a readable test plan document that a test engineer could review and hand off to a LabVIEW developer.

Run the Test Case Generator and Review the Output

To test the generator independently, you need some analysis results to feed into it. You will create a sample set of analysis results and then generate test cases from them.

Add the following __main__ block at the bottom of test_case_generator.py.

if __name__ == '__main__':
    specs = get_instrument_specs()

    # Sample analysis results simulating what the analyzer would produce
    sample_analysis = {
        "total_measurements": 50,
        "pass_count": 42,
        "fail_count": 6,
        "warning_count": 2,
        "anomalies": [
            {
                "measurement_index": 7,
                "status": "FAIL",
                "anomaly_type": "EVM_SPIKE",
                "severity": "HIGH",
                "explanation": "EVM of 22.3% on 64QAM exceeds 8.0% limit by 178%",
                "modulation_type": "64QAM"
            },
            {
                "measurement_index": 15,
                "status": "FAIL",
                "anomaly_type": "POWER_DRIFT",
                "severity": "MEDIUM",
                "explanation": "Power level at -14.8 dBm, outside -10 +/- 2 dBm tolerance",
                "modulation_type": "QPSK"
            },
            {
                "measurement_index": 23,
                "status": "FAIL",
                "anomaly_type": "FREQUENCY_OFFSET",
                "severity": "CRITICAL",
                "explanation": "Frequency offset of 450 Hz exceeds +/- 0.1 PPM limit",
                "modulation_type": "16QAM"
            },
            {
                "measurement_index": 31,
                "status": "WARNING",
                "anomaly_type": "IQ_IMBALANCE",
                "severity": "LOW",
                "explanation": "IQ offset at -26.1 dBc, approaching -25 dBc limit",
                "modulation_type": "256QAM"
            }
        ]
    }

    print("Generating test cases based on analysis results...")
    print("(This may take 30-60 seconds)\n")

    test_cases = generate_test_cases(specs, sample_analysis, num_cases=5)

    # Display the formatted test plan
    test_plan = format_test_plan(test_cases)
    print(test_plan)

    # Save to JSON
    with open('test_plan.json', 'w') as f:
        json.dump(test_cases, f, indent=2)
    print(f"\nSaved {len(test_cases)} test cases to test_plan.json")

Run the test case generator.

python test_case_generator.py

The LLM will take 30-60 seconds to generate the test cases. Once complete, you will see a formatted test plan printed to the terminal and a test_plan.json file saved to your project directory.

What Should I See?

Your test plan should contain 5 test cases sorted by priority. Look for these quality indicators:

Test cases that target the specific anomalies in the sample data (EVM spikes on 64QAM, power drift, frequency offset issues).
VI input parameters with realistic ranges. Center frequency should be around 3.5 GHz. Power levels should fall in the -30 to +10 dBm range.
Expected output ranges that align with the instrument specifications (e.g., EVM below 8.0% for 64QAM).
Rationale text that explains why each test case matters in the context of the anomalies found.

Getting Empty or Malformed Output?

If the LLM returns malformed JSON, the parse_test_cases() function will print the raw response so you can see what went wrong. LLMs occasionally add commentary before or after the JSON. The three-stage parser handles most cases, but if you consistently get empty results, try running the script again. LLM outputs are non-deterministic, so a second run often produces cleaner JSON.

If you are still stuck, ask NextWork for help debugging your test case generator output.

Your test case generator is now producing intelligent, targeted test recommendations based on real anomaly data. In the next step, you will wire all three modules together into a single pipeline that generates data, analyzes it, and produces test cases in one command.

Integrate the Full Pipeline

You have three independent modules that each do their job well. Now it is time to connect them into a single, seamless pipeline that a user can run with one command. In this step, you will wire up main.py as the entry point that orchestrates data generation, analysis, and test case generation, with command-line arguments for full control.

In This Step, Get Ready To:

Wire up the main entry point to orchestrate all three modules in sequence.
Run the full pipeline end-to-end and verify the JSON output files.
Experiment with different anomaly rates and measurement counts to stress-test the LLM's analysis.

Wire Up the Main Entry Point

Right now, each module has its own __main__ block for standalone testing. The real power comes from chaining them together. The argparse module from Python's standard library lets you define command-line arguments so users can control the pipeline without editing code.

Open main.py and replace the connectivity test with the full pipeline code.

Your main.py should:

Import argparse, json, and os from the standard library, plus the functions from your three modules.
Set up an argument parser with four arguments: --num-measurements (default 30), --anomaly-rate (default 0.2), --num-test-cases (default 5), and --output-dir (default output).
Create the output directory if it does not exist using os.makedirs(output_dir, exist_ok=True).
Call the data generator to produce a CSV file in the output directory.
Pass the CSV path to the analyzer and capture the analysis results.
Feed the analysis results and instrument specs into the test case generator.
Save the analysis report to analysis_report.json and the test plan to test_plan.json in the output directory.
Print a summary of the pipeline run to the terminal, including the number of measurements analyzed, anomalies found, and test cases generated.

Pro tip: print progress messages

Each LLM call can take several seconds. Add print statements like print("Generating measurement data...") before each pipeline stage so the user knows the application has not frozen.

I'll try it first

Use the structure above to write your main.py. Remember to use json.dump() with indent=2 when saving the JSON output files for readability.

Let me check my work

import argparse
import json
import os

from data_generator import generate_dataset, INSTRUMENT_SPECS
from analyzer import analyze_measurements, generate_report
from test_case_generator import generate_test_cases, format_test_plan

def main():
    # Set up command-line arguments for pipeline control
    parser = argparse.ArgumentParser(description='NI Measurement Intelligence Pipeline')
    parser.add_argument('--num-measurements', type=int, default=30,
                        help='Number of measurement records to generate')
    parser.add_argument('--anomaly-rate', type=float, default=0.2,
                        help='Fraction of records that should be anomalous (0.0 to 1.0)')
    parser.add_argument('--num-test-cases', type=int, default=5,
                        help='Number of test cases to generate')
    parser.add_argument('--output-dir', type=str, default='output',
                        help='Directory to save output files')
    args = parser.parse_args()

    # Create output directory if it doesn't exist
    os.makedirs(args.output_dir, exist_ok=True)

    # Stage 1: Generate measurement data
    csv_path = os.path.join(args.output_dir, 'measurement_data.csv')
    print(f'Generating {args.num_measurements} measurements '
          f'with {args.anomaly_rate:.0%} anomaly rate...')
    generate_dataset(args.num_measurements, args.anomaly_rate, csv_path)
    print(f'Saved measurement data to {csv_path}')

    # Stage 2: Analyze measurements with the LLM
    print('Analyzing measurements with Qwen 2.5 14B...')
    analysis_results = analyze_measurements(csv_path)
    report = generate_report(analysis_results)

    # Save analysis report to JSON
    report_path = os.path.join(args.output_dir, 'analysis_report.json')
    with open(report_path, 'w') as f:
        json.dump(report, f, indent=2)
    print(f'Saved analysis report to {report_path}')

    # Stage 3: Generate test cases based on analysis
    print(f'Generating {args.num_test_cases} test cases...')
    test_cases = generate_test_cases(
        INSTRUMENT_SPECS, analysis_results, args.num_test_cases
    )
    test_plan = format_test_plan(test_cases)

    # Save test plan to JSON
    plan_path = os.path.join(args.output_dir, 'test_plan.json')
    with open(plan_path, 'w') as f:
        json.dump(test_plan, f, indent=2)
    print(f'Saved test plan to {plan_path}')

    # Print summary
    print('\n--- Pipeline Complete ---')
    print(f'Measurements analyzed: {report.get("total_measurements", "N/A")}')
    print(f'Anomalies found: {report.get("fail_count", "N/A")}')
    print(f'Test cases generated: {len(test_cases)}')
    print(f'Output files saved to: {args.output_dir}/')

if __name__ == '__main__':
    main()

What does this code do?

This script is the orchestrator. It parses command-line arguments, then runs each pipeline stage in order: generate data, analyze it, and produce test cases. Each stage prints a progress message so you can see what is happening. The results are saved as JSON files in the output directory for later review.

Why a single entry point?

In real test engineering workflows, you want one command that runs the entire analysis. This mirrors how NI InstrumentStudio and TestStand orchestrate test sequences. A single entry point also makes it easy to integrate into CI/CD pipelines or scheduled test runs later.

Run the Full Pipeline End-to-End

With everything wired up, it is time to run the complete pipeline and verify that all three stages work together.

Make sure Ollama is running. If it is not, start it with ollama serve in a separate terminal.
Run the pipeline with the following command:

python main.py --num-measurements 30 --anomaly-rate 0.2 --num-test-cases 5

What does this command do?

This generates 30 measurement records where 20% are anomalous, analyzes them with the Qwen 2.5 14B model, and produces 5 test case recommendations. The entire process may take 1-3 minutes depending on your hardware, since each LLM call processes a batch of measurements.

You should see progress messages for each stage, followed by a summary showing the number of measurements analyzed, anomalies found, and test cases generated.

Open the output directory and inspect the generated files.
Open analysis_report.json and verify that it contains pass/fail counts, a list of anomalies with severity ratings, and an overall health assessment.
Open test_plan.json and verify that each test case has a name, VI input parameters, expected output ranges, and a rationale tied to the anomalies found.

What should I see?

Your analysis_report.json should show that roughly 6 out of 30 measurements failed (20% anomaly rate). The anomalies should have severity ratings like HIGH or CRITICAL for measurements with EVM values far above the spec limit. Your test_plan.json should contain 5 test cases with realistic LabVIEW VI parameters like center frequencies around 3.5 GHz and power levels in the -30 to +10 dBm range.

Pipeline failing partway through?

If the analyzer or test case generator fails with a JSON parsing error, the LLM may have returned malformed output. Check that your analysis and test case generation functions include retry logic for malformed responses. Help me debug my pipeline error.

Experiment With Different Scenarios

A good analysis pipeline should adapt its conclusions based on the data it receives. Running the pipeline with different configurations reveals how well the LLM adjusts its reasoning.

Run the pipeline with a high anomaly rate to simulate a systemic failure:

python main.py --num-measurements 30 --anomaly-rate 0.5 --num-test-cases 5 --output-dir output_high_anomaly

With 50% of measurements failing, the LLM should recognize this as a systemic problem rather than isolated incidents. Look for language in the analysis report that flags a broader equipment issue or calibration failure.

Run the pipeline with zero anomalies to confirm the LLM gives a clean report:

python main.py --num-measurements 30 --anomaly-rate 0.0 --num-test-cases 5 --output-dir output_clean

The analysis report should show all measurements passing. The test cases should still be generated, but their rationale should focus on standard verification rather than anomaly investigation.

Compare the analysis_report.json files across your three runs. Note how the LLM's overall health assessment, severity distribution, and recommended actions change based on the data patterns.

Pro tip: test modulation types

If your data generator supports different modulation types (QPSK, 16QAM, 64QAM, 256QAM), check that the LLM applies the correct EVM thresholds for each. A 10% EVM reading should pass for QPSK (limit: 17.5%) but fail for 64QAM (limit: 8%). This is a great way to verify the LLM is actually using the instrument specifications you provided in the prompt.

Your pipeline is complete. You have a fully integrated, command-line tool that generates data, analyzes it with a local LLM, and produces actionable test recommendations. Next up, take on the secret mission to add trend analysis across multiple measurement runs.

Secret mission

Add Trend Analysis Across Multiple Runs

Your analyzer works great for a single measurement session. But real test systems run continuously for days or weeks. What if you could detect a slow drift toward failure before it actually happens? In this secret mission, you will build a trend analysis mode that compares multiple measurement runs over time and has the LLM identify degradation patterns.

Before you go

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs or billing concerns. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

Ollama service running locally.
Qwen 2.5 14B model (~9 GB on disk).
Python virtual environment (venv) with the ollama package installed.
Project directory ni-measurement-intelligence with Python modules and generated output files (measurement_data.csv, analysis_report.json, test_plan.json).

Keep everything running

No action needed. Choose this if you are still actively building or want to experiment with different anomaly rates and measurement configurations.

Ollama will continue running in the background. The Qwen 2.5 14B model stays downloaded and ready to use.
Your project files and virtual environment remain in the ni-measurement-intelligence directory.
The model uses about 14 GB of RAM when actively loaded. Ollama unloads it from memory after a period of inactivity.

Pause - I'll come back to this later

Shut down running processes to free up memory while keeping your files and model intact for later.

Stop the Ollama service. On macOS, quit the Ollama app from the menu bar. On Linux, run systemctl stop ollama or close the terminal running ollama serve.
Deactivate your virtual environment by running deactivate in your terminal.
The Qwen 2.5 14B model (~9 GB) stays on disk so you do not need to re-download it. Your project files remain untouched.
To resume, start Ollama again and reactivate your virtual environment with source venv/bin/activate.

Delete - I don't want to use this again

Remove all project resources and free up disk space.

Delete the Qwen 2.5 14B model by running ollama rm qwen2.5:14b. This frees up approximately 9 GB of disk space.
Delete the project directory by running rm -rf ni-measurement-intelligence. This removes all Python files, the virtual environment, and generated output files.
If you no longer need Ollama for other projects, you can uninstall it entirely. On macOS, drag the Ollama app to Trash. On Linux, follow the Ollama uninstall instructions for your distribution.
Uninstalling Ollama removes all downloaded models, not just Qwen 2.5 14B. Only uninstall if you are sure you do not need any local models.

Mission accomplished

That's a Wrap!

Nice work! You've built a fully functional LLM-powered measurement data analysis pipeline that bridges the gap between traditional NI test engineering and modern AI tooling.

Here's what you learned:

Built a realistic NI PXI/VST measurement data simulator that generates RF metrics like EVM, power levels, and frequency offsets matching real instrument output formats.
Created an LLM-powered anomaly detection system that analyzes measurement data against instrument specifications and produces structured JSON reports with severity ratings and recommended actions.
Developed an automated test case generator that uses LLM reasoning to propose LabVIEW VI test parameters based on observed anomalies and instrument specs.
Added trend analysis to detect measurement degradation patterns across multiple test runs and project time-to-failure thresholds.

Ready to quiz yourself?