Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -19,3 +19,4 @@ data/
/.pytest_cache/
*/super-linter.log
.cemsdev
.cems-box
82 changes: 82 additions & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
# CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

## What is pygef

pygef is a Python library that parses soil measurement data from two file formats:
- **GEF files** (`.gef`) — a Dutch legacy human-readable data format
- **BRO XML files** (`.xml`) — the modern Dutch BRO (Basisregistratie Ondergrond) format

It exposes two public functions: `read_cpt()` for cone penetration test data and `read_bore()` for borehole data. Both return frozen dataclasses (`CPTData`, `BoreData`) with measurement data stored as Polars DataFrames.

## Commands

Install for development:
```bash
pip install -r requirements.txt
pip install -e .[test,plot,map]
```

Run tests:
```bash
coverage run -m pytest
pytest tests/gef/test_gef.py # single test file
pytest -k "test_name" # single test by name
```

Format code:
```bash
black --config "pyproject.toml" .
isort --settings-path "pyproject.toml" .
```

Type check:
```bash
mypy src/pygef
```

Build docs:
```bash
sphinx-build -b html docs public
```

## Architecture

```
src/pygef/
shim.py # Entry point: read_cpt() and read_bore() dispatch to gef or xml parsers
cpt.py # CPTData frozen dataclass + post-processing (frictionRatio, depthOffset)
bore.py # BoreData frozen dataclass
common.py # Shared: Location, VerticalDatumClass enum, coordinate helpers
plotting.py # Optional: plot_cpt() and plot_bore() (requires matplotlib)
gef/ # GEF file parser
gef.py # Low-level GEF header/column parsing
parse_cpt.py # _GefCpt: maps GEF columns to internal representation
parse_bore.py # _GefBore: maps GEF columns to internal representation
mapping.py # Column name mappings from GEF MEASUREMENTVAR codes
broxml/ # BRO XML parser
xml_parser.py # lxml-based XML parsing
parse_cpt.py # read_cpt() returns list of CPTData
parse_bore.py # read_bore() returns list of BoreData
mapping.py # Maps BRO XML parameter names to DataFrame column names
resolvers.py # Value resolvers (unit conversions, void handling)
```

**Data flow:** `shim.read_cpt(file)` → `_classify_input()` returns `(kind, format, payload)` → dispatches to `_GefCpt` (GEF) or `read_cpt_xml` (XML). GEF parser `ValueError`s are wrapped into `ParseGefError`.

**Input classification** (`_classify_input` in `shim.py`): `io.BytesIO` and `pathlib.Path` are routed by sniffing the first ~128 bytes (`#KEY=` → GEF, `<` → XML). A `str` is GEF content if it starts with `#KEY=`, XML content if it starts with `<`, an existing filesystem path if `os.path.exists()` returns True, and otherwise unparseable (`ValueError`). A missing `pathlib.Path` raises `FileNotFoundError`; a missing `str` does not — it falls through to the same `ValueError`. When `engine` is `"gef"` or `"xml"`, it must match the detected format or `ValueError` is raised.

**Key design notes:**
- `CPTData.__post_init__` runs post-processing on the DataFrame (computes `frictionRatioComputed`, `depthOffset`, sorts by `penetrationLength`). Because the dataclass is frozen, it uses `object.__setattr__` to replace `data`.
- XML files can contain multiple measurements; `index` parameter selects which one.
- `replace_column_voids` and `remove_pre_excavated_rows` are GEF-only options passed through `read_cpt()`.
- Coordinates use Dutch RD New (EPSG:28992) for GEF files; BRO XML may include WGS84 standardized locations.
- The `src/` layout requires `pythonpath = ["src"]` in pytest config (already set in `pyproject.toml`).

## Test structure

- `tests/gef/` — GEF file parsing tests
- `tests/xml/` — BRO XML parsing tests
- `tests/test_files/` — sample `.gef` files; XML fixtures in subdirectories
- `tests/test_shim.py` — integration tests for the public API
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
name = "pygef"
version = "0.14.0"
description = "Parse soil measurument data."
dependencies = ["polars>=1,<2", "lxml>=5,<7", "gef-file-to-map>=1,<2"]
dependencies = ["polars>=1,<2", "lxml>=5,<7", "gef-file-to-map>=1,<2", "numpy>2,<3"]
requires-python = ">=3.11,<3.15"
license = { file = "LICENSE.txt" }
readme = "README.md"
Expand Down
6 changes: 5 additions & 1 deletion src/pygef/__init__.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,12 @@
from pygef._version import __version__
from pygef.exceptions import ParseCptGefError, ParseGefError, UserError
from pygef.shim import read_bore, read_cpt

__all__ = [
"__version__",
"read_cpt",
"ParseCptGefError",
"ParseGefError",
"UserError",
"read_bore",
"read_cpt",
]
188 changes: 146 additions & 42 deletions src/pygef/shim.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

