mirror of
https://github.com/echemdata/galvani.git
synced 2025-12-14 09:15:34 +00:00
Optionally read Arbin into in-memory sqlite without temporary file
This commit is contained in:
@@ -571,13 +571,25 @@ def mdb_get_version(filename):
|
|||||||
return version_tuple
|
return version_tuple
|
||||||
|
|
||||||
|
|
||||||
def convert_arbin_to_sqlite(input_file, output_file):
|
def convert_arbin_to_sqlite(input_file, output_file=None):
|
||||||
"""Read data from an Arbin .res data file and write to a sqlite file.
|
"""Read data from an Arbin .res data file and write to a sqlite file.
|
||||||
|
|
||||||
Any data currently in the sqlite file will be erased!
|
Any data currently in an sqlite file at `output_file` will be erased!
|
||||||
|
|
||||||
|
Parameters:
|
||||||
|
input_file (str): The path to the Arbin .res file to read from.
|
||||||
|
output_file (str or None): The path to the sqlite file to write to; if None,
|
||||||
|
return a `sqlite3.Connection` into an in-memory database.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
None or sqlite3.Connection
|
||||||
|
|
||||||
"""
|
"""
|
||||||
arbin_version = mdb_get_version(input_file)
|
arbin_version = mdb_get_version(input_file)
|
||||||
|
|
||||||
|
if output_file is None:
|
||||||
|
output_file = ":memory:"
|
||||||
|
|
||||||
s3db = sqlite3.connect(output_file)
|
s3db = sqlite3.connect(output_file)
|
||||||
|
|
||||||
tables_to_convert = copy(mdb_tables)
|
tables_to_convert = copy(mdb_tables)
|
||||||
@@ -602,6 +614,11 @@ def convert_arbin_to_sqlite(input_file, output_file):
|
|||||||
print("Vacuuming database...")
|
print("Vacuuming database...")
|
||||||
s3db.executescript("VACUUM; ANALYZE;")
|
s3db.executescript("VACUUM; ANALYZE;")
|
||||||
|
|
||||||
|
if output_file == ":memory:":
|
||||||
|
return s3db
|
||||||
|
|
||||||
|
s3db.close()
|
||||||
|
|
||||||
|
|
||||||
def main(argv=None):
|
def main(argv=None):
|
||||||
parser = argparse.ArgumentParser(
|
parser = argparse.ArgumentParser(
|
||||||
|
|||||||
@@ -53,6 +53,21 @@ def test_convert_Arbin_to_sqlite_function(testdata_dir, tmpdir, basename):
|
|||||||
csr.fetchone()
|
csr.fetchone()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("basename", ["arbin1", "UM34_Test005E"])
|
||||||
|
def test_convert_Arbin_to_sqlite_function_in_memory(testdata_dir, tmpdir, basename):
|
||||||
|
"""Convert an Arbin file to an in-memory SQLite database."""
|
||||||
|
res_file = os.path.join(testdata_dir, basename + ".res")
|
||||||
|
conn = None
|
||||||
|
try:
|
||||||
|
conn = res2sqlite.convert_arbin_to_sqlite(res_file)
|
||||||
|
assert conn is not None
|
||||||
|
csr = conn.execute("SELECT * FROM Channel_Normal_Table;")
|
||||||
|
csr.fetchone()
|
||||||
|
finally:
|
||||||
|
if conn is not None:
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.skipif(
|
@pytest.mark.skipif(
|
||||||
not have_mdbtools, reason="Reading the Arbin file requires MDBTools"
|
not have_mdbtools, reason="Reading the Arbin file requires MDBTools"
|
||||||
)
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user