# Copyright (c) 2016-2024 The Regents of the University of Michigan # Part of GSD, released under the BSD 2-Clause License. """Test gsd.fl.""" import os import pathlib import platform import random import shutil import sys import numpy import pytest import gsd.fl import gsd.pygsd test_path = pathlib.Path(os.path.realpath(__file__)).parent current_gsd_version = (2, 1) def test_create(tmp_path, open_mode): """Test creation of GSD files.""" gsd.fl.open( mode=open_mode.write, name=tmp_path / 'test_create.gsd', application='test_create', schema='none', schema_version=[1, 2], ) @pytest.mark.parametrize( 'typ', [ numpy.uint8, numpy.uint16, numpy.uint32, numpy.uint64, numpy.int8, numpy.int16, numpy.int32, numpy.int64, numpy.float32, numpy.float64, ], ) def test_nonstring_dtypes(tmp_path, typ): """Test all supported data types except for strings.""" data1d = numpy.array([1, 2, 3, 4, 5, 127], dtype=typ) data2d = numpy.array([[10, 20], [30, 40], [50, 80]], dtype=typ) data_zero = numpy.array([], dtype=typ) gsd.fl.open( mode='x', name=tmp_path / 'test_dtype.gsd', application='test_dtype', schema='none', schema_version=[1, 2], ) with gsd.fl.open( name=tmp_path / 'test_dtype.gsd', mode='w', application='test_dtype', schema='none', schema_version=[1, 2], ) as f: f.write_chunk(name='data1d', data=data1d) f.write_chunk(name='data2d', data=data2d) f.write_chunk(name='data_zero', data=data_zero) f.end_frame() with gsd.fl.open( name=tmp_path / 'test_dtype.gsd', mode='r', application='test_dtype', schema='none', schema_version=[1, 2], ) as f: read_data1d = f.read_chunk(frame=0, name='data1d') read_data2d = f.read_chunk(frame=0, name='data2d') read_data_zero = f.read_chunk(frame=0, name='data_zero') assert data1d.dtype.type == read_data1d.dtype.type numpy.testing.assert_array_equal(data1d, read_data1d) assert data2d.dtype.type == read_data2d.dtype.type numpy.testing.assert_array_equal(data2d, read_data2d) assert data_zero.dtype.type == read_data_zero.dtype.type assert data_zero.shape == (0,) # test again with pygsd with gsd.pygsd.GSDFile(file=open(str(tmp_path / 'test_dtype.gsd'), mode='rb')) as f: read_data1d = f.read_chunk(frame=0, name='data1d') read_data2d = f.read_chunk(frame=0, name='data2d') assert data1d.dtype.type == read_data1d.dtype.type numpy.testing.assert_array_equal(data1d, read_data1d) assert data2d.dtype.type == read_data2d.dtype.type numpy.testing.assert_array_equal(data2d, read_data2d) def test_string_dtype(tmp_path): """Test string datatype. Note that the string datatype does not support 0-D or 2-D data. """ data1d = 'test' gsd.fl.open( mode='x', name=tmp_path / 'test_dtype.gsd', application='test_dtype', schema='none', schema_version=[1, 2], ) with gsd.fl.open( name=tmp_path / 'test_dtype.gsd', mode='w', application='test_dtype', schema='none', schema_version=[1, 2], ) as f: f.write_chunk(name='data1d', data=data1d) f.end_frame() with gsd.fl.open( name=tmp_path / 'test_dtype.gsd', mode='r', application='test_dtype', schema='none', schema_version=[1, 2], ) as f: read_data1d = f.read_chunk(frame=0, name='data1d') numpy.testing.assert_string_equal(data1d, read_data1d) # test again with pygsd with gsd.pygsd.GSDFile(file=open(str(tmp_path / 'test_dtype.gsd'), mode='rb')) as f: read_data1d = f.read_chunk(frame=0, name='data1d') numpy.testing.assert_string_equal(data1d, read_data1d) def test_metadata(tmp_path, open_mode): """Test file metadata.""" data = numpy.array([1, 2, 3, 4, 5, 10012], dtype=numpy.int64) with gsd.fl.open( name=tmp_path / 'test_metadata.gsd', mode=open_mode.write, application='test_metadata', schema='none', schema_version=[1, 2], ) as f: assert f.mode == open_mode.write for _i in range(150): f.write_chunk(name='data', data=data) f.end_frame() with gsd.fl.open( name=tmp_path / 'test_metadata.gsd', mode=open_mode.read, application='test_metadata', schema='none', schema_version=[1, 2], ) as f: assert f.name == str(tmp_path / 'test_metadata.gsd') assert f.mode == open_mode.read assert f.application == 'test_metadata' assert f.schema == 'none' assert f.schema_version == (1, 2) assert f.nframes == 150 assert f.gsd_version == current_gsd_version # test again with pygsd with gsd.pygsd.GSDFile( file=open(str(tmp_path / 'test_metadata.gsd'), mode='rb') ) as f: assert f.name == str(tmp_path / 'test_metadata.gsd') assert f.mode == 'r' assert f.application == 'test_metadata' assert f.schema == 'none' assert f.schema_version == (1, 2) assert f.nframes == 150 assert f.gsd_version == current_gsd_version def test_append(tmp_path, open_mode): """Test that data chunks can be appended to existing files.""" with gsd.fl.open( name=tmp_path / 'test_append.gsd', mode=open_mode.write, application='test_append', schema='none', schema_version=[1, 2], ): pass data = numpy.array([10], dtype=numpy.int64) nframes = 1024 with gsd.fl.open( name=tmp_path / 'test_append.gsd', mode='a', application='test_append', schema='none', schema_version=[1, 2], ) as f: assert f.mode == 'a' for i in range(nframes): data[0] = i f.write_chunk(name='data1', data=data) data[0] = i * 10 f.write_chunk(name='data10', data=data) f.end_frame() with gsd.fl.open( name=tmp_path / 'test_append.gsd', mode=open_mode.read, application='test_append', schema='none', schema_version=[1, 2], ) as f: assert f.nframes == nframes for i in range(nframes): data1 = f.read_chunk(frame=i, name='data1') data10 = f.read_chunk(frame=i, name='data10') assert data1[0] == i assert data10[0] == i * 10 # test again with pygsd with gsd.pygsd.GSDFile( file=open(str(tmp_path / 'test_append.gsd'), mode='rb') ) as f: assert f.nframes == nframes for i in range(nframes): data1 = f.read_chunk(frame=i, name='data1') data10 = f.read_chunk(frame=i, name='data10') assert data1[0] == i assert data10[0] == i * 10 def test_chunk_exists(tmp_path, open_mode): """Test the chunk_exists API.""" data = numpy.array([1, 2, 3, 4, 5, 10012], dtype=numpy.int64) with gsd.fl.open( name=tmp_path / 'test_chunk_exists.gsd', mode=open_mode.write, application='test_chunk_exists', schema='none', schema_version=[1, 2], ) as f: f.write_chunk(name='chunk1', data=data) f.end_frame() f.write_chunk(name='abcdefg', data=data) f.end_frame() f.write_chunk(name='test', data=data) f.end_frame() with gsd.fl.open( name=tmp_path / 'test_chunk_exists.gsd', mode=open_mode.read, application='test_chunk_exists', schema='none', schema_version=[1, 2], ) as f: assert f.chunk_exists(frame=0, name='chunk1') read_data = f.read_chunk(frame=0, name='chunk1') assert f.chunk_exists(frame=1, name='abcdefg') read_data = f.read_chunk(frame=1, name='abcdefg') assert f.chunk_exists(frame=2, name='test') read_data = f.read_chunk(frame=2, name='test') assert not f.chunk_exists(frame=1, name='chunk1') with pytest.raises(KeyError): read_data = f.read_chunk(frame=1, name='chunk1') assert not f.chunk_exists(frame=2, name='abcdefg') with pytest.raises(KeyError): read_data = f.read_chunk(frame=2, name='abcdefg') assert not f.chunk_exists(frame=0, name='test') with pytest.raises(KeyError): read_data = f.read_chunk(frame=0, name='test') assert not f.chunk_exists(frame=2, name='chunk1') with pytest.raises(KeyError): read_data = f.read_chunk(frame=2, name='chunk1') assert not f.chunk_exists(frame=0, name='abcdefg') with pytest.raises(KeyError): read_data = f.read_chunk(frame=0, name='abcdefg') assert not f.chunk_exists(frame=1, name='test') with pytest.raises(KeyError): read_data = f.read_chunk(frame=1, name='test') # test again with pygsd with gsd.pygsd.GSDFile( file=open(str(tmp_path / 'test_chunk_exists.gsd'), mode='rb') ) as f: assert f.chunk_exists(frame=0, name='chunk1') read_data = f.read_chunk(frame=0, name='chunk1') assert f.chunk_exists(frame=1, name='abcdefg') read_data = f.read_chunk(frame=1, name='abcdefg') assert f.chunk_exists(frame=2, name='test') read_data = f.read_chunk(frame=2, name='test') assert not f.chunk_exists(frame=1, name='chunk1') with pytest.raises(KeyError): read_data = f.read_chunk(frame=1, name='chunk1') assert not f.chunk_exists(frame=2, name='abcdefg') with pytest.raises(KeyError): read_data = f.read_chunk(frame=2, name='abcdefg') assert not f.chunk_exists(frame=0, name='test') with pytest.raises(KeyError): read_data = f.read_chunk(frame=0, name='test') assert not f.chunk_exists(frame=2, name='chunk1') with pytest.raises(KeyError): read_data = f.read_chunk(frame=2, name='chunk1') assert not f.chunk_exists(frame=0, name='abcdefg') with pytest.raises(KeyError): read_data = f.read_chunk(frame=0, name='abcdefg') assert not f.chunk_exists(frame=1, name='test') with pytest.raises(KeyError): read_data = f.read_chunk(frame=1, name='test') # noqa def test_readonly_errors(tmp_path, open_mode): """Test that read only files provide the appropriate errors.""" data = numpy.array([1, 2, 3, 4, 5, 10012], dtype=numpy.int64) with gsd.fl.open( name=tmp_path / 'test_readonly_errors.gsd', mode=open_mode.write, application='test_readonly_errors', schema='none', schema_version=[1, 2], ) as f: for _i in range(10): f.write_chunk(name='chunk1', data=data) f.end_frame() data = numpy.array([1, 2, 3, 4, 5, 10012], dtype=numpy.int64) with gsd.fl.open( name=tmp_path / 'test_readonly_errors.gsd', mode='r', application='test_readonly_errors', schema='none', schema_version=[1, 2], ) as f: with pytest.raises(RuntimeError): f.end_frame() with pytest.raises(RuntimeError): f.write_chunk(name='chunk1', data=data) # test again with pygsd with gsd.pygsd.GSDFile( file=open(str(tmp_path / 'test_readonly_errors.gsd'), mode='rb') ) as f: with pytest.raises(NotImplementedError): f.end_frame() with pytest.raises(NotImplementedError): f.write_chunk(name='chunk1', data=data) def test_fileio_errors(tmp_path, open_mode): """Test that OS file I/O errors pass through.""" # These test cause python to crash on windows.... if platform.system() != 'Windows': with pytest.raises(FileNotFoundError): gsd.fl.open( name='/this/file/does/not/exist', mode='r', application='test_readonly_errors', schema='none', schema_version=[1, 2], ) with open(str(tmp_path / 'test_fileio_errors.gsd'), 'wb') as f: f.write(b'test') with pytest.raises(RuntimeError): f = gsd.fl.open( name=tmp_path / 'test_fileio_errors.gsd', mode=open_mode.read, application='test_readonly_errors', schema='none', schema_version=[1, 2], ) def test_dtype_errors(tmp_path, open_mode): """Test that unsupported data types result in errors.""" data = numpy.array([1, 2, 3, 4, 5, 10012], dtype=numpy.bool_) with gsd.fl.open( name=tmp_path / 'test_dtype_errors1.gsd', mode=open_mode.write, application='test_dtype_errors', schema='none', schema_version=[1, 2], ) as f: with pytest.raises(ValueError): f.write_chunk(name='chunk1', data=data) f.end_frame() data = numpy.array([1, 2, 3, 4, 5, 10012], dtype=numpy.float16) with gsd.fl.open( name=tmp_path / 'test_dtype_errors2.gsd', mode=open_mode.write, application='test_dtype_errors', schema='none', schema_version=[1, 2], ) as f: with pytest.raises(ValueError): f.write_chunk(name='chunk1', data=data) f.end_frame() data = numpy.array([1, 2, 3, 4, 5, 10012], dtype=numpy.complex64) with gsd.fl.open( name=tmp_path / 'test_dtype_errors3.gsd', mode=open_mode.write, application='test_dtype_errors', schema='none', schema_version=[1, 2], ) as f: with pytest.raises(ValueError): f.write_chunk(name='chunk1', data=data) f.end_frame() data = numpy.array([1, 2, 3, 4, 5, 10012], dtype=numpy.complex128) with gsd.fl.open( name=tmp_path / 'test_dtype_errors4.gsd', mode=open_mode.write, application='test_dtype_errors', schema='none', schema_version=[1, 2], ) as f: with pytest.raises(ValueError): f.write_chunk(name='chunk1', data=data) f.end_frame() def test_truncate(tmp_path): """Test that the truncate method functions.""" rng = numpy.random.default_rng() data = numpy.ascontiguousarray(rng.random(size=(1000, 3)), dtype=numpy.float32) with gsd.fl.open( name=tmp_path / 'test_truncate.gsd', mode='w', application='test_truncate', schema='none', schema_version=[1, 2], ) as f: assert f.mode == 'w' for _i in range(10): f.write_chunk(name='data', data=data) f.end_frame() assert f.nframes == 10 f.truncate() assert f.nframes == 0 assert f.application == 'test_truncate' assert f.schema == 'none' assert f.schema_version == (1, 2) f.write_chunk(name='data', data=data) f.end_frame() with gsd.fl.open( name=tmp_path / 'test_truncate.gsd', mode='r', application='test_truncate', schema='none', schema_version=[1, 2], ) as f: assert f.name == str(tmp_path / 'test_truncate.gsd') assert f.mode == 'r' assert f.application == 'test_truncate' assert f.schema == 'none' assert f.schema_version == (1, 2) assert f.nframes == 1 def test_namelen(tmp_path, open_mode): """Test that long names are truncated as documented.""" app_long = 'abcdefga' * 100 schema_long = 'ijklmnop' * 100 chunk_long = '12345678' * 100 with gsd.fl.open( name=tmp_path / 'test_namelen.gsd', mode=open_mode.write, application=app_long, schema=schema_long, schema_version=[1, 2], ) as f: assert f.application == app_long[0:63] assert f.schema == schema_long[0:63] data = numpy.array([1, 2, 3, 4, 5, 10012], dtype=numpy.int64) f.write_chunk(name=chunk_long, data=data) f.end_frame() with gsd.fl.open( name=tmp_path / 'test_namelen.gsd', mode=open_mode.read, application=app_long, schema=schema_long, schema_version=[1, 2], ) as f: data_read = f.read_chunk(0, name=chunk_long) numpy.testing.assert_array_equal(data, data_read) # test again with pygsd with gsd.pygsd.GSDFile( file=open(str(tmp_path / 'test_namelen.gsd'), mode='rb') ) as f: data_read = f.read_chunk(0, name=chunk_long) numpy.testing.assert_array_equal(data, data_read) def test_open(tmp_path): """Test the open() API.""" data = numpy.array([1, 2, 3, 4, 5, 10012], dtype=numpy.int64) with gsd.fl.open( name=tmp_path / 'test.gsd', mode='x', application='test_open', schema='none', schema_version=[1, 2], ) as f: f.write_chunk(name='chunk1', data=data) f.end_frame() with gsd.fl.open( name=tmp_path / 'test_2.gsd', mode='x', application='test_open', schema='none', schema_version=[1, 2], ) as f: f.write_chunk(name='chunk1', data=data) f.end_frame() f.read_chunk(0, name='chunk1') with gsd.fl.open( name=tmp_path / 'test.gsd', mode='w', application='test_open', schema='none', schema_version=[1, 2], ) as f: f.write_chunk(name='chunk1', data=data) f.end_frame() with gsd.fl.open( name=tmp_path / 'test.gsd', mode='w', application='test_open', schema='none', schema_version=[1, 2], ) as f: f.write_chunk(name='chunk1', data=data) f.end_frame() f.read_chunk(0, name='chunk1') with gsd.fl.open( name=tmp_path / 'test.gsd', mode='a', application='test_open', schema='none', schema_version=[1, 2], ) as f: f.write_chunk(name='chunk1', data=data) f.end_frame() with gsd.fl.open( name=tmp_path / 'test.gsd', mode='r', application='test_open', schema='none', schema_version=[1, 2], ) as f: f.read_chunk(0, name='chunk1') f.read_chunk(1, name='chunk1') with gsd.fl.open( name=tmp_path / 'test.gsd', mode='r+', application='test_open', schema='none', schema_version=[1, 2], ) as f: f.write_chunk(name='chunk1', data=data) f.end_frame() f.read_chunk(0, name='chunk1') f.read_chunk(1, name='chunk1') f.read_chunk(2, name='chunk1') def test_find_matching_chunk_names(tmp_path, open_mode): """Test the find_matching_chunk_names API.""" data = numpy.array([1, 2, 3, 4, 5], dtype=numpy.float32) with gsd.fl.open( name=tmp_path / 'test.gsd', mode=open_mode.write, application='test_find_matching_chunk_names', schema='none', schema_version=[1, 2], ) as f: f.write_chunk(name='log/A', data=data) f.write_chunk(name='log/chunk2', data=data) f.end_frame() f.write_chunk(name='data/B', data=data) f.end_frame() with gsd.fl.open( name=tmp_path / 'test.gsd', mode=open_mode.read, application='test_find_matching_chunk_names', schema='none', schema_version=[1, 2], ) as f: all_chunks = f.find_matching_chunk_names('') assert len(all_chunks) == 3 assert 'log/A' in all_chunks assert 'log/chunk2' in all_chunks assert 'data/B' in all_chunks log_chunks = f.find_matching_chunk_names('log/') assert len(log_chunks) == 2 assert 'log/A' in log_chunks assert 'log/chunk2' in log_chunks data_chunks = f.find_matching_chunk_names('data/') assert len(data_chunks) == 1 assert 'data/B' in data_chunks other_chunks = f.find_matching_chunk_names('other/') assert len(other_chunks) == 0 # test again with pygsd with gsd.pygsd.GSDFile(file=open(str(tmp_path / 'test.gsd'), mode='rb')) as f: all_chunks = f.find_matching_chunk_names('') assert len(all_chunks) == 3 assert 'log/A' in all_chunks assert 'log/chunk2' in all_chunks assert 'data/B' in all_chunks log_chunks = f.find_matching_chunk_names('log/') assert len(log_chunks) == 2 assert 'log/A' in log_chunks assert 'log/chunk2' in log_chunks data_chunks = f.find_matching_chunk_names('data/') assert len(data_chunks) == 1 assert 'data/B' in data_chunks other_chunks = f.find_matching_chunk_names('other/') assert len(other_chunks) == 0 def test_chunk_name_limit(tmp_path, open_mode): """Test that providing more than the maximum allowed chunk names errors.""" with gsd.fl.open( name=tmp_path / 'test.gsd', mode=open_mode.write, application='test_chunk_name_limit', schema='none', schema_version=[1, 2], ) as f: for i in range(65535): f.write_chunk(name=str(i), data=numpy.array([i], dtype=numpy.int32)) # The GSD specification limits to 65535 names: with pytest.raises(RuntimeError): f.write_chunk(name='65536', data=numpy.array([i], dtype=numpy.int32)) def test_many_names(tmp_path, open_mode): """Test that many chunk names can be written to a file.""" values = list(range(1000)) with gsd.fl.open( name=tmp_path / 'test.gsd', mode=open_mode.write, application='test_many_names', schema='none', schema_version=[1, 2], ) as f: for _ in range(5): random.shuffle(values) for value in values: f.write_chunk( name=str(value), data=numpy.array([value * 13], dtype=numpy.int32) ) f.end_frame() with gsd.fl.open( name=tmp_path / 'test.gsd', mode=open_mode.read, application='test_many_names', schema='none', schema_version=[1, 2], ) as f: for frame in range(5): random.shuffle(values) for value in values: data = numpy.array([value * 13], dtype=numpy.int32) data_read = f.read_chunk(frame=frame, name=str(value)) numpy.testing.assert_array_equal(data, data_read) with gsd.pygsd.GSDFile(file=open(str(tmp_path / 'test.gsd'), mode='rb')) as f: for frame in range(5): random.shuffle(values) for value in values: data = numpy.array([value * 13], dtype=numpy.int32) data_read = f.read_chunk(frame=frame, name=str(value)) numpy.testing.assert_array_equal(data, data_read) def test_gsd_v1_read(): """Test that the GSD v2 API can read v1 files.""" values = list(range(127)) values_str = [str(v) for v in values] values_str.sort() # test that we can: # 1) Read chunk values correctly # 2) Iterate through chunk names correctly def check_v1_file_read(f): assert f.gsd_version == (1, 0) for frame in range(5): random.shuffle(values) for value in values: data = numpy.array([value * 13], dtype=numpy.int32) data_read = f.read_chunk(frame=frame, name=str(value)) numpy.testing.assert_array_equal(data, data_read) chunk_names = f.find_matching_chunk_names('') chunk_names.sort() assert chunk_names == values_str # test with the C implemantation with gsd.fl.open( name=test_path / 'test_gsd_v1.gsd', mode='r', application='test_gsd_v1', schema='none', schema_version=[1, 2], ) as f: check_v1_file_read(f) # and the pure Python implementation with gsd.pygsd.GSDFile( file=open(str(test_path / 'test_gsd_v1.gsd'), mode='rb') ) as f: assert f.gsd_version == (1, 0) check_v1_file_read(f) def test_gsd_v1_upgrade_read(tmp_path, open_mode): """Test that v1 files can be upgraded to v2.""" values = list(range(127)) values_str = [str(v) for v in values] values_str.sort() # test that we can: # 1) Read chunk values correctly # 2) Iterate through chunk names correctly def check_v1_file_read(f): for frame in range(5): random.shuffle(values) for value in values: data = numpy.array([value * 13], dtype=numpy.int32) data_read = f.read_chunk(frame=frame, name=str(value)) numpy.testing.assert_array_equal(data, data_read) chunk_names = f.find_matching_chunk_names('') chunk_names.sort() assert chunk_names == values_str shutil.copy(test_path / 'test_gsd_v1.gsd', tmp_path / 'test_gsd_v1.gsd') with gsd.fl.open( name=tmp_path / 'test_gsd_v1.gsd', mode='r+', application='test_gsd_v1', schema='none', schema_version=[1, 2], ) as f: assert f.gsd_version == (1, 0) f.upgrade() # check that we can read the file contents after the upgrade in memory check_v1_file_read(f) # and the same tests again after closing and opening the file with gsd.fl.open( name=tmp_path / 'test_gsd_v1.gsd', mode=open_mode.read, application='test_gsd_v1', schema='none', schema_version=[1, 2], ) as f: assert f.gsd_version == current_gsd_version check_v1_file_read(f) with gsd.pygsd.GSDFile( file=open(str(tmp_path / 'test_gsd_v1.gsd'), mode='rb') ) as f: assert f.gsd_version == current_gsd_version check_v1_file_read(f) def test_gsd_v1_write(tmp_path, open_mode): """Test that v2 can write to v1 files.""" values = list(range(256)) # include a very long chunk name to check that the name is truncated # properly for the v1 format limitations long_name = 'abcdefg' * 1000 values.append(long_name) values_str = [] for v in values: check_v = v if isinstance(v, str) and len(v) > 63: # v1 files truncate names to 63 chars check_v = v[0:63] values_str.append(str(check_v)) values_str.sort() shutil.copy(test_path / 'test_gsd_v1.gsd', tmp_path / 'test_gsd_v1.gsd') # test that we can: # 1) Read chunk values correctly # 2) Iterate through chunk names correctly def check_v1_file_read(f): assert f.gsd_version == (1, 0) chunk_names = f.find_matching_chunk_names('') chunk_names.sort() assert chunk_names == values_str frame = 5 random.shuffle(values) for value in values: check_value = value if isinstance(value, int): data = numpy.array([value * 13], dtype=numpy.int32) else: data = numpy.array([hash(value)], dtype=numpy.int64) # v1 files truncate names to 63 chars if len(value) > 63: check_value = value[0:63] data_read = f.read_chunk(frame=frame, name=str(check_value)) numpy.testing.assert_array_equal(data, data_read) # test that we can write new entries to the file with gsd.fl.open( name=tmp_path / 'test_gsd_v1.gsd', mode='r+', application='test_gsd_v1', schema='none', schema_version=[1, 2], ) as f: assert f.gsd_version == (1, 0) for value in values: if isinstance(value, int): data = numpy.array([value * 13], dtype=numpy.int32) else: data = numpy.array([hash(value)], dtype=numpy.int64) f.write_chunk(name=str(value), data=data) f.end_frame() check_v1_file_read(f) # test opening again with the C implemantation with gsd.fl.open( name=tmp_path / 'test_gsd_v1.gsd', mode=open_mode.read, application='test_gsd_v1', schema='none', schema_version=[1, 2], ) as f: check_v1_file_read(f) # and the pure Python implementation with gsd.pygsd.GSDFile( file=open(str(tmp_path / 'test_gsd_v1.gsd'), mode='rb') ) as f: assert f.gsd_version == (1, 0) check_v1_file_read(f) def test_gsd_v1_upgrade_write(tmp_path, open_mode): """Test that upgraded files can be written to after upgraded.""" values = list(range(256)) # include a very long chunk name to check that the name can be written # after the upgrade long_name = 'abcdefg' * 1000 values.append(long_name) values_str = [str(v) for v in values] values_str.sort() shutil.copy(test_path / 'test_gsd_v1.gsd', tmp_path / 'test_gsd_v1.gsd') # test that we can: # 1) Read chunk values correctly # 2) Iterate through chunk names correctly def check_v1_file_read(f): chunk_names = f.find_matching_chunk_names('') chunk_names.sort() assert chunk_names == values_str frame = 5 random.shuffle(values) for value in values: if isinstance(value, int): data = numpy.array([value * 13], dtype=numpy.int32) else: data = numpy.array([hash(value)], dtype=numpy.int64) data_read = f.read_chunk(frame=frame, name=str(value)) numpy.testing.assert_array_equal(data, data_read) # test that we can write new entries to the file with gsd.fl.open( name=tmp_path / 'test_gsd_v1.gsd', mode='r+', application='test_gsd_v1', schema='none', schema_version=[1, 2], ) as f: assert f.gsd_version == (1, 0) f.upgrade() assert f.gsd_version == current_gsd_version for value in values: if isinstance(value, int): data = numpy.array([value * 13], dtype=numpy.int32) else: data = numpy.array([hash(value)], dtype=numpy.int64) f.write_chunk(name=str(value), data=data) f.end_frame() check_v1_file_read(f) # test opening again with the C implemantation with gsd.fl.open( name=tmp_path / 'test_gsd_v1.gsd', mode=open_mode.read, application='test_gsd_v1', schema='none', schema_version=[1, 2], ) as f: assert f.gsd_version == current_gsd_version check_v1_file_read(f) # and the pure Python implementation with gsd.pygsd.GSDFile( file=open(str(tmp_path / 'test_gsd_v1.gsd'), mode='rb') ) as f: assert f.gsd_version == current_gsd_version check_v1_file_read(f) def test_zero_size(tmp_path, open_mode): """Test that zero-size data chunks are allowed.""" data = numpy.array([], dtype=numpy.float32) with gsd.fl.open( name=tmp_path / 'test_zero.gsd', mode=open_mode.write, application='test_zero', schema='none', schema_version=[1, 2], ) as f: f.write_chunk(name='data', data=data) f.end_frame() with gsd.fl.open( name=tmp_path / 'test_zero.gsd', mode=open_mode.read, application='test_zero', schema='none', schema_version=[1, 2], ) as f: assert f.nframes == 1 data_read = f.read_chunk(frame=0, name='data') assert data_read.shape == (0,) assert data_read.dtype == numpy.float32 # test again with pygsd with gsd.pygsd.GSDFile(file=open(str(tmp_path / 'test_zero.gsd'), mode='rb')) as f: assert f.nframes == 1 data_read = f.read_chunk(frame=0, name='data') assert data_read.shape == (0,) assert data_read.dtype == numpy.float32 @pytest.mark.skipif( sys.version_info < (3, 7), reason='Python 3.6 fails to handle non-ascii characters.' ) def test_utf8(tmp_path): """Test that the API handles UTF-8 encoding for the filename.""" data = numpy.array([1, 2, 3, 4, 5, 10012], dtype=numpy.int64) fname = '中文.gsd' with gsd.fl.open( name=tmp_path / fname, mode='x', application='test_open', schema='none', schema_version=[1, 2], ) as f: f.write_chunk(name='chunk1', data=data) f.end_frame() dir_list = os.listdir(tmp_path) print(dir_list) assert fname in dir_list with gsd.fl.open( name=tmp_path / fname, mode='w', application='test_open', schema='none', schema_version=[1, 2], ) as f: f.write_chunk(name='chunk1', data=data) f.end_frame() with gsd.fl.open( name=tmp_path / fname, mode='r', application='test_open', schema='none', schema_version=[1, 2], ) as f: f.read_chunk(0, name='chunk1') @pytest.mark.parametrize('mode', ['w', 'x', 'a', 'r+']) def test_read_write(tmp_path, mode): """Test that data chunks can read from files opened in all write modes.""" if mode[0] == 'r' or mode[0] == 'a': with gsd.fl.open( name=tmp_path / 'test_read_write.gsd', mode='w', application='test_read_write', schema='none', schema_version=[1, 2], ): pass data = numpy.array([10], dtype=numpy.int64) nframes = 1024 with gsd.fl.open( name=tmp_path / 'test_read_write.gsd', mode=mode, application='test_read_write', schema='none', schema_version=[1, 2], ) as f: assert f.mode == mode for i in range(nframes): data[0] = i f.write_chunk(name='data1', data=data) data[0] = i * 10 f.write_chunk(name='data10', data=data) f.end_frame() for i in range(nframes): data1 = f.read_chunk(frame=i, name='data1') data10 = f.read_chunk(frame=i, name='data10') assert data1[0] == i assert data10[0] == i * 10 @pytest.mark.parametrize('n_flush', [0, 1, 2]) def test_flush(tmp_path, open_mode, n_flush): """Test flush.""" data = numpy.array([1, 2, 3, 4, 5, 10012], dtype=numpy.int64) with gsd.fl.open( name=tmp_path / 'test_flush.gsd', mode=open_mode.write, application='test_flush', schema='none', schema_version=[1, 2], ) as f: f.write_chunk(name='chunk1', data=data) f.end_frame() f.write_chunk(name='chunk2', data=data) f.end_frame() f.write_chunk(name='chunk3', data=data) # Ensure that the data is buffered by opening the file with a 2nd # handle read-only and checking it. with gsd.fl.open(name=tmp_path / 'test_flush.gsd', mode='r') as f_readonly: assert not f_readonly.chunk_exists(frame=0, name='chunk1') assert not f_readonly.chunk_exists(frame=1, name='chunk2') assert f_readonly.nframes == 0 # 0 calls to flush tests the implicit flush on close, 2 calls to flush # tests that repeated calls are handled properly. for _i in range(n_flush): f.flush() with gsd.fl.open(name=tmp_path / 'test_flush.gsd', mode=open_mode.read) as f: assert f.chunk_exists(frame=0, name='chunk1') assert f.chunk_exists(frame=1, name='chunk2') # The third chunk is not written because end_frame is not called a # third time. assert not f.chunk_exists(frame=2, name='chunk3') assert f.nframes == 2 def test_maximum_write_buffer_size(tmp_path, open_mode): """Test maximum_write_buffer_size.""" with gsd.fl.open( name=tmp_path / 'test_maximum_write_buffer_size.gsd', mode=open_mode.write, application='test_maximum_write_buffer_size', schema='none', schema_version=[1, 2], ) as f: assert f.maximum_write_buffer_size > 0 f.maximum_write_buffer_size = 1024 assert f.maximum_write_buffer_size == 1024 with pytest.raises(RuntimeError): f.maximum_write_buffer_size = 0 def test_index_entries_to_buffer(tmp_path, open_mode): """Test index_entries_to_buffer.""" with gsd.fl.open( name=tmp_path / 'test_index_entries_to_buffer.gsd', mode=open_mode.write, application='test_index_entries_to_buffer', schema='none', schema_version=[1, 2], ) as f: assert f.index_entries_to_buffer > 0 f.index_entries_to_buffer = 1024 assert f.index_entries_to_buffer == 1024 with pytest.raises(RuntimeError): f.index_entries_to_buffer = 0 def test_file_exists_error(): """Test that IO errors throw the correct Python Excetion.""" with pytest.raises(FileExistsError): with gsd.fl.open( name=test_path / 'test_gsd_v1.gsd', mode='x', application='test_gsd_v1', schema='none', schema_version=[1, 2], ): pass def test_pending_index_entries(tmp_path): """Ensure that gsd preserves pending index entries.""" with gsd.fl.open( tmp_path / 'test_pending_index_entries.gsd', 'w', application='My application', schema='My Schema', schema_version=[1, 0], ) as f: # Frame 0 must be complete to trigger the bug. f.write_chunk(name='0', data=numpy.array([0])) f.end_frame() for i in range(16): f.write_chunk(name=str(i), data=numpy.array([i])) # Flush with pending chunks in the frame index. f.flush() f.end_frame() # All test chunks should be present in the file. for i in range(16): assert f.chunk_exists(name=str(i), frame=1)