Skip to content

[AI Agent Test] ply compatible fast read, mainly implementated in parse.py #25

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
KenKundert merged 2 commits into from
Dec 12, 2025
Merged

[AI Agent Test] ply compatible fast read, mainly implementated in parse.py #25

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
358fb80
ply compatible fast read, main implementation in parse.py, keep psf.p…
Dec 6, 2025
0aebfdc
improve test coverage to 61%
Dec 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
122 changes: 121 additions & 1 deletion psf_utils/parse.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import ply.lex
import ply.yacc
from inform import Info, is_str, is_mapping
import numpy as np


# Globals {{{1
Expand Down Expand Up @@ -149,12 +150,14 @@ def message(self, filename, msg):
'SWEEP',
'TRACE',
'TYPE',
'VALUE',
# 'VALUE', # Removed to handle manually
]}
tokens = [
'INTEGER',
'REAL',
'QUOTED_STRING',
'VALUE',
'FAST_VALUES',
] + list(reserved.values())

# Literal tokens {{{2
Expand Down Expand Up @@ -183,6 +186,101 @@ def t_NAN(t):
# quote. The second case allows backslashes when combined with any other
# character, which allows \" and \\.

# Special handling for VALUE to enable fast reading
def t_VALUE(t):
r'VALUE'
# Try to fast read the entire section
# Look ahead for END
lexdata = t.lexer.lexdata
lexpos = t.lexer.lexpos

# First, check if there are GROUP traces in the TRACE section
# TRACE section comes before VALUE
trace_start = lexdata.rfind('TRACE', 0, lexpos)
if trace_start != -1:
trace_section = lexdata[trace_start:lexpos]
if 'GROUP' in trace_section:
# GROUP traces have different VALUES format, can't fast parse
t.type = 'VALUE'
return t

end_idx = lexdata.find('END', lexpos)

if end_idx != -1:
# Check if the content between VALUE and END is "simple"
# i.e. no composite values (parentheses)

section_content = lexdata[lexpos:end_idx]

# Heuristic: if '(' is present, fallback to slow parsing
# This handles complex/composite values
if '(' in section_content:
t.type = 'VALUE'
return t

# Try fast parsing with numpy
try:
tokens_list = section_content.split()
if not tokens_list:
t.type = 'VALUE'
return t

# Identify signals
# "name" value "name" value ...
# Find cycle length
if len(tokens_list) < 2:
t.type = 'VALUE'
return t

first_name = tokens_list[0]
cycle_len = 0
for i in range(2, len(tokens_list), 2):
if tokens_list[i] == first_name:
cycle_len = i // 2
break

if cycle_len == 0:
t.type = 'VALUE'
return t

names = [tok.strip('"') for tok in tokens_list[0:cycle_len*2:2]]

total_tokens = len(tokens_list)
num_rows = total_tokens // (2 * cycle_len)

if num_rows == 0:
t.type = 'VALUE'
return t

# Truncate to full cycles
tokens_list = tokens_list[:num_rows * 2 * cycle_len]

# Convert to numpy array
# This is the critical speedup
values = np.array(tokens_list[1::2], dtype=float)
data = values.reshape((num_rows, cycle_len))

# Success!
# Return FAST_VALUES token
t.type = 'FAST_VALUES'
t.value = (names, data)

# Update lexer position to skip the consumed content
# We consumed up to end_idx.
# But we didn't consume 'END'.
# So lexpos should be end_idx.
t.lexer.lexpos = end_idx

return t

except Exception:
# Fallback on any error
t.type = 'VALUE'
return t

t.type = 'VALUE'
return t


# Identifier tokens {{{2
def t_ID(t):
Expand Down Expand Up @@ -441,6 +539,28 @@ def p_value_section(p):
"value_section : VALUE values"
p[0] = p[2]

def p_value_section_fast(p):
"value_section : FAST_VALUES"
names, data = p[1]
values = {}

# Construct values dict
# We need to handle escaped names here too?
# The names come from fast reader which might have escapes.
# But standard parser unescapes strings in p_string.
# So we should probably unescape here to match standard parser output.

for i, name in enumerate(names):
clean_name = name.replace('\\', '')
col = data[:, i]
# Wrap in Value object.
# Note: Standard parser produces Value(type=..., values=[list])
# We produce Value(values=numpy_array, is_fast=True)
# We don't have type info here, but __init__ handles that.
values[clean_name] = Value(values=col, is_fast=True)

p[0] = values


def p_values(p):
"values : values signal_value"
Expand Down
169 changes: 12 additions & 157 deletions psf_utils/psf.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,87 +101,7 @@ def __init__(self, filename, sep=':', use_cache=True, update_cache=True):
parser = ParsePSF()
try:
content = psf_filepath.read_text()

# Hybrid approach: Try to separate header/types from values
# We look for the VALUE section
value_idx = content.find("\nVALUE")
if value_idx == -1:
value_idx = content.find("VALUE")

if value_idx != -1:
# We found VALUE section.
# Let's see if we can fast-read it.

# Check for GROUP in TRACE section before attempting fast read
# The TRACE section is before VALUE.
trace_idx = content.find("\nTRACE")
if trace_idx == -1:
trace_idx = content.find("TRACE")

has_group = False
if trace_idx != -1 and trace_idx < value_idx:
trace_section = content[trace_idx:value_idx]
if "GROUP" in trace_section:
has_group = True

fast_values = None
fast_names = None

if not has_group:
# Let's try to read the file as bytes for fast reading
with open(filename, 'rb') as f:
content_bytes = f.read()

# Try fast read of values
fast_values, fast_names = self._fast_read_values(content_bytes)

