diff --git a/psf_utils/parse.py b/psf_utils/parse.py index 720fcbe..80a2370 100644 --- a/psf_utils/parse.py +++ b/psf_utils/parse.py @@ -23,6 +23,7 @@ import ply.lex import ply.yacc from inform import Info, is_str, is_mapping +import numpy as np # Globals {{{1 @@ -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 @@ -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): @@ -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" diff --git a/psf_utils/psf.py b/psf_utils/psf.py index 23f675c..3df5d67 100644 --- a/psf_utils/psf.py +++ b/psf_utils/psf.py @@ -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: @@ -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 @@ -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 @@ -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 = '' @@ -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, @@ -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] @@ -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] @@ -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 diff --git a/tests/test_psf.py b/tests/test_psf.py index 3fcc3b6..f57923b 100644 --- a/tests/test_psf.py +++ b/tests/test_psf.py @@ -195,3 +195,97 @@ def test_utils(name, command, psf_file, arguments, expected): # remove svg_file if it was created rm(svg_file) + +# Error Handling Tests {{{1 +def test_binary_psf_error(): + """Test that binary PSF files raise appropriate error""" + test_dir = Path(__file__).parent + binary_file = test_dir / "binary_test.psf" + + # Create a binary file + binary_file.write_bytes(b'\x00\xff\xfe\xfd' * 100) + + try: + with pytest.raises(Exception) as exc_info: + PSF(binary_file) + # Should raise Error about binary PSF + assert "binary PSF" in str(exc_info.value) or "UnicodeDecodeError" in str(type(exc_info.value)) + finally: + # Clean up + rm(binary_file) + +def test_missing_file_error(): + """Test that missing files raise appropriate error""" + from inform import Error + with pytest.raises(Error): + PSF("nonexistent_file_12345.psf") + +def test_malformed_psf_error(): + """Test that malformed PSF files raise appropriate error""" + from inform import Error + test_dir = Path(__file__).parent + malformed_file = test_dir / "malformed_test.psf" + + # Create a malformed PSF file + malformed_file.write_text("HEADER\nGARBAGE DATA!!!\nMORE GARBAGE\n") + + try: + with pytest.raises(Error): + PSF(malformed_file) + finally: + # Clean up + rm(malformed_file) + +def test_corrupted_cache(): + """Test that corrupted cache files are handled gracefully""" + test_dir = Path(__file__).parent + psf_file = test_dir / "../samples/pnoise.raw/aclin.ac" + cache_file = psf_file.with_suffix(".ac.cache") + + # Remove existing cache + rm(cache_file) + + # Create corrupted cache + cache_file.write_bytes(b"corrupted cache data\x00\xff") + + try: + # Should fall back to parsing PSF file + psf = PSF(psf_file) + assert psf is not None + assert psf.get_sweep() is not None + finally: + # Clean up + rm(cache_file) + +# Static Method Tests {{{1 +def test_unknown_signal(): + """Test that accessing unknown signal raises UnknownSignal""" + from psf_utils.psf import UnknownSignal + test_dir = Path(__file__).parent + psf_file = test_dir / "../samples/pnoise.raw/aclin.ac" + + # Clean cache + cache_file = psf_file.with_suffix(".ac.cache") + rm(cache_file) + + try: + psf = PSF(psf_file) + with pytest.raises(UnknownSignal): + psf.get_signal("nonexistent_signal_xyz_12345") + finally: + rm(cache_file) + +def test_units_to_unicode(): + """Test units conversion to unicode""" + assert PSF.units_to_unicode("Ohm") == "Ω" + assert PSF.units_to_unicode("V^2") == "V²" + assert PSF.units_to_unicode("sqrt(Hz)") == "√Hz" + assert PSF.units_to_unicode("R") == "Ω" # Resistance + assert PSF.units_to_unicode("") == "" + assert PSF.units_to_unicode(None) == "" + +def test_units_to_latex(): + """Test units conversion to latex (not implemented)""" + result = PSF.units_to_latex("V/A") + # Currently returns input unchanged + assert result == "V/A"