Inital version
This commit is contained in:
9
scope_parser/__init__.py
Normal file
9
scope_parser/__init__.py
Normal file
@@ -0,0 +1,9 @@
|
||||
"""
|
||||
Joe's Really Simple Scope Parser
|
||||
"""
|
||||
|
||||
from .parsers import parse_owon_data, parse_gwinstek_data
|
||||
from .data import ScopeData, ChannelData
|
||||
|
||||
__version__ = "0.1.0"
|
||||
__all__ = ["parse_owon_data", "parse_gwinstek_data", "ScopeData", "ChannelData"]
|
||||
72
scope_parser/base_parser.py
Normal file
72
scope_parser/base_parser.py
Normal file
@@ -0,0 +1,72 @@
|
||||
"""
|
||||
Base parser class for oscilloscope data parsers.
|
||||
"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Dict, List, Optional
|
||||
import numpy as np
|
||||
|
||||
from .data import ScopeData, ChannelData
|
||||
|
||||
|
||||
class BaseOscilloscopeParser(ABC):
|
||||
"""Base class for oscilloscope data parsers."""
|
||||
|
||||
@abstractmethod
|
||||
def parse(self, file_path: str) -> ScopeData:
|
||||
"""Parse oscilloscope data from file."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def can_parse(self, file_path: str) -> bool:
|
||||
"""Check if this parser can handle the given file."""
|
||||
pass
|
||||
|
||||
def _extract_time_interval(self, metadata: Dict[str, str]) -> float:
|
||||
"""Extract time interval between samples in seconds."""
|
||||
# Default implementation - can be overridden by subclasses
|
||||
interval_str = metadata.get("Time interval", "")
|
||||
if interval_str:
|
||||
if "uS" in interval_str:
|
||||
return float(interval_str.replace("uS", "")) * 1e-6
|
||||
elif "nS" in interval_str:
|
||||
return float(interval_str.replace("nS", "")) * 1e-9
|
||||
elif "mS" in interval_str:
|
||||
return float(interval_str.replace("mS", "")) * 1e-3
|
||||
elif "S" in interval_str:
|
||||
return float(interval_str.replace("S", ""))
|
||||
return 1e-6 # Default to 1 µs
|
||||
|
||||
def _extract_frequency(self, metadata: Dict[str, str]) -> Optional[float]:
|
||||
"""Extract frequency from metadata if available."""
|
||||
freq_str = metadata.get("Frequency", "")
|
||||
if freq_str.startswith("F="):
|
||||
# Extract frequency value and convert units
|
||||
freq_value = freq_str[2:]
|
||||
if "kHz" in freq_value:
|
||||
return float(freq_value.replace("kHz", "")) * 1000
|
||||
elif "Hz" in freq_value:
|
||||
return float(freq_value.replace("Hz", ""))
|
||||
return None
|
||||
|
||||
def _extract_vpp(self, metadata: Dict[str, str]) -> Optional[float]:
|
||||
"""Extract peak-to-peak voltage from metadata."""
|
||||
pp_str = metadata.get("PK-PK", "")
|
||||
if pp_str.startswith("Vpp="):
|
||||
value = float(pp_str[4:].replace("mV", "").replace("V", ""))
|
||||
# Convert mV to V if needed
|
||||
if "mV" in pp_str:
|
||||
return value / 1000.0
|
||||
return value
|
||||
return None
|
||||
|
||||
def _extract_average(self, metadata: Dict[str, str]) -> Optional[float]:
|
||||
"""Extract average voltage from metadata."""
|
||||
avg_str = metadata.get("Average", "")
|
||||
if avg_str.startswith("V="):
|
||||
value = float(avg_str[2:].replace("mV", "").replace("V", ""))
|
||||
# Convert mV to V if needed
|
||||
if "mV" in avg_str:
|
||||
return value / 1000.0
|
||||
return value
|
||||
return None
|
||||
100
scope_parser/data.py
Normal file
100
scope_parser/data.py
Normal file
@@ -0,0 +1,100 @@
|
||||
"""
|
||||
Data structures for oscilloscope measurements.
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Dict, Optional
|
||||
import numpy as np
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelData:
|
||||
"""Represents data from a single oscilloscope channel."""
|
||||
|
||||
channel_name: str
|
||||
voltage_values: np.ndarray # in volts
|
||||
time_values: np.ndarray # in seconds
|
||||
metadata: Dict[str, str]
|
||||
|
||||
@property
|
||||
def frequency(self) -> Optional[float]:
|
||||
"""Extract frequency from metadata if available."""
|
||||
freq_str = self.metadata.get("Frequency", "")
|
||||
if freq_str.startswith("F="):
|
||||
# Extract frequency value and convert units
|
||||
freq_value = freq_str[2:]
|
||||
if "kHz" in freq_value:
|
||||
return float(freq_value.replace("kHz", "")) * 1000
|
||||
elif "Hz" in freq_value:
|
||||
return float(freq_value.replace("Hz", ""))
|
||||
return None
|
||||
|
||||
@property
|
||||
def vpp(self) -> Optional[float]:
|
||||
"""Peak-to-peak voltage in volts."""
|
||||
pp_str = self.metadata.get("PK-PK", "")
|
||||
if pp_str.startswith("Vpp="):
|
||||
value = float(pp_str[4:].replace("mV", "").replace("V", ""))
|
||||
# Convert mV to V if needed
|
||||
if "mV" in pp_str:
|
||||
return value / 1000.0
|
||||
return value
|
||||
return None
|
||||
|
||||
@property
|
||||
def average(self) -> Optional[float]:
|
||||
"""Average voltage in volts (computed from data)."""
|
||||
return float(np.mean(self.voltage_values))
|
||||
|
||||
@property
|
||||
def metadata_average(self) -> Optional[float]:
|
||||
"""Average voltage from metadata in volts."""
|
||||
avg_str = self.metadata.get("Average", "")
|
||||
if avg_str.startswith("V="):
|
||||
value = float(avg_str[2:].replace("mV", "").replace("V", ""))
|
||||
# Convert mV to V if needed
|
||||
if "mV" in avg_str:
|
||||
return value / 1000.0
|
||||
return value
|
||||
return None
|
||||
|
||||
@property
|
||||
def time_interval(self) -> Optional[float]:
|
||||
"""Time interval between samples in seconds."""
|
||||
interval_str = self.metadata.get("Time interval", "")
|
||||
if interval_str:
|
||||
if "uS" in interval_str:
|
||||
return float(interval_str.replace("uS", "")) * 1e-6
|
||||
elif "nS" in interval_str:
|
||||
return float(interval_str.replace("nS", "")) * 1e-9
|
||||
elif "mS" in interval_str:
|
||||
return float(interval_str.replace("mS", "")) * 1e-3
|
||||
elif "S" in interval_str:
|
||||
return float(interval_str.replace("S", ""))
|
||||
return None
|
||||
|
||||
|
||||
@dataclass
|
||||
class ScopeData:
|
||||
"""Container for all oscilloscope measurement data."""
|
||||
|
||||
channels: Dict[str, ChannelData]
|
||||
metadata: Dict[str, str]
|
||||
|
||||
def __getitem__(self, channel_name: str) -> ChannelData:
|
||||
"""Allow dictionary-like access to channels."""
|
||||
return self.channels[channel_name]
|
||||
|
||||
def __iter__(self):
|
||||
"""Allow iteration over channels."""
|
||||
return iter(self.channels.values())
|
||||
|
||||
@property
|
||||
def channel_names(self) -> List[str]:
|
||||
"""Get list of all channel names."""
|
||||
return list(self.channels.keys())
|
||||
|
||||
@property
|
||||
def num_channels(self) -> int:
|
||||
"""Get number of channels."""
|
||||
return len(self.channels)
|
||||
118
scope_parser/gwinstek_parser.py
Normal file
118
scope_parser/gwinstek_parser.py
Normal file
@@ -0,0 +1,118 @@
|
||||
"""
|
||||
Parser for Gwinstek oscilloscope CSV files.
|
||||
"""
|
||||
|
||||
import csv
|
||||
from typing import Dict, List
|
||||
import numpy as np
|
||||
|
||||
from .base_parser import BaseOscilloscopeParser
|
||||
from .data import ScopeData, ChannelData
|
||||
|
||||
|
||||
class GwinstekParser(BaseOscilloscopeParser):
|
||||
"""Parser for Gwinstek oscilloscope CSV files."""
|
||||
|
||||
def can_parse(self, file_path: str) -> bool:
|
||||
"""Check if file is from Gwinstek scope."""
|
||||
try:
|
||||
with open(file_path, "r") as f:
|
||||
first_lines = [f.readline().strip() for _ in range(10)]
|
||||
# Gwinstek-specific
|
||||
return any("Memory Length" in line for line in first_lines)
|
||||
except:
|
||||
return False
|
||||
|
||||
def parse(self, file_path: str) -> ScopeData:
|
||||
"""Parse Gwinstek oscilloscope CSV file."""
|
||||
|
||||
with open(file_path, "r") as f:
|
||||
# Read header metadata
|
||||
metadata_lines = []
|
||||
data_start_line = 0
|
||||
|
||||
for i, line in enumerate(f):
|
||||
line = line.strip()
|
||||
if line.startswith("Waveform Data"): # Gwinstek uses this as separator
|
||||
data_start_line = i + 1
|
||||
break
|
||||
metadata_lines.append(line)
|
||||
|
||||
# Parse metadata
|
||||
metadata = self._parse_metadata(metadata_lines)
|
||||
|
||||
# Reset file pointer to data section
|
||||
f.seek(0)
|
||||
for _ in range(data_start_line):
|
||||
next(f)
|
||||
|
||||
# Parse waveform data (just numbers, one per line)
|
||||
voltage_data = []
|
||||
time_data = []
|
||||
|
||||
for i, line in enumerate(f):
|
||||
line = line.strip()
|
||||
if not line: # Skip empty lines
|
||||
continue
|
||||
|
||||
# Remove trailing comma TODO: If 2ch this might not be trailing
|
||||
if line.endswith(","):
|
||||
line = line[:-1]
|
||||
|
||||
try:
|
||||
# Gwinstek data is in pixels, yes its dumb
|
||||
pixel_value = float(line)
|
||||
|
||||
# Get volts per division from metadata
|
||||
volts_per_division = float(metadata.get("Vertical Scale", "1.0"))
|
||||
vertical_position = float(metadata.get("Vertical Position", "0.0"))
|
||||
|
||||
# Convert pixels to voltage: 25 pixels per division
|
||||
# Pixel 0 is in the middle, so we center around vertical_position
|
||||
voltage = (
|
||||
pixel_value / 25.0
|
||||
) * volts_per_division + vertical_position
|
||||
voltage_data.append(voltage)
|
||||
|
||||
# Calculate time values from sampling period
|
||||
sampling_period = float(metadata.get("Sampling Period", "2.0E-06"))
|
||||
time_data.append(i * sampling_period)
|
||||
|
||||
except ValueError:
|
||||
# Skip non-numeric lines
|
||||
continue
|
||||
|
||||
# Create channel data
|
||||
channel_name = metadata.get("Source", "CH1")
|
||||
|
||||
channel_data = ChannelData(
|
||||
channel_name=channel_name,
|
||||
voltage_values=np.array(voltage_data),
|
||||
time_values=np.array(time_data),
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
return ScopeData(channels={channel_name: channel_data}, metadata=metadata)
|
||||
|
||||
def _parse_metadata(self, metadata_lines: List[str]) -> Dict[str, str]:
|
||||
"""Parse metadata from header lines."""
|
||||
metadata = {}
|
||||
|
||||
for line in metadata_lines:
|
||||
if "," in line:
|
||||
parts = line.split(",", 2) # Gwinstek has 3 parts: key, value, empty
|
||||
if len(parts) >= 2:
|
||||
key = parts[0].strip()
|
||||
value = parts[1].strip()
|
||||
if value: # Only add non-empty values
|
||||
metadata[key] = value
|
||||
|
||||
return metadata
|
||||
|
||||
def _extract_time_interval(self, metadata: Dict[str, str]) -> float:
|
||||
"""Extract time interval between samples for Gwinstek format."""
|
||||
sampling_period = metadata.get("Sampling Period", "2.0E-06")
|
||||
try:
|
||||
return float(sampling_period)
|
||||
except ValueError:
|
||||
return 2.0e-6 # Default to 2 µs
|
||||
107
scope_parser/owon_parser.py
Normal file
107
scope_parser/owon_parser.py
Normal file
@@ -0,0 +1,107 @@
|
||||
"""
|
||||
Parser for OWON oscilloscope CSV files.
|
||||
"""
|
||||
|
||||
import csv
|
||||
from typing import Dict, List
|
||||
import numpy as np
|
||||
|
||||
from .base_parser import BaseOscilloscopeParser
|
||||
from .data import ScopeData, ChannelData
|
||||
|
||||
|
||||
class OwonParser(BaseOscilloscopeParser):
|
||||
"""Parser for OWON oscilloscope CSV files."""
|
||||
|
||||
def can_parse(self, file_path: str) -> bool:
|
||||
"""Check if file is from OWON scope."""
|
||||
try:
|
||||
with open(file_path, "r") as f:
|
||||
first_lines = [f.readline().strip() for _ in range(10)]
|
||||
# Look for OWON-specific patterns
|
||||
return any("Channel" in line and "CH" in line for line in first_lines)
|
||||
except:
|
||||
return False
|
||||
|
||||
def parse(self, file_path: str) -> ScopeData:
|
||||
"""Parse OWON oscilloscope CSV file."""
|
||||
|
||||
with open(file_path, "r") as f:
|
||||
# Read header metadata
|
||||
metadata_lines = []
|
||||
data_start_line = 0
|
||||
|
||||
for i, line in enumerate(f):
|
||||
line = line.strip()
|
||||
if not line: # Empty line indicates end of metadata
|
||||
data_start_line = i + 1
|
||||
break
|
||||
metadata_lines.append(line)
|
||||
|
||||
# Parse metadata
|
||||
metadata = self._parse_metadata(metadata_lines)
|
||||
|
||||
# Reset file pointer to data section
|
||||
f.seek(0)
|
||||
for _ in range(data_start_line):
|
||||
next(f)
|
||||
|
||||
# Parse CSV data
|
||||
reader = csv.DictReader(f)
|
||||
|
||||
# Extract data columns
|
||||
voltage_data = []
|
||||
time_data = []
|
||||
|
||||
for row in reader:
|
||||
# Get voltage data (look for CH*_Voltage column)
|
||||
voltage_col = None
|
||||
for col_name in row.keys():
|
||||
if "Voltage" in col_name:
|
||||
voltage_col = col_name
|
||||
break
|
||||
|
||||
if voltage_col is None:
|
||||
raise ValueError("Could not find voltage column in CSV data")
|
||||
|
||||
# Convert mV to volts
|
||||
voltage_mv = float(row[voltage_col])
|
||||
voltage_data.append(voltage_mv / 1000.0)
|
||||
|
||||
# Calculate time values from index and time interval
|
||||
index = int(row["index"])
|
||||
time_interval = self._extract_time_interval(metadata)
|
||||
time_data.append(index * time_interval)
|
||||
|
||||
# Create channel data
|
||||
channel_name = metadata.get("Channel", "CH1").replace(":", "").strip()
|
||||
|
||||
channel_data = ChannelData(
|
||||
channel_name=channel_name,
|
||||
voltage_values=np.array(voltage_data),
|
||||
time_values=np.array(time_data),
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
return ScopeData(channels={channel_name: channel_data}, metadata=metadata)
|
||||
|
||||
def _parse_metadata(self, metadata_lines: List[str]) -> Dict[str, str]:
|
||||
"""
|
||||
Parse metadata from header lines.
|
||||
|
||||
You can access the metadata like this:
|
||||
data = parse_owon_data(".CSV")
|
||||
print(data.frequency)
|
||||
print(data.vpp)
|
||||
"""
|
||||
metadata = {}
|
||||
|
||||
for line in metadata_lines:
|
||||
if "," in line:
|
||||
parts = line.split(",", 1)
|
||||
if len(parts) == 2:
|
||||
key = parts[0].strip().rstrip(":").strip()
|
||||
value = parts[1].strip()
|
||||
metadata[key] = value
|
||||
|
||||
return metadata
|
||||
19
scope_parser/parsers.py
Normal file
19
scope_parser/parsers.py
Normal file
@@ -0,0 +1,19 @@
|
||||
"""
|
||||
Main Parsers module
|
||||
"""
|
||||
|
||||
from .owon_parser import OwonParser
|
||||
from .gwinstek_parser import GwinstekParser
|
||||
from .data import ScopeData
|
||||
|
||||
|
||||
def parse_owon_data(file_path: str) -> ScopeData:
|
||||
"""Parse OWON oscilloscope CSV file."""
|
||||
parser = OwonParser()
|
||||
return parser.parse(file_path)
|
||||
|
||||
|
||||
def parse_gwinstek_data(file_path: str) -> ScopeData:
|
||||
"""Parse Gwinstek oscilloscope CSV file."""
|
||||
parser = GwinstekParser()
|
||||
return parser.parse(file_path)
|
||||
Reference in New Issue
Block a user