Close LGVISIUM-63: Extraction of the groundwater logo using computer vision

config/matching_params.yml

@@ -115,6 +115,13 @@ coordinate_keys:
  - coordonnées
  - coordonn
 
+coordinate_fp_keys:
+
+
+groundwater_fp_keys:
+ - Wasserstau
+ - Grundwasser-
+ - Grundwasserfassung
 
 groundwater_keys:
  # German
@@ -132,7 +139,6 @@ groundwater_keys:
  - W SP
  - Gr.W.spiegel
  - GrW Sp
- - Wsp.
  - Wsp
  - GW-Spiegel
  - Grundwasser
@@ -170,3 +176,5 @@ elevation_keys:
  - Ansatzhöhe
  - Terrainkote
 
+elevation_fp_keys:
+

pyproject.toml

@@ -35,6 +35,7 @@ dependencies = [
  "opencv-python-headless",
  "quads>=1.1.0",
  "numpy<2",
+ "scikit-image==0.24.0"
 ]
 
 [project.optional-dependencies]

src/stratigraphy/data_extractor/data_extractor.py

@@ -40,6 +40,7 @@ class DataExtractor(ABC):
 
  doc: fitz.Document = None
  feature_keys: list[str] = None
+ feature_fp_keys: list[str] = None
  feature_name: str = None
 
  # How much to the left of a key do we look for the feature information, as a multiple of the key line width
@@ -48,6 +49,8 @@ class DataExtractor(ABC):
  search_right_factor: float = 0
  # How much below a key do we look for the feature information, as a multiple of the key line height
  search_below_factor: float = 0
+ # How much above a key do we look for the feature information, as a multiple of the key line height
+ search_above_factor: float = 0
 
  preprocess_replacements: dict[str, str] = {}
 
@@ -63,6 +66,11 @@ def __init__(self, document: fitz.Document):
 
  self.doc = document
  self.feature_keys = read_params("matching_params.yml")[f"{self.feature_name}_keys"]
+ self.feature_fp_keys = (
+ read_params("matching_params.yml")[f"{self.feature_name}_fp_keys"]
+ if read_params("matching_params.yml")[f"{self.feature_name}_fp_keys"]
+ else []
+ )
 
  def preprocess(self, value: str) -> str:
  for old, new in self.preprocess_replacements.items():
