CU-869574kvp update snomed preprocessing naming

medcat/utils/preprocess_snomed.py

@@ -74,10 +74,12 @@ class Snomed:
         uk_drug_ext (bool, optional): Specifies whether the version is a SNOMED UK drug extension. Defaults to False.
         au_ext (bool, optional): Specifies wether the version is a AU release. Defaults to False.
     """
+    SNOMED_RELEASE_PATTERN = re.compile("^SnomedCT_([A-Za-z0-9]+)_([A-Za-z0-9]+)_(\d{8}T\d{6}Z$)")
+    NO_VERSION_DETECTED = 'N/A'
 
     def __init__(self, data_path, uk_ext=False, uk_drug_ext=False, au_ext: bool = False):
         self.data_path = data_path
-        self.release = data_path[-16:-8]
+        self.release = self._determine_release(data_path, strict=False)
         self.uk_ext = uk_ext
         self.uk_drug_ext = uk_drug_ext
         self.opcs_refset_id = "1126441000000105"
@@ -95,6 +97,17 @@ def __init__(self, data_path, uk_ext=False, uk_drug_ext=False, au_ext: bool = Fa
             raise ValueError("Cannot both be a UK and and a AU version. "
                              f"Got UK={uk_ext}, UK_Drug={uk_drug_ext}, AU={au_ext}")
 
+    @classmethod
+    def _determine_release(cls, folder_path: str, strict: bool = True,
+                           _group_nr: int = 3, _keep_chars: int = 8) -> str:
+        folder_basename = os.path.basename(folder_path)
+        match = cls.SNOMED_RELEASE_PATTERN.match(folder_basename)
+        if match is None and strict:
+            raise UnkownSnomedReleaseException(f"No version found in '{folder_path}'")
+        elif match is None:
+            return cls.NO_VERSION_DETECTED
+        return match.group(_group_nr)[:_keep_chars]
+
     def to_concept_df(self):
         """
         Create a SNOMED CT concept DataFrame.
@@ -368,7 +381,8 @@ def _check_path_and_release(self):
             for folder in os.listdir(self.data_path):
                 if "SnomedCT" in folder:
                     paths.append(os.path.join(self.data_path, folder))
-                    snomed_releases.append(folder[-16:-8])
+                    rel = self._determine_release(folder, strict=True)
+                    snomed_releases.append(rel)
         if len(paths) == 0:
             raise FileNotFoundError('Incorrect path to SNOMED CT directory')
         return paths, snomed_releases
@@ -447,3 +461,9 @@ def _map_snomed2refset(self):
             return icd10_df, opcs_df
         else:
             return mapping_df
+
+
+class UnkownSnomedReleaseException(ValueError):
+
+    def __init__(self, *args) -> None:
+        super().__init__(*args)

tests/utils/test_preprocess_snomed.py

@@ -1,3 +1,4 @@
+import os
 from typing import Dict
 from medcat.utils import preprocess_snomed
 
@@ -62,3 +63,72 @@ def test_old_gets_old_OPCS4_mapping_uk_ext(self):
     def test_new_gets_new_OCPS4_mapping_uk_ext(self):
         snomed = preprocess_snomed.Snomed(EXAMPLE_SNOMED_PATH_NEW, uk_ext=True)
         self.assertEqual(snomed.opcs_refset_id, "1382401000000109")
+
+
+class TestSnomedModelGetter(unittest.TestCase):
+    WORKING_BASE_NAMES = [
+        "SnomedCT_InternationalRF2_PRODUCTION_20240201T120000Z",
+        "SnomedCT_InternationalRF2_PRODUCTION_20240601T120000Z",
+        "SnomedCT_UKClinicalRF2_PRODUCTION_20240410T000001Z",
+        "SnomedCT_UKClinicalRefsetsRF2_PRODUCTION_20240410T000001Z",
+        "SnomedCT_UKDrugRF2_PRODUCTION_20240508T000001Z",
+        "SnomedCT_UKEditionRF2_PRODUCTION_20240410T000001Z",
+        "SnomedCT_UKEditionRF2_PRODUCTION_20240508T000001Z",
+        "SnomedCT_Release_AU1000036_20240630T120000Z",
+    ]
+    FAILING_BASE_NAMES = [
+        "uk_sct2cl_38.2.0_20240605000001Z",
+        "uk_sct2cl_32.6.0_20211027000001Z",
+    ]
+    PATH = os.path.join("path", "to", "release")
+
+    def _pathify(self, in_list: list) -> list:
+        return [os.path.join(self.PATH, folder) for folder in in_list]
+
+    def assert_got_version(self, snomed: preprocess_snomed.Snomed, raw_name: str):
+        rel = snomed.release
+        self.assertIsInstance(rel, str)
+        self.assertIn(rel, raw_name)
+        self.assertEqual(rel, raw_name[-16:-8])
+
+    def assert_all_work(self, all_paths: list):
+        for path in all_paths:
+            with self.subTest(f"Rrelease name: {path}"):
+                snomed = preprocess_snomed.Snomed(path)
+                self.assert_got_version(snomed, path)
+
+    def test_gets_model_form_basename(self):
+        self.assert_all_work(self.WORKING_BASE_NAMES)
+
+    def test_gets_model_from_path(self):
+        full_paths = self._pathify(self.WORKING_BASE_NAMES)
+        self.assert_all_work(full_paths)
+
+    def assert_raises(self, folder_path: str):
+        with self.assertRaises(preprocess_snomed.UnkownSnomedReleaseException):
+            preprocess_snomed.Snomed._determine_release(folder_path, strict=True)
+
+    def assert_all_raise(self, folder_paths: list):
+        for folder_path in folder_paths:
+            with self.subTest(f"Folder: {folder_path}"):
+                self.assert_raises(folder_path)
+
+    def test_fails_on_incorrect_names_strict(self):
+        self.assert_all_raise(self.FAILING_BASE_NAMES)
+
+    def test_fails_on_incorrect_paths_strict(self):
+        full_paths = self._pathify(self.FAILING_BASE_NAMES)
+        self.assert_all_raise(full_paths)
+
+    def assert_all_get_no_version(self, folder_paths: list):
+        for folder_path in folder_paths:
+            with self.subTest(f"Folder: {folder_path}"):
+                snomed = preprocess_snomed.Snomed(folder_path)
+                self.assertEqual(snomed.release, preprocess_snomed.Snomed.NO_VERSION_DETECTED)
+
+    def test_gets_no_version_incorrect_names_nonstrict(self):
+        self.assert_all_get_no_version(self.FAILING_BASE_NAMES)
+
+    def test_gets_no_version_incorrect_paths_nonstrict(self):
+        full_paths = self._pathify(self.FAILING_BASE_NAMES)
+        self.assert_all_get_no_version(full_paths)