import io
import os
import re
from enum import Enum
from pathlib import Path
from typing import Any, Literal

Expand All @@ -10,28 +12,115 @@
from pygef.broxml.parse_cpt import read_cpt as read_cpt_xml
from pygef.common import Location, VerticalDatumClass, convert_coordinate_system_to_gml
from pygef.cpt import CPTData
from pygef.exceptions import ParseGefError
from pygef.gef.parse_bore import _GefBore
from pygef.gef.parse_cpt import _GefCpt

GEF_ID = "#GEFID"
# A GEF header line looks like "#KEY= value" or "#KEY = value". Accepting any
# such line (not only #GEFID) makes string-vs-path disambiguation robust to
# GEF files that don't put #GEFID first.
_GEF_HEADER_RE = re.compile(r"^\s*#[A-Z][A-Z0-9]*\s*=")

_PEEK_BYTES = 128

def is_gef_file(file: io.BytesIO | Path | str) -> bool:

class _InputKind(Enum):
BYTES = "bytes"
PATH = "path"
GEF_TEXT = "gef_text"
XML_TEXT = "xml_text"


class _DetectedFormat(str, Enum):
GEF = "gef"
XML = "xml"


_LSTRIP_CHARS = " \t\r\n" # whitespace + optional UTF-8 BOM

# A string looks like a filesystem path if it contains a path separator OR ends
# with a short alphanumeric extension. Bare random words ("foo bar baz") fail
# this test and so are reported as ambiguous input rather than missing files.
_PATH_LIKE_RE = re.compile(r"[/\\]|\.[A-Za-z0-9]{1,8}$")


def _looks_like_gef(head: str) -> bool:
return bool(_GEF_HEADER_RE.match(head.lstrip(_LSTRIP_CHARS)))


def _looks_like_xml(head: str) -> bool:
return head.lstrip(_LSTRIP_CHARS).startswith("<")


def _detect_format(head: str, source: str) -> _DetectedFormat:
if _looks_like_gef(head):
return _DetectedFormat.GEF
if _looks_like_xml(head):
return _DetectedFormat.XML
raise ValueError(
f"Could not detect file format of {source}: content matches neither "
"the GEF header pattern (#KEY=...) nor the XML root element pattern "
"(starting with '<')."
)


def _classify_input(
file: io.BytesIO | Path | str,
) -> tuple[_InputKind, _DetectedFormat, Any]:
"""
gef files start with '#GEFID' so we check the content
of the file
Classify ``file`` into one of the four input kinds and detect whether the
content looks like GEF or XML.

Returns ``(kind, detected_format, payload)`` where ``payload`` is the value
to hand off to the chosen parser (BytesIO/Path/string).

Raises ``FileNotFoundError`` if ``file`` resolves to a path that doesn't
exist on disk; raises ``ValueError`` if the content cannot be classified
as GEF or XML.
"""
if isinstance(file, io.BytesIO):
pos = file.tell()
is_gef = file.read(6).decode().startswith(GEF_ID)
head = file.read(_PEEK_BYTES).decode(errors="ignore")
file.seek(pos)
return is_gef
if os.path.exists(file):
with open(file, errors="ignore") as f:
return f.read(6).startswith(GEF_ID)
return _InputKind.BYTES, _detect_format(head, "BytesIO input"), file

if isinstance(file, Path):
if not file.exists():
raise FileNotFoundError(f"File not found: {file}")
with open(file, encoding="utf-8", errors="ignore") as f:
head = f.read(_PEEK_BYTES)
return _InputKind.PATH, _detect_format(head, str(file)), file

if isinstance(file, str):
return file[:6].startswith(GEF_ID)
raise FileNotFoundError("Could not find the GEF file.")
if _looks_like_gef(file):
return _InputKind.GEF_TEXT, _DetectedFormat.GEF, file
if _looks_like_xml(file):
return _InputKind.XML_TEXT, _DetectedFormat.XML, file
if os.path.exists(file):
with open(file, encoding="utf-8", errors="ignore") as f:
head = f.read(_PEEK_BYTES)
return _InputKind.PATH, _detect_format(head, file), file

raise ValueError(
"Could not interpret string input: it does not look like GEF or "
"XML and is not an existing filesystem path."
)

raise TypeError(
f"Unsupported file type {type(file).__name__}; "
"expected io.BytesIO, pathlib.Path, or str."
)


def _check_engine_match(engine: str, detected_format: _DetectedFormat) -> None:
"""Raise if the user-forced engine doesn't match the detected format."""
if engine == "auto":
return
if engine != detected_format:
raise ValueError(
f"engine={engine!r} but file content looks like {detected_format!r}; "
"refusing to parse."
)


