IO - Change origin attribute when not find on system

Orange/data/io_util.py

@@ -1,3 +1,4 @@
+import os.path
 import subprocess
 from collections import defaultdict
 
@@ -6,7 +7,7 @@
 
 from Orange.data import (
     is_discrete_values, MISSING_VALUES, Variable,
-    DiscreteVariable, StringVariable, ContinuousVariable, TimeVariable,
+    DiscreteVariable, StringVariable, ContinuousVariable, TimeVariable, Table,
 )
 from Orange.misc.collections import natural_sorted
 
@@ -207,3 +208,44 @@ def mapvalues(arr):
         values = [_var.parse(i) for i in orig_values]
 
     return values, var
+
+
+def _extract_new_origin(attr, table, lookup_dirs):
+    # origin exist
+    if os.path.exists(attr.attributes["origin"]):
+        return attr.attributes["origin"]
+
+    # last dir of origin in lookup dirs
+    dir_ = os.path.basename(os.path.normpath(attr.attributes["origin"]))
+    for ld in lookup_dirs:
+        new_dir = os.path.join(ld, dir_)
+        if os.path.exists(new_dir):
+            return new_dir
+
+    # all first part of column content in lookup dirs
+    for ld in lookup_dirs:
+        if all(os.path.join(ld, v) for v in table.get_column(attr)):
+            return ld
+
+
+def update_origin(table: Table, file_path: str) -> Table:
+    """
+    Lookup dirs:
+    1. Dir where CSV
+    2. Parent dir (CSV in root of parent dir)
+
+    Possible situations:
+    1. Last dir of origin in lookup dirs
+    2. Any first part of column content in lookup dirs
+    """
+    file_dir = os.path.dirname(file_path)
+    parent_dir = os.path.dirname(file_dir)
+    # if file_dir already root file_dir == parent_dir
+    lookup_dirs = tuple({file_dir, parent_dir})
+    for attr in table.domain:
+        if "origin" in attr.attributes:
+            new_orig = _extract_new_origin(attr, table, lookup_dirs)
+            if new_orig:
+                attr.attributes["origin"] = new_orig
+    return table
+

Orange/data/tests/test_io_util.py

@@ -1,6 +1,12 @@
+import os.path
 import unittest
+from tempfile import TemporaryDirectory
 
-from Orange.data import ContinuousVariable, guess_data_type
+import numpy as np
+
+from Orange.data import ContinuousVariable, guess_data_type, Table, Domain, \
+    StringVariable
+from Orange.data.io_util import update_origin
 
 
 class TestIoUtil(unittest.TestCase):
@@ -10,5 +16,58 @@ def test_guess_continuous_w_nans(self):
             ContinuousVariable)
 
 
+class TestUpdateOrigin(unittest.TestCase):
+    FILE_NAMES = [["file1.txt"], ["file2.txt"], ["file3.text"]]
+
+    def setUp(self) -> None:
+        self.alt_dir = adr = TemporaryDirectory()
+        # create empty files
+        for f in self.FILE_NAMES:
+            with open(os.path.join(adr.name, f[0]), 'w'):
+                pass
+
+    def tearDown(self) -> None:
+        self.alt_dir.cleanup()
+
+    def test_origin_not_changed(self):
+        """
+        Origin exist; keep it unchanged, even though dataset path also includes
+        files from column.
+        """
+        with TemporaryDirectory() as dir_name:
+            var = StringVariable("Files")
+            var.attributes["origin"] = dir_name
+            table = Table.from_list(Domain([], metas=[var]), self.FILE_NAMES)
+            update_origin(table, self.alt_dir.name)
+            self.assertEqual(table.domain[var].attributes["origin"], dir_name)
+
+    def test_origin_subdir(self):
+        """Origin is wrong but last dir in origin exit in the dataset file's path"""
+        var = StringVariable("Files")
+        var.attributes["origin"] = f"/a/b/{os.path.basename(self.alt_dir.name)}"
+        table = Table.from_list(Domain([], metas=[var]), self.FILE_NAMES)
+        update_origin(table, self.alt_dir.name)
+        self.assertEqual(table.domain[var].attributes["origin"], self.alt_dir.name)
+
+    def test_origin_parents_subdir(self):
+        """
+        Origin is wrong but last dir in origin exit in the dataset file parents's path
+        """
+        # make the dir where dataset is placed
+        dataset_dir = os.path.join(self.alt_dir.name, "subdir")
+
+        var = StringVariable("Files")
+        var.attributes["origin"] = f"/a/b/{os.path.basename(self.alt_dir.name)}"
+        table = Table.from_list(Domain([], metas=[var]), self.FILE_NAMES)
+        update_origin(table, dataset_dir)
+        self.assertEqual(table.domain[var].attributes["origin"], self.alt_dir.name)
+
+    def test_column_paths_subdir(self):
+        pass
+
+    def test_column_paths_parents_subdir(self):
+        pass
+
+
 if __name__ == '__main__':
     unittest.main()