@@ -105,7 +113,15 @@ def find_feature_key(self, lines: list[TextLine], allowed_error_rate: float = 0.
  for line in lines:
  match = pattern.search(line.text)
  if match:
- matches.add(line)
+ # Make sure the key is not in the false positive list
+ is_fp_key = False
+ for fp_key in self.feature_fp_keys:
+ if fp_key in line.text:
+ is_fp_key = True
+ break
+
+ if not is_fp_key:
+ matches.add(line)
 
  return list(matches)
 
@@ -122,14 +138,28 @@ def get_lines_near_key(self, lines, key_line: TextLine) -> list[TextLine]:
  list[TextLine]: The lines close to the key.
  """
  key_rect = key_line.rect
- elevation_search_rect = fitz.Rect(
- key_rect.x0 - self.search_left_factor * key_rect.width,
- key_rect.y0,
- key_rect.x1 + self.search_right_factor * key_rect.width,
- key_rect.y1 + self.search_below_factor * key_rect.height,
- )
- feature_lines = [line for line in lines if line.rect.intersects(elevation_search_rect)]
+ feature_lines = self.get_lines_near_rect(lines, key_rect)
 
  # makes sure the line with the key is included first in the extracted information and the duplicate removed
  feature_lines.insert(0, key_line)
  return list(dict.fromkeys(feature_lines))
+
+ def get_lines_near_rect(self, lines, rect: fitz.Rect) -> list[TextLine]:
+ """Find the lines of the text that are close to a given rectangle.
+
+ Args:
+ lines (list[TextLine]): Arbitrary text lines to search in.
+ rect (fitz.Rect): The rectangle to search around.
+
+ Returns:
+ list[TextLine]: The lines close to the rectangle.
+ """
+ search_rect = fitz.Rect(
+ rect.x0 - self.search_left_factor * rect.width,
+ rect.y0 - self.search_above_factor * rect.height,
+ rect.x1 + self.search_right_factor * rect.width,
+ rect.y1 + self.search_below_factor * rect.height,
+ )
+ feature_lines = [line for line in lines if line.rect.intersects(search_rect)]
+
+ return feature_lines

src/stratigraphy/groundwater/groundwater_extraction.py

@@ -2,12 +2,16 @@
 
 import abc
 import logging
+import math
+import os
 from dataclasses import dataclass
 from datetime import date as dt
 from datetime import datetime
+from pathlib import Path
 
 import fitz
 import numpy as np
+import skimage as ski
 from stratigraphy.data_extractor.data_extractor import DataExtractor, ExtractedFeature
 from stratigraphy.groundwater.utility import extract_date, extract_depth, extract_elevation
 from stratigraphy.lines.line import TextLine
@@ -137,9 +141,11 @@ class GroundwaterLevelExtractor(DataExtractor):
  feature_name = "groundwater"
 
  # look for elevation values to the left, right and/or immediately below the key
- search_left_factor: float = 2
+ search_left_factor: float = 3 # NOTE: check files 267125334-bp.pdf, 267125338-bp.pdf, and 267125339-bp.pdf if this
+ # value is too high, as it might lead to false positives
  search_right_factor: float = 10
  search_below_factor: float = 4
+ search_above_factor: float = 4
 
  preprocess_replacements = {",": ".", "'": ".", "o": "0", "\n": " ", "ü": "u"}
 
@@ -210,7 +216,6 @@ def get_groundwater_info_from_lines(self, lines: list[TextLine], page: int) -> G
 
  elevation = extract_elevation(text)
 
- # Pattern for matching depth (e.g., "1,48 m u.T.")
  matched_lines_rect.append(line.rect)
  else:
  # Pattern for matching date
@@ -219,6 +224,12 @@ def get_groundwater_info_from_lines(self, lines: list[TextLine], page: int) -> G
  if extracted_date_str:
  text = text.replace(extracted_date_str, "").strip()
  date = extracted_date
+ matched_lines_rect.append(line.rect)
+ else:
+ # in case several dates are present, we skip the other dates
+ extracted_date, extracted_date_str = extract_date(text)
+ if extracted_date_str:
+ continue
 
  # Pattern for matching depth (e.g., "1,48 m u.T.")
  if not depth:
@@ -269,6 +280,184 @@ def get_groundwater_info_from_lines(self, lines: list[TextLine], page: int) -> G
  else:
  raise ValueError("Could not extract all required information from the lines provided.")
 
+ def load_templates(self) -> list[np.ndarray]:
+ """Load the templates for the groundwater information.
+
+ Returns:
+ list[np.ndarray]: the loaded templates
+ """
+ templates = []
+ template_dir = os.path.join(os.path.dirname(__file__), "assets")
+ for template in os.listdir(template_dir):
+ if template.endswith(".npy"): # and template.startswith("700246002-bp_page1_template"):
+ templates.append(np.load(os.path.join(template_dir, template)))
+ return templates
+
+ def get_groundwater_from_illustration(
+ self, lines: list[TextLine], page_number: int, terrain_elevation: Elevation | None
+ ) -> list[GroundwaterInformationOnPage]:
+ """Extracts the groundwater information from an illustration.
+
+ Args:
+ lines (list[TextLine]): the lines of text to extract the groundwater information from
+ page_number (int): the page number (1-based) of the PDF document
+ terrain_elevation (Elevation | None): The elevation of the terrain.
+
+ Returns:
+ list[GroundwaterInformationOnPage]: the extracted groundwater information
+ """
+ extracted_groundwater_list = []
+ confidence_list = []
+
+ # convert the doc to an image
+ page = self.doc.load_page(page_number - 1)
+ filename = Path(self.doc.name).stem
+ png_filename = f"{filename}-{page_number + 1}.png"
+ png_path = f"/tmp/{png_filename}" # Local path to save the PNG
+ fitz.utils.get_pixmap(page, matrix=fitz.Matrix(2, 2), clip=page.rect).save(png_path)
+
+ # load the image
+ img = ski.io.imread(png_path)
+ N_BEST_MATCHES = 5
+ TEMPLATE_MATCH_THRESHOLD = 0.66
+
+ # extract the groundwater information from the image
+ for template in self.load_templates():
+ # Compute the match of the template and the image (correlation coef)
+ result = ski.feature.match_template(img, template)
+
+ for _ in range(N_BEST_MATCHES):
+ ij = np.unravel_index(np.argmax(result), result.shape)
+ confidence = np.max(result) # TODO - use confidence to filter out bad matches
+ if confidence < TEMPLATE_MATCH_THRESHOLD:
+ # skip this template if the confidence is too low to avoid false positives
+ continue
+ top_left = (ij[1], ij[0])
+ illustration_rect = fitz.Rect(
+ top_left[0], top_left[1], top_left[0] + template.shape[1], top_left[1] + template.shape[0]
+ )
+
+ # remove the matched area from the result to avoid finding the same area again
+ x_area_to_remove = int(0.75 * template.shape[1])
+ y_area_to_remove = int(0.75 * template.shape[0])
+ result[
+ int(illustration_rect.y0) - y_area_to_remove : int(illustration_rect.y1) + y_area_to_remove,
+ int(illustration_rect.x0) - x_area_to_remove : int(illustration_rect.x1) + x_area_to_remove,
+ ] = float("-inf")
+
+ # convert the illustration_rect to the coordinate system of the PDF
+ horizontal_scaling = page.rect.width / img.shape[1]
+ vertical_scaling = page.rect.height / img.shape[0]
+ pdf_illustration_rect = fitz.Rect(
+ illustration_rect.x0 * horizontal_scaling,
+ illustration_rect.y0 * vertical_scaling,
+ illustration_rect.x1 * horizontal_scaling,
+ illustration_rect.y1 * vertical_scaling,
+ )
+
+ # extract the groundwater information from the image using the text
+ groundwater_info_lines = self.get_lines_near_rect(lines, pdf_illustration_rect)
+
+ # sort the lines by their proximity to the key line center, compute the distance to the key line center
+ def distance_to_key_center(line_rect: fitz.Rect, illustration_rect: fitz.Rect) -> float:
+ key_center_x = (illustration_rect.x0 + illustration_rect.x1) / 2
+ key_center_y = (illustration_rect.y0 + illustration_rect.y1) / 2
+ line_center_x = (line_rect.x0 + line_rect.x1) / 2
+ line_center_y = (line_rect.y0 + line_rect.y1) / 2
+ return math.sqrt((line_center_x - key_center_x) ** 2 + (line_center_y - key_center_y) ** 2)
+
+ groundwater_info_lines.sort(key=lambda line: distance_to_key_center(line.rect, pdf_illustration_rect))
+ try:
+ extracted_gw = self.get_groundwater_info_from_lines(groundwater_info_lines, page_number)
+ if extracted_gw.groundwater.depth or extracted_gw.groundwater.elevation:
+ # Fill in the depth and elevation if they are not already filled in based on the terrain
+ if terrain_elevation:
+ if not extracted_gw.groundwater.depth and extracted_gw.groundwater.elevation:
+ extracted_gw.groundwater.depth = round(
+ terrain_elevation.elevation - extracted_gw.groundwater.elevation, 2
+ )
+ if not extracted_gw.groundwater.elevation and extracted_gw.groundwater.depth:
+ extracted_gw.groundwater.elevation = round(
+ terrain_elevation.elevation - extracted_gw.groundwater.depth, 2
+ )
+
+ # Make a sanity check to see if elevation and depth make sense (i.e., they add up:
+ # elevation + depth = terrain elevation)
+ if extracted_gw.groundwater.elevation and extracted_gw.groundwater.depth:
+ extract_terrain_elevation = round(
+ extracted_gw.groundwater.elevation + extracted_gw.groundwater.depth, 2
+ )
+ if extract_terrain_elevation != terrain_elevation.elevation:
+ # If the extracted elevation and depth do not match the terrain elevation, we try
+ # to remove one of the items from the match and see if we can find a better match.
+ logger.warning(
+ "The extracted elevation and depth do not match the terrain elevation."
+ )
+ logger.warning(
+ "Elevation: %s, Depth: %s, Terrain Elevation: %s",
+ extracted_gw.groundwater.elevation,
+ extracted_gw.groundwater.depth,
+ terrain_elevation.elevation,
+ )
+
+ # re-run the extraction and see if we can find a better match by removing one
+ # item from the current match
+ groundwater_info_lines_without_depth = [
+ line
+ for line in groundwater_info_lines
+ if str(extracted_gw.groundwater.depth) not in line.text
+ ]
+ groundwater_info_lines_without_elevation = [
+ line
+ for line in groundwater_info_lines
+ if str(extracted_gw.groundwater.elevation) not in line.text
+ ]
+ extracted_gw = self.get_groundwater_info_from_lines(
+ groundwater_info_lines_without_depth, page_number
+ )
+
+ if not extracted_gw.groundwater.depth:
+ extracted_gw = self.get_groundwater_info_from_lines(
+ groundwater_info_lines_without_elevation, page_number
+ )
+
+ if extracted_gw.groundwater.elevation and extracted_gw.groundwater.depth:
+ extract_terrain_elevation = round(
+ extracted_gw.groundwater.elevation + extracted_gw.groundwater.depth, 2
+ )
+
+ if extract_terrain_elevation != terrain_elevation.elevation:
+ logger.warning(
+ "The extracted elevation and depth do not match the terrain elevation."
+ )
+ logger.warning(
+ "Elevation: %s, Depth: %s, Terrain Elevation: %s",
+ extracted_gw.groundwater.elevation,
+ extracted_gw.groundwater.depth,
+ terrain_elevation.elevation,
+ )
+ continue
+
+ # Only if the groundwater information is not already in the list
+ if extracted_gw not in extracted_groundwater_list and extracted_gw.groundwater.date:
+ extracted_groundwater_list.append(extracted_gw)
+ confidence_list.append(confidence)
+
+ # Remove the extracted groundwater information from the lines to avoid double extraction
+ for line in groundwater_info_lines:
+ # if the rectangle of the line is in contact with the rectangle of the extracted
+ # groundwater information, remove the line
+ if line.rect.intersects(extracted_gw.rect):
+ lines.remove(line)
+
+ except ValueError as error:
+ logger.warning("ValueError: %s", error)
+ continue
+
+ # TODO: Maybe we could stop the search if we found a good match with one of the templates
+
+ return extracted_groundwater_list, confidence_list
+
  def extract_groundwater(self, terrain_elevation: Elevation | None) -> list[GroundwaterInformationOnPage]:
  """Extracts the groundwater information from a borehole profile.
 
@@ -287,10 +476,18 @@ def extract_groundwater(self, terrain_elevation: Elevation | None) -> list[Groun
  lines = extract_text_lines(page)
  page_number = page.number + 1 # page.number is 0-based
 
- found_groundwater = (
- self.get_groundwater_near_key(lines, page_number)
- # or XXXX # Add other techniques here
- )
+ found_groundwater = self.get_groundwater_near_key(lines, page_number)
+ if not found_groundwater:
+ logger.info("No groundwater found near the key on page %s.", page_number)
+ found_groundwater, confidence_list = self.get_groundwater_from_illustration(
+ lines, page_number, terrain_elevation
+ )
+ logger.info("Confidence list: %s", confidence_list)
+ print("Confidence list: %s", confidence_list)
+ logger.info("Found groundwater from illustration on page %s: %s", page_number, found_groundwater)
+ print("Found groundwater from illustration on page %s: %s", page_number, found_groundwater)
+ if not found_groundwater:
+ logger.info("No groundwater found in the illustration on page %s.", page_number)
 
  if terrain_elevation:
  # If the elevation is provided, calculate the depth of the groundwater

src/stratigraphy/groundwater/utility.py

@@ -45,12 +45,15 @@ def extract_depth(text: str, max_depth: int) -> float | None:
  for pattern in depth_patterns:
  depth_match = regex.search(pattern, corrected_text)
  if depth_match:
- depth = float(depth_match.group(1).replace(",", "."))
- if depth > max_depth:
- # If the extracted depth is greater than the max depth, set it to None and continue searching.
- depth = None
- else:
- break
+ try:
+ depth = float(depth_match.group(1).replace(",", "."))
+ if depth > max_depth:
+ # If the extracted depth is greater than the max depth, set it to None and continue searching.
+ depth = None
+ else:
+ break
+ except ValueError:
+ continue
  return depth