def read_bore(
Expand All @@ -42,21 +131,32 @@ def read_bore(
"""
Parse the bore file. Can either be BytesIO, Path or str

:param file: bore file
:param file: bore file. A ``str`` argument is interpreted as raw file
content if it starts with a GEF header line (``#KEY=...``) or with
``<`` (XML); otherwise it is treated as a filesystem path.
:param index: only valid for xml files
:param engine: default is "auto". parsing engine.
Please note that auto engine checks if the files starts with `#GEFID`.
:param engine: default is "auto". Parsing engine.
When set to "gef" or "xml" the engine must match the detected file
content, otherwise a ValueError is raised.
"""
if engine == "gef" or is_gef_file(file) and engine == "auto":
kind, detected_format, payload = _classify_input(file)
_check_engine_match(engine, detected_format)

if detected_format == "gef":
if index > 0:
raise ValueError("an index > 0 not supported for GEF files")
if isinstance(file, io.BytesIO):
return gef_bore_to_bore_data(_GefBore(string=file.read().decode()))
if os.path.exists(file):
return gef_bore_to_bore_data(_GefBore(path=file))
else:
return gef_bore_to_bore_data(_GefBore(string=file))
return read_bore_xml(file)[index]
try:
if kind is _InputKind.BYTES:
gef_bore = _GefBore(string=payload.read().decode())
elif kind is _InputKind.PATH:
gef_bore = _GefBore(path=payload)
else:
gef_bore = _GefBore(string=payload)
except ValueError as e:
raise ParseGefError(str(e)) from e
return gef_bore_to_bore_data(gef_bore)

return read_bore_xml(payload)[index]


def read_cpt(
Expand All @@ -69,44 +169,48 @@ def read_cpt(
"""
Parse the cpt file. Can either be BytesIO, Path or str

:param file: bore file
:param file: cpt file. A ``str`` argument is interpreted as raw file
content if it starts with a GEF header line (``#KEY=...``) or with
``<`` (XML); otherwise it is treated as a filesystem path.
:param index: only valid for xml files
:param engine: default is "auto". parsing engine.
Please note that auto engine checks if the files starts with `#GEFID`.
:param engine: default is "auto". Parsing engine.
When set to "gef" or "xml" the engine must match the detected file
content, otherwise a ValueError is raised.
:param replace_column_voids: default True. How to handle rows with void values.
If true, replace void values with nulls or interpolate; else retain value.
:param remove_pre_excavated_rows: default True. How to handle pre-excavated row values.
If true, drop rows above pre-excavated depth; else retain.
"""
kind, detected_format, payload = _classify_input(file)
_check_engine_match(engine, detected_format)

if engine == "gef" or is_gef_file(file) and engine == "auto":
if detected_format == "gef":
if index > 0:
raise ValueError("an index > 0 not supported for GEF files")
if isinstance(file, io.BytesIO):
return gef_cpt_to_cpt_data(
_GefCpt(
string=file.read().decode(),
try:
if kind is _InputKind.BYTES:
gef = _GefCpt(
string=payload.read().decode(),
replace_column_voids=replace_column_voids,
remove_pre_excavated_rows=remove_pre_excavated_rows,
)
)
if os.path.exists(file):
return gef_cpt_to_cpt_data(
_GefCpt(
path=file,
elif kind is _InputKind.PATH:
gef = _GefCpt(
path=payload,
replace_column_voids=replace_column_voids,
remove_pre_excavated_rows=remove_pre_excavated_rows,
)
)
else:
return gef_cpt_to_cpt_data(
_GefCpt(
string=file,
else:
gef = _GefCpt(
string=payload,
replace_column_voids=replace_column_voids,
remove_pre_excavated_rows=remove_pre_excavated_rows,
)
)
return read_cpt_xml(file)[index]
except ValueError as e:
raise ParseGefError(str(e)) from e
return gef_cpt_to_cpt_data(gef)

return read_cpt_xml(payload)[index]


def convert_height_system_to_vertical_datum(height_system: float) -> str:
Expand Down
5 changes: 5 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,3 +53,8 @@ def cpt_gef_3() -> str:
@fixture()
def cpt_gef_4() -> str:
return os.path.join(TEST_FILES, "cpt4.gef")


@fixture()
def erroneous_gef() -> str:
return os.path.join(TEST_FILES, "erroneous.gef")
6 changes: 6 additions & 0 deletions tests/test_files/erroneous.gef
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
#GEFID= 1, 1, 0
#COMPANYID= ErroneousCorp
#FILEDATE= 2026, 06, 17
#COMMENT= This file is intentionally corrupt: it begins with a GEF header
#COMMENT= but is missing REPORTCODE/PROCEDURECODE and column metadata.
not a valid GEF data section
Loading
Loading