Add optional Parquet read/write extension (#24)

README.md

@@ -11,8 +11,8 @@ A tool to fetch results from EnergyPlus output files (`.sql` and `.eso` formats)
 - Filter variables by key, type, and units
 - Support for exact and substring (alike) matching
 - Filter results by date range
-- Export results to CSV
-- Zero runtime dependencies
+- Export results to CSV (and optionally Parquet via the `parquet` extra)
+- Zero runtime dependencies (the `parquet` extra adds `pyarrow`)
 
 ## DesignBuilder Compatibility
 
@@ -176,6 +176,27 @@ results.to_csv(
 )
 ```
 
+### Parquet (optional)
+
+Parquet is handy for object-storage workflows. It's an optional extension —
+install the extra to enable it:
+
+```bash
+pip install db-eplusout-reader[parquet]
+```
+
+```python
+from db_eplusout_reader import get_results, to_parquet, read_parquet
+
+results = get_results(path, variables, frequency=M)
+
+# Write to Parquet (extra kwargs forwarded to pyarrow, e.g. compression)
+to_parquet(results, r"C:\output.parquet", compression="snappy")
+
+# Read back into a ResultsDictionary (frequency, variables and time series preserved)
+results = read_parquet(r"C:\output.parquet")
+```
+
 ## Complete Example
 
 ```python

db_eplusout_reader/__init__.py

@@ -2,6 +2,7 @@
 
 from db_eplusout_reader.db_esofile import DBEsoFile, DBEsoFileCollection
 from db_eplusout_reader.get_results import get_results
+from db_eplusout_reader.parquet import read_parquet, to_parquet
 from db_eplusout_reader.processing.esofile_reader import Variable
 from db_eplusout_reader.sql_reader import (
  get_all_variables,
@@ -14,6 +15,8 @@
  "DBEsoFileCollection",
  "get_results",
  "Variable",
+ "read_parquet",
+ "to_parquet",
  "get_tables",
  "get_variables",
  "get_all_variables",

db_eplusout_reader/parquet.py

@@ -0,0 +1,138 @@
+"""Optional Parquet I/O for results dictionaries.
+
+Parquet is handy for object-storage workflows. Support is an optional
+extension: install the extra to enable it::
+
+ pip install db-eplusout-reader[parquet]
+
+A results dictionary is stored as a columnar table with one column per
+variable plus an optional ``timestamp`` column. The variable fields
+(key, type, units) and the reporting frequency are preserved in Arrow
+metadata so the table round-trips back into a ``ResultsDictionary``.
+"""
+
+from db_eplusout_reader.processing.esofile_reader import Variable
+from db_eplusout_reader.results_dict import ResultsDictionary
+
+try:
+ import pyarrow as pa
+ import pyarrow.parquet as pq
+except ImportError: # pragma: no cover - exercised only without the extra
+ pa = None
+ pq = None
+
+_TIMESTAMP_COLUMN = "timestamp"
+_ROLE_KEY = b"db_role"
+_ROLE_TIMESTAMP = b"timestamp"
+_ROLE_VARIABLE = b"variable"
+_FREQUENCY_KEY = b"frequency"
+_KEY_KEY = b"key"
+_TYPE_KEY = b"type"
+_UNITS_KEY = b"units"
+
+
+def _require_pyarrow():
+ if pa is None:
+ raise ImportError(
+ "Parquet support requires the optional 'parquet' extra. "
+ "Install it with: pip install db-eplusout-reader[parquet]"
+ )
+
+
+def _encode(value):
+ """Encode a (possibly None) variable field as bytes for Arrow metadata."""
+ return (value if value is not None else "").encode("utf-8")
+
+
+def to_parquet(results_dictionary, path, **kwargs):
+ """
+ Save a results dictionary as a Parquet file.
+
+ Parameters
+ ----------
+ results_dictionary : ResultsDictionary
+ Results to store. Must contain at least one variable.
+ path : os.PathLike
+ Destination Parquet file path.
+ **kwargs
+ Additional keyword arguments forwarded to ``pyarrow.parquet.write_table``
+ (e.g. ``compression``).
+
+ Returns
+ -------
+ None
+
+ """
+ _require_pyarrow()
+ fields = []
+ columns = []
+
+ if results_dictionary.time_series:
+ fields.append(
+ pa.field(
+ _TIMESTAMP_COLUMN,
+ pa.timestamp("us"),
+ metadata={_ROLE_KEY: _ROLE_TIMESTAMP},
+ )
+ )
+ columns.append(pa.array(results_dictionary.time_series, type=pa.timestamp("us")))
+
+ for variable, array in zip(results_dictionary.variables, results_dictionary.arrays):
+ metadata = {
+ _ROLE_KEY: _ROLE_VARIABLE,
+ _KEY_KEY: _encode(variable.key),
+ _TYPE_KEY: _encode(variable.type),
+ _UNITS_KEY: _encode(variable.units),
+ }
+ name = "{}|{}|{}".format(variable.key, variable.type, variable.units)
+ fields.append(pa.field(name, pa.float64(), metadata=metadata))
+ columns.append(pa.array(array, type=pa.float64()))
+
+ schema = pa.schema(fields, metadata={_FREQUENCY_KEY: _encode(results_dictionary.frequency)})
+ pq.write_table(pa.table(columns, schema=schema), path, **kwargs)
+
+
+def read_parquet(path):
+ """
+ Read a results dictionary from a Parquet file written by ``to_parquet``.
+
+ Parameters
+ ----------
+ path : os.PathLike
+ Parquet file path.
+
+ Returns
+ -------
+ ResultsDictionary
+ Reconstructed results, including frequency and time series.
+
+ """
+ _require_pyarrow()
+ table = pq.read_table(path)
+ schema = table.schema
+
+ frequency = ""
+ if schema.metadata and _FREQUENCY_KEY in schema.metadata:
+ frequency = schema.metadata[_FREQUENCY_KEY].decode("utf-8")
+
+ results_dictionary = ResultsDictionary(frequency)
+ time_series = None
+ for i, field in enumerate(schema):
+ metadata = field.metadata or {}
+ role = metadata.get(_ROLE_KEY)
+ column = table.column(i).to_pylist()
+ is_timestamp = role == _ROLE_TIMESTAMP or (
+ role is None and field.name == _TIMESTAMP_COLUMN
+ )
+ if is_timestamp:
+ time_series = column
+ else:
+ variable = Variable(
+ metadata.get(_KEY_KEY, b"").decode("utf-8"),
+ metadata.get(_TYPE_KEY, b"").decode("utf-8"),
+ metadata.get(_UNITS_KEY, b"").decode("utf-8"),
+ )
+ results_dictionary[variable] = column
+
+ results_dictionary.time_series = time_series
+ return results_dictionary

pyproject.toml

@@ -11,6 +11,9 @@ license = { text = "MIT" }
 requires-python = ">=3.9"
 dependencies = []
 
+[project.optional-dependencies]
+parquet = ["pyarrow>=14"]
+
 [dependency-groups]
 dev = [
  "eppy>=0.5.69",
@@ -19,6 +22,7 @@ dev = [
  "pre-commit>=3.6.1",
  "pytest>=8.0.0",
  "pytest-cov>=4.1.0",
+ "pyarrow>=14",
 ]
 
 [build-system]

tests/test_parquet.py

@@ -0,0 +1,77 @@
+"""Tests for the optional Parquet I/O extension.
+
+Skipped entirely when pyarrow (the 'parquet' extra) is not installed.
+Covers round-tripping a ResultsDictionary to Parquet and back via the
+to_parquet / read_parquet functions, including frequency, variables,
+arrays and time series.
+"""
+
+from datetime import datetime
+
+import pytest
+
+pytest.importorskip("pyarrow")
+
+from db_eplusout_reader import Variable, get_results, read_parquet, to_parquet
+from db_eplusout_reader.constants import H
+from db_eplusout_reader.results_dict import ResultsDictionary
+
+_DRYBULB = Variable("Environment", "Site Outdoor Air Drybulb Temperature", "C")
+
+
+@pytest.fixture
+def float_results():
+ rd = ResultsDictionary(frequency=H)
+ rd.time_series = [datetime(2002, 1, 1, 1), datetime(2002, 1, 1, 2), datetime(2002, 1, 1, 3)]
+ rd[Variable("ZONE1", "Zone Mean Air Temperature", "C")] = [20.0, 21.0, 20.5]
+ rd[Variable("ZONE2", "Zone Mean Air Temperature", "C")] = [22.0, 23.0, 19.0]
+ return rd
+
+
+class TestParquetRoundTrip:
+ def test_round_trip(self, float_results, tmp_path):
+ path = tmp_path / "results.parquet"
+ to_parquet(float_results, path)
+ loaded = read_parquet(path)
+
+ assert loaded.frequency == float_results.frequency
+ assert loaded.variables == float_results.variables
+ assert loaded.arrays == float_results.arrays
+ assert loaded.time_series == float_results.time_series
+
+ def test_round_trip_from_sql(self, sql_path, tmp_path):
+ rd = get_results(sql_path, _DRYBULB, frequency=H)
+ path = tmp_path / "sql.parquet"
+ to_parquet(rd, path)
+ loaded = read_parquet(path)
+
+ assert loaded.frequency == rd.frequency
+ assert loaded.first_variable == _DRYBULB
+ assert loaded.first_array == rd.first_array
+ assert loaded.time_series == rd.time_series
+
+ def test_round_trip_without_time_series(self, tmp_path):
+ rd = ResultsDictionary(frequency=H)
+ rd[_DRYBULB] = [1.0, 2.0, 3.0]
+ path = tmp_path / "no_time.parquet"
+ to_parquet(rd, path)
+ loaded = read_parquet(path)
+
+ assert loaded.time_series is None
+ assert loaded[_DRYBULB] == [1.0, 2.0, 3.0]
+
+ def test_duplicate_variables_preserved(self, tmp_path):
+ # variables with identical (key, type, units) must both survive
+ rd = ResultsDictionary(frequency=H)
+ same = Variable("ZONE1", "Zone Mean Air Temperature", "C")
+ rd[same] = [1.0, 2.0]
+ path = tmp_path / "dup.parquet"
+ to_parquet(rd, path)
+ loaded = read_parquet(path)
+ assert loaded[same] == [1.0, 2.0]
+
+ def test_compression_kwarg_forwarded(self, float_results, tmp_path):
+ path = tmp_path / "compressed.parquet"
+ to_parquet(float_results, path, compression="gzip") # must not raise
+ loaded = read_parquet(path)
+ assert loaded.arrays == float_results.arrays