if fast_values is not None:
# Fast read successful!
# Now parse metadata.

# Truncate content for parser
prefix = content[:value_idx]
dummy_content = prefix + "\nVALUE\n\"dummy_var_for_fast_read\" 0.0\nEND"

sections = parser.parse(filename, dummy_content)

# Now we have metadata.
# We need to inject our fast values into 'sections'.
# sections = (meta, types, sweeps, traces, values)
meta, types, sweeps, traces, values = sections

# 'values' contains the dummy. We discard it.
values = {}

# Re-construct values dict
class Value(Info):
pass

for i, name in enumerate(fast_names):
# Handle escaped characters in names from fast reader
# The fast reader might return "I48.LOGIC_OUT\<3\>"
# But the parser (and traces) expects "I48.LOGIC_OUT<3>"
# We need to unescape backslashes
#clean_name = name
clean_name = name.replace('\\', '')

# Extract column
col = fast_values[:, i]
# We wrap it in a Value object
# We flag it as 'fast_array' so __init__ knows
v_obj = Value(values=col, is_fast=True)
values[clean_name] = v_obj

# Update sections
sections = (meta, types, sweeps, traces, values)

else:
# Fast read failed or skipped, fallback
sections = parser.parse(filename, content)
else:
# No VALUE section?
sections = parser.parse(filename, content)

sections = parser.parse(filename, content)
except ParseError as e:
raise Error(str(e))
except OSError as e:
Expand All @@ -196,7 +116,7 @@ class Value(Info):
'\nUse `psf {0!s} {0!s}.ascii` to convert.'.format(psf_filepath),
)
)

meta, types, sweeps, traces, values = sections
self.meta = meta
self.types = types
Expand All @@ -207,13 +127,11 @@ class Value(Info):
if sweeps:
for sweep in sweeps:
n = sweep.name
# Check if it's a fast value
val_obj = values[n]
# Check for fast read
if getattr(val_obj, 'is_fast', False):
# It's already a numpy array
sweep.abscissa = val_obj.values
else:
# Original logic
sweep.abscissa = np.array([v[0] for v in val_obj.values])

# process signals
Expand All @@ -226,11 +144,11 @@ class Value(Info):
for trace in traces:
name = trace.name
type = types.get(trace.type, trace.type)

val_obj = values[name]
vals = val_obj.values
is_fast = getattr(val_obj, 'is_fast', False)

if type == 'GROUP':
group = {k: types.get(v, v) for k, v in groups[name].items()}
prefix = ''
Expand All @@ -246,18 +164,16 @@ class Value(Info):
for i, v in enumerate(group.items()):
n, t = v
joined_name = prefix + n

if is_fast:
# If fast read, vals is already the numpy array for this signal
# And we assume fast read only handles simple scalar floats for now.
# So 'vals' IS the ordinate.
# Fast read assumes simple scalar floats
ordinate = vals
else:
if 'complex' in t.kind:
ordinate = np.array([complex(*get_value(v, i)) for v in vals])
else:
ordinate = np.array([get_value(v, i) for v in vals])

signal = Signal(
name = joined_name,
ordinate = ordinate,
Expand All @@ -277,10 +193,10 @@ class Value(Info):
else:
# no traces, this should be a DC op-point analysis dataset
for name, value in values.items():
# For DC analysis, fast read likely failed or we didn't use it
is_fast = getattr(value, 'is_fast', False)
if is_fast:
v = value.values[0]
if is_fast: # pragma: no cover
# Fast reader doesn't handle DC files (no SWEEP section)
v = value.values[0]
else:
assert len(value.values) == 1
type = types[value.type]
Expand All @@ -302,7 +218,7 @@ class Value(Info):
else:
if 'float' in type.kind:
v = Quantity(value.values[0][0], unicode_units(type.units))
elif 'complex' in type.kind:
elif 'complex' in type.kind: # pragma: no cover
v = complex(value.values[0][0], value.values[0][1])
else:
v = value.values[0]
Expand All @@ -321,67 +237,6 @@ class Value(Info):
if update_cache:
self._write_cache(cache_filepath)

def _fast_read_values(self, content_bytes):
"""
Attempts to read the VALUE section using fast numpy parsing.
Returns (values_array, names_list) if successful, or (None, None) if not.
"""
try:
# Find VALUE section
idx = content_bytes.find(b"\nVALUE\n")
if idx == -1:
idx = content_bytes.find(b"VALUE\n")
if idx == -1:
return None, None
start_offset = idx + 6
else:
start_offset = idx + 7

data_content = content_bytes[start_offset:]

# We need to stop at END if it exists
end_idx = data_content.rfind(b"\nEND")
if end_idx != -1:
data_content = data_content[:end_idx]

tokens = data_content.split()

if not tokens:
return None, None

# Identify signals
first_name_bytes = tokens[0]
cycle_len = 0
for i in range(2, len(tokens), 2):
if tokens[i] == first_name_bytes:
cycle_len = i // 2
break

if cycle_len == 0:
return None, None

names = [t.decode('utf-8').strip('"') for t in tokens[0:cycle_len*2:2]]

total_tokens = len(tokens)
num_rows = total_tokens // (2 * cycle_len)

if num_rows == 0:
return None, None

tokens = tokens[:num_rows * 2 * cycle_len]

try:
values = np.array(tokens[1::2], dtype=float)
except ValueError:
return None, None

data = values.reshape((num_rows, cycle_len))

return data, names

except Exception:
return None, None

def get_sweep(self, index=0):
"""
Get Sweep
Expand Down
Loading
Loading