Fix/revert to line by line parsing

CHANGES.md

@@ -1,5 +1,11 @@
 # NGI Python SGF Parser Package
 
+_2025-05-13_
+
+Version 0.0.8
+
+- Reverted the parsing strategy to go line by line, such that K-codes do not overrule the other data.
+
 _2025-04-11_
 
 Version 0.0.7

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "sgf-parser"
-version = "0.0.7"
+version = "0.0.8"
 description = "Parser for the Swedish Geotechnical Society / Svenska Geotekniska Föreningen (SGF) data format"
 authors = [{ name = "Jostein Leira", email = "jostein@leira.net" }]
 requires-python = ">=3.11,<4"

src/sgf_parser/models/method.py

@@ -8,7 +8,6 @@
 
 from sgf_parser.datetime_parser import convert_str_to_datetime, convert_str_to_time
 from sgf_parser.models import MethodType
-from sgf_parser.models.types import FlushingVariant, HammeringVariant, RotationVariant
 
 
 class MethodData(BaseModel, abc.ABC):
@@ -135,11 +134,8 @@ def __repr__(self):
 
 # class Method(BaseModel, abc.ABC):
 class Method(BaseModel):
-    _flushing_variant: FlushingVariant | None = None
     _current_flushing_active_state: bool = False
-    _hammering_variant: HammeringVariant | None = None
     _current_hammer_active_state: bool = False
-    _rotation_variant: RotationVariant | None = None
     _current_increased_rotation_state: bool = False
 
     def __init__(self, **kwargs):
@@ -148,31 +144,6 @@ def __init__(self, **kwargs):
     def post_processing(self):
         pass
 
-    def detect_flushing_rule(self) -> FlushingVariant:
-        """
-        Call this with the method data before parsing the method data.
-
-        The result of calling this method will set the self.detected_flushing_variant to either use the "K" code if any
-        regulating the flushing are present, the "AR" (flushing on/off) code or the "I" (flushing pressure) code.
-        """
-        if self._flushing_variant:
-            return self._flushing_variant
-
-        if any(
-            [
-                row.comment_code in (72, 73, 76, 77)
-                or any(x in (72, 73, 76, 77) for x in self.extract_codes(row.remarks))
-                for row in self.method_data
-            ]
-        ):
-            self._flushing_variant = FlushingVariant.CODE_K
-        elif any([getattr(row, "flushing") is not None for row in self.method_data]):
-            self._flushing_variant = FlushingVariant.CODE_AR
-        else:
-            self._flushing_variant = FlushingVariant.CODE_I
-
-        return self._flushing_variant
-
     @classmethod
     def extract_codes(cls, remarks: str | None) -> tuple[int, ...]:
         """
@@ -195,10 +166,10 @@ def is_flushing_active(
 
         The following priority should be used to figure out if flushing is active:
 
-        1. Check K (kode) regulating flushing in file, use only K codes
-        2. If no K codes present in file, then check if "AR" code is present and has
+        1. Check K (kode) regulating flushing in file
+        2. If no K code present, then check if "AR" code is present and has
            a value (0 or 0.0 = off, 1 or 1.0 = on)
-        3. If no "AR" code is present in the file, then check if "I" (flushing pressure) > 0.1
+        3. If no "AR" code is present, then check if "I" (flushing pressure) > 0.1
         4. Otherwise, return False
 
         Codes used:
@@ -207,49 +178,28 @@ def is_flushing_active(
         Kode 76 (hammer and flushing on)
         Kode 77 (hammer and flushing off)
         """
-        if self._flushing_variant == FlushingVariant.CODE_K:
-            if data_row.comment_code in (72, 76) or any(x in (72, 76) for x in self.extract_codes(data_row.remarks)):
+        if data_row.comment_code in (72, 76) or any(x in (72, 76) for x in self.extract_codes(data_row.remarks)):
+            self._current_flushing_active_state = True
+            return self._current_flushing_active_state
+        # TODO: check remark for extra codes
+        elif data_row.comment_code in (73, 77) or any(x in (73, 77) for x in self.extract_codes(data_row.remarks)):
+            self._current_flushing_active_state = False
+            return self._current_flushing_active_state
+
+        if data_row.flushing is not None:
+            self._current_flushing_active_state = data_row.flushing
+            return self._current_flushing_active_state
+
+        if data_row.flushing_pressure is not None:
+            if data_row.flushing_pressure > Decimal("0.1"):
                 self._current_flushing_active_state = True
-            # TODO: check remark for extra codes
-            elif data_row.comment_code in (73, 77) or any(x in (73, 77) for x in self.extract_codes(data_row.remarks)):
+            else:
                 self._current_flushing_active_state = False
 
-        elif self._flushing_variant == FlushingVariant.CODE_AR:
-            if data_row.flushing is not None:
-                self._current_flushing_active_state = data_row.flushing
-
-        elif self._flushing_variant == FlushingVariant.CODE_I:
-            if data_row.flushing_pressure is not None:
-                if data_row.flushing_pressure > Decimal("0.1"):
-                    self._current_flushing_active_state = True
-                else:
-                    self._current_flushing_active_state = False
+            return self._current_flushing_active_state
 
         return self._current_flushing_active_state
 
-    def detect_hammering_rule(self) -> HammeringVariant | None:
-        """
-        Call this with the method data loaded before parsing the hammering in the method data.
-
-        The result of calling this method will set the self._hammering_variant to either use the "K" code if
-        any regulating the hammering are present, the "AP" (hammering on/off) code.
-        """
-        if self._hammering_variant:
-            return self._hammering_variant
-
-        if any(
-            [
-                row.comment_code in (74, 75, 76, 77)
-                or any(x in (74, 75, 76, 77) for x in self.extract_codes(row.remarks))
-                for row in self.method_data
-            ]
-        ):
-            self._hammering_variant = HammeringVariant.K
-        else:
-            self._hammering_variant = HammeringVariant.AP
-
-        return self._hammering_variant
-
     def is_hammer_active(
         self,
         data_row,  #: models.MethodCPTData| models.MethodTOTData| models.MethodRPData| models.MethodSRSData,
@@ -269,46 +219,18 @@ def is_hammer_active(
         Kode 76 (hammer and flushing on)
         Kode 77 (hammer and flushing off)
         """
-        if self._hammering_variant == HammeringVariant.K:
-            if data_row.comment_code in (74, 76) or any(x in (74, 76) for x in self.extract_codes(data_row.remarks)):
-                self._current_hammer_active_state = True
-                return self._current_hammer_active_state
-            elif data_row.comment_code in (75, 77) or any(x in (75, 77) for x in self.extract_codes(data_row.remarks)):
-                self._current_hammer_active_state = False
-                return self._current_hammer_active_state
-
-        elif self._hammering_variant == HammeringVariant.AP:
-            if data_row.hammering is not None:
-                self._current_hammer_active_state = data_row.hammering
-                return self._current_hammer_active_state
-
-        return self._current_hammer_active_state
-
-    def detect_increased_rotation_rule(self) -> RotationVariant | None:
-        """
-        Call this with the method data before updating the increased rotation in the method data.
+        if data_row.comment_code in (74, 76) or any(x in (74, 76) for x in self.extract_codes(data_row.remarks)):
+            self._current_hammer_active_state = True
+            return self._current_hammer_active_state
+        elif data_row.comment_code in (75, 77) or any(x in (75, 77) for x in self.extract_codes(data_row.remarks)):
+            self._current_hammer_active_state = False
+            return self._current_hammer_active_state
 
-        The result of calling this method will set the self._increased_rotation_variant to either use the
-        "K" code if any regulating the increased rotation are present, the "AQ" (increased rotation on/off) code
-        or the "R" (rotation rate) code.
-        """
+        if data_row.hammering is not None:
+            self._current_hammer_active_state = data_row.hammering
+            return self._current_hammer_active_state
 
-        if self._rotation_variant:
-            return self._rotation_variant
-
-        if any(
-            [
-                row.comment_code in (70, 71) or any(x in (70, 71) for x in self.extract_codes(row.remarks))
-                for row in self.method_data
-            ]
-        ):
-            self._rotation_variant = RotationVariant.K
-        elif any([row.increased_rotation_rate is not None for row in self.method_data]):
-            self._rotation_variant = RotationVariant.AQ
-        else:
-            self._rotation_variant = RotationVariant.R
-
-        return self._rotation_variant
+        return self._current_hammer_active_state
 
     def is_increased_rotation_active(
         self,
@@ -328,53 +250,24 @@ def is_increased_rotation_active(
         Kode 70 (increased rotation speed on)
         Kode 71 (increased rotation speed off)
         """
-        if self._rotation_variant == RotationVariant.K:
-            if data_row.comment_code == 70:
+        if data_row.comment_code == 70:
+            self._current_increased_rotation_state = True
+            return self._current_increased_rotation_state
+        elif data_row.comment_code == 71:
+            self._current_increased_rotation_state = False
+            return self._current_increased_rotation_state
+        if data_row.increased_rotation_rate is not None:
+            self._current_increased_rotation_state = data_row.increased_rotation_rate
+            return self._current_increased_rotation_state
+        if data_row.rotation_rate is not None:
+            if data_row.rotation_rate > 35:
                 self._current_increased_rotation_state = True
-            elif data_row.comment_code == 71:
+            else:
                 self._current_increased_rotation_state = False
-        elif self._rotation_variant == RotationVariant.AQ:
-            if data_row.increased_rotation_rate is not None:
-                self._current_increased_rotation_state = data_row.increased_rotation_rate
-        else:
-            if data_row.rotation_rate is not None:
-                if data_row.rotation_rate > 35:
-                    self._current_increased_rotation_state = True
-                else:
-                    self._current_increased_rotation_state = False
+            return self._current_increased_rotation_state
 
         return self._current_increased_rotation_state
 
-    def flushing_update(self):
-        """
-        Update flushing
-
-        """
-        self._flushing_variant = self.detect_flushing_rule()
-
-        for data in self.method_data:
-            data.flushing = self.is_flushing_active(data)
-
-    def hammering_update(self):
-        """
-        Update hammering
-
-        """
-        self._hammering_variant = self.detect_hammering_rule()
-
-        for data in self.method_data:
-            data.hammering = self.is_hammer_active(data)
-
-    def rotation_update(self):
-        """
-        Update rotation
-
-        """
-        self._rotation_variant = self.detect_increased_rotation_rule()
-
-        for data in self.method_data:
-            data.increased_rotation_rate = self.is_increased_rotation_active(data)
-
     @model_validator(mode="before")
     @classmethod
     def guess_date_format(cls, data: Any) -> Any:

src/sgf_parser/models/method_rp.py

@@ -39,16 +39,3 @@ def __init__(self, **kwargs):
     method_data_type: type[MethodRPData] = MethodRPData
 
     method_data: list[MethodRPData] = []
-
-    def post_processing(self):
-        """
-        Post-processing
-
-        """
-
-        if not self.method_data:
-            return
-
-        # Update flushing and increased rotation
-        self.flushing_update()
-        self.rotation_update()

src/sgf_parser/models/method_srs.py

@@ -93,20 +93,6 @@ def set_sounding_class(cls, data: Any) -> Any:
 
         return data
 
-    def post_processing(self):
-        """
-        Post-processing
-
-        """
-
-        if not self.method_data:
-            return
-
-        # Update flushing, hammering and increased rotation
-        self.flushing_update()
-        self.hammering_update()
-        self.rotation_update()
-
     @computed_field
     def depth_in_rock(self) -> Decimal | None:
         _rock_top_depth = None

src/sgf_parser/models/method_tot.py

@@ -159,17 +159,3 @@ def bedrock_elevation(self) -> Decimal | None:
             return Decimal(self.point_z) - _depth_in_soil
 
         return None
-
-    def post_processing(self):
-        """
-        Post-processing
-
-        """
-
-        if not self.method_data:
-            return
-
-        # Update flushing, hammering and increased rotation
-        self.flushing_update()
-        self.hammering_update()
-        self.rotation_update()

src/sgf_parser/models/types.py

@@ -81,35 +81,6 @@ class ParseState(enum.Enum):
     QUIT = 3
 
 
-class FlushingVariant(enum.StrEnum):
-    """
-    Flushing variants
-    """
-
-    CODE_K = "K"
-    CODE_AR = "AR"
-    CODE_I = "I"
-
-
-class HammeringVariant(enum.StrEnum):
-    """
-    Hammering variants
-    """
-
-    K = "K"
-    AP = "AP"
-
-
-class RotationVariant(enum.StrEnum):
-    """
-    Rotation variants
-    """
-
-    K = "K"
-    AQ = "AQ"
-    R = "R"
-
-
 class SoundingClass(enum.StrEnum):
     """
     Soil-Rock-Sounding (Swedish Jord-bergsondering) classes

src/sgf_parser/parser.py

@@ -142,4 +142,10 @@ def parse_header(self, header: dict[str, Any]) -> Method:
     def parse_data(self, method: Method, row: str) -> MethodData:
         row_dict = self._convert_str_to_dict(row)
         method_data = method.method_data_type.model_validate(row_dict)
+        if hasattr(method_data, "flushing"):
+                method_data.flushing = method.is_flushing_active(method_data)
+        if hasattr(method_data, "hammering"):
+            method_data.hammering = method.is_hammer_active(method_data)
+        if hasattr(method_data, "increased_rotation_rate"):
+            method_data.increased_rotation_rate = method.is_increased_rotation_active(method_data)
         return method_data

tests/integration/test_parse.py

@@ -510,9 +510,9 @@ def test_return_one_method_one_placeholder(
         datetime(2018, 11, 20, 10, 50,21),
         {
             # load=A, hammering=AP, load=W, turning=H, rotation_rate=R
-            Decimal("3.625"): {"load":Decimal("1.020"),  "turning":Decimal("152"), "penetration_rate": Decimal("154.208"), "comment_code": None, "remarks": None, "hammering": None},
-            Decimal("3.700"): {"load": Decimal("1.020"), "turning": Decimal("24"), "penetration_rate": Decimal("10.287"), "hammering": None,},
-            Decimal("3.775"): {"load": Decimal("5.681"), "turning": Decimal("24"), "penetration_rate": Decimal("1.506"), "hammering": None, "comment_code": 91, "remarks": "Sond kan ej drivas normalt"},
+            Decimal("3.625"): {"load":Decimal("1.020"),  "turning":Decimal("152"), "penetration_rate": Decimal("154.208"), "comment_code": None, "remarks": None, "hammering": False},
+            Decimal("3.700"): {"load": Decimal("1.020"), "turning": Decimal("24"), "penetration_rate": Decimal("10.287"), "hammering": False,},
+            Decimal("3.775"): {"load": Decimal("5.681"), "turning": Decimal("24"), "penetration_rate": Decimal("1.506"), "hammering": False, "comment_code": 91, "remarks": "Sond kan ej drivas normalt"},
         },
 ),
          ),