Merge pull request #162 from mraspaud/fix-remote-segments

Ensure filesystem is passed for remote file checks

Author: mraspaud

pytroll_collectors/harvest_EUM_schedules.py

@@ -21,7 +21,7 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-"""Harvest EUM schedules. Download schedules from EUM and parse to limit gatherer"""
+"""Harvest EUM schedules. Download schedules from EUM and parse to limit gatherer."""
 
 import re
 import os
@@ -94,13 +94,14 @@ def _generate_pass_list_file_name(params, save_basename, eum_base_url):
     else:
         logger.error("sensor not given in params in granule_metadata. Can not continue.")
         return (None, None)
-    eum_url = EUM_BASE_URL + pass_list_file
+    eum_url = eum_base_url + pass_list_file
     save_file = os.path.join(save_basename, pass_list_file)
     logger.debug("Pass list save file, %s", save_file)
     return eum_url, save_file
 
 
 def harvest_schedules(params, save_basename=None, eum_base_url=EUM_BASE_URL):
+    """Harvest schedules."""
     if save_basename is None:
         save_basename = tempfile.gettempdir()
     logger.debug("harvest_schedules params: %s", params)

pytroll_collectors/region_collector.py

@@ -28,10 +28,7 @@
 
 from pyresample import parse_area_file
 
-try:
-    from trollsched.satpass import Pass
-except ImportError:
-    Pass = None
+from trollsched.satpass import Pass
 
 import logging
 

pytroll_collectors/segments.py

@@ -66,7 +66,7 @@ class Status(Enum):
     SLOT_OBSOLETE_TIMEOUT = 4
 
 
-DO_NOT_COPY_KEYS = ("uid", "uri", "channel_name", "segment", "sensor")
+DO_NOT_COPY_KEYS = ("uid", "uri", "channel_name", "segment", "sensor", "filesystem", "path")
 REMOVE_TAGS = {'path', 'segment'}
 
 
@@ -397,7 +397,12 @@ def _add_message_to_metadata(self, message):
     def _add_file_info_to_metadata(self, metadata, message):
         msg_data = message.message_data
         if message.type == 'file':
-            metadata['dataset'].append({'uri': msg_data['uri'], 'uid': msg_data['uid']})
+            mda = {'uri': msg_data['uri'], 'uid': msg_data['uid']}
+            if "filesystem" in msg_data:
+                mda["filesystem"] = msg_data["filesystem"]
+            if "path" in msg_data:
+                mda["path"] = msg_data["path"]
+            metadata['dataset'].append(mda)
         elif message.type == 'dataset':
             metadata['dataset'].extend(message.message_data['dataset'])
         else:
@@ -898,11 +903,11 @@ def _add_existing_files_to_slot(self, slot, fnames, message):
 def _get_existing_files_from_message(message):
     mask = message.pattern.parser.globify({})
     url_parts = urlparse(message.message_data["uri"])
+    storage_options = message.message_data.get("filesystem")
+    return _fsspec_glob(url_parts, mask, storage_options)
 
-    return _fsspec_glob(url_parts, mask)
 
-
-def _fsspec_glob(url_parts, mask):
+def _fsspec_glob(url_parts, mask, storage_options):
     import fsspec
 
     pattern = urlunparse(
@@ -915,8 +920,8 @@ def _fsspec_glob(url_parts, mask):
             ''
         )
     )
-
-    fs_ = fsspec.filesystem(url_parts.scheme)
+    storage_options = storage_options or dict()
+    fs_ = fsspec.filesystem(url_parts.scheme, **storage_options)
     files = fs_.glob(pattern)
     # There might be no scheme in the returned filenames, so add it if scheme is defined
     if url_parts.scheme:

pytroll_collectors/tests/test_region_collector.py

@@ -67,6 +67,7 @@
 _granule_metadata_metop_b = {"platform_name": "Metop-B",
                              "sensor": "avhrr"}
 
+
 def granule_metadata(s_min):
     """Return common granule_metadata dictionary."""
     return {**_granule_metadata,
@@ -125,17 +126,29 @@ def europe_collector_schedule_cut_custom_method_failed(europe, schedule_cut=True
     return RegionCollector(europe, schedule_cut=schedule_cut, schedule_cut_method=schedule_cut_method)
 
 
-def _fakeopen(url):
+def _fakeopen_celestrak(url):
     return io.BytesIO(tles)
 
 
+ears_avhrr_pass_predictions = b"""EARS-AVHRR Pass Predictions
+2025/06/19 05:01:28
+regionBeginEumetsat,regionEndEumetsat,Satellite
+2025-06-19 00:20,2025-06-19 00:37,metopc
+2025-06-19 01:10,2025-06-19 01:28,metopb
+"""
+
+
+def _fakeopen_eum(url):
+    return io.BytesIO(ears_avhrr_pass_predictions)
+
+
 def test_init(europe):
     """Test that initialisation appears to work."""
     from pytroll_collectors.region_collector import RegionCollector
     RegionCollector(europe)
 
 
-@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen)
+@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen_celestrak)
 def test_collect(europe_collector, caplog):
     """Test that granules can be collected."""
     with caplog.at_level(logging.DEBUG):
@@ -150,7 +163,7 @@ def test_collect(europe_collector, caplog):
     assert "Granule file://18 is not overlapping euro_ma"
 
 
-@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen)
+@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen_celestrak)
 def test_collect_duration(europe):
     """Test with tle_platform_name, without end_time, using call syntax."""
     from pytroll_collectors.region_collector import RegionCollector
@@ -165,7 +178,8 @@ def test_collect_duration(europe):
     alt_europe_collector(granule_metadata)
 
 
-@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen)
+@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen_celestrak)
+@unittest.mock.patch("pytroll_collectors.harvest_EUM_schedules.urlopen", new=_fakeopen_eum)
 def test_collect_check_schedules(europe_collector_schedule_cut, caplog):
     """Test default schedule cut method."""
     with caplog.at_level(logging.DEBUG):
@@ -180,7 +194,7 @@ def test_collect_check_schedules(europe_collector_schedule_cut, caplog):
     assert "harvest_EUM_schedules.py'>, with type <class 'module'>" in caplog.text
 
 
-@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen)
+@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen_celestrak)
 def test_collect_check_schedules_custom_method(europe_collector_schedule_cut_custom_method, caplog):
     """Test custom schedule cut method."""
     with caplog.at_level(logging.DEBUG):
@@ -196,7 +210,7 @@ def test_collect_check_schedules_custom_method(europe_collector_schedule_cut_cus
     assert "test_region_collector.py'>, with type <class 'module'>" in caplog.text
 
 
-@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen)
+@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen_celestrak)
 def test_collect_check_schedules_custom_method_failed(europe_collector_schedule_cut_custom_method_failed, caplog):
     """Test custom schedule cut method failed import."""
     with caplog.at_level(logging.DEBUG):
@@ -208,16 +222,16 @@ def test_collect_check_schedules_custom_method_failed(europe_collector_schedule_
     assert test_string in caplog.text
 
 
-@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen)
+@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen_celestrak)
 def test_collect_missing_tle_from_file(europe_collector, caplog):
-    """Test that granules can be collected, but missing TLE raises and exception"""
+    """Test that granules can be collected, but missing TLE raises and exception."""
     with caplog.at_level(logging.DEBUG):
         for s_min in (0, 3, 6, 9, 12, 15, 18):
             with pytest.raises(KeyError):
                 europe_collector.collect({**granule_metadata_metop_b(s_min)})
 
 
-@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen)
+@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen_celestrak)
 def test_adjust_timeout(europe, caplog):
     """Test timeout adjustment."""
     from pytroll_collectors.region_collector import RegionCollector
@@ -241,7 +255,7 @@ def test_adjust_timeout(europe, caplog):
 
 
 @pytest.mark.skip(reason="test never finishes")
-@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen)
+@unittest.mock.patch("pyorbital.tlefile.urlopen", new=_fakeopen_celestrak)
 def test_faulty_end_time(europe_collector, caplog):
     """Test adapting if end_time before start_time."""
     granule_metadata = {

pytroll_collectors/tests/test_segments.py

@@ -1455,7 +1455,7 @@ def test_floor_grouping_does_not_affect_other_pattern(self):
 FCI_CONFIG = """patterns:
   fci_nc:
     pattern:
-      "W_XX-EUMETSAT-Darmstadt,IMG+SAT,{platform_name:4s}+FCI-1C-RRAD-FDHSI-FD--CHK-{segment_type}---NC4E_C_EUMT_{processing_time}_GTT_DEV_{start_time:%Y%m%d%H%M%S}_{end_time:%Y%m%d%H%M%S}_N_{special_compression}_T_{repeat_cycle_in_day:>04d}_{segment:0>4s}.nc"
+      "W_XX-EUMETSAT-Darmstadt,IMG+SAT,{platform_name:4s}+FCI-1C-RRAD-FDHSI-FD--CHK-{segment_type}--{locality}-NC4E_C_EUMT_{processing_time}_{processor}_{processor_status}_{start_time:%Y%m%d%H%M%S}_{end_time:%Y%m%d%H%M%S}_N_{special_compression}_{service_status}_{repeat_cycle_in_day:>04d}_{segment:0>4s}.nc"
     critical_files:
     wanted_files: :0001-0040
     all_files: :0001-0040
@@ -1503,3 +1503,41 @@ def test_process_message_fci_with_nonzero_seconds(self, caplog):
         assert len(slot.output_metadata["dataset"]) == 2
         uids = set(info["uid"] for info in slot.output_metadata["dataset"])
         assert uids == expected_uids
+
+
+fci_remote = ('pytroll://1b/FCI-segment/FDHSI file user@collecting.host.se 2025-06-26T07:39:40.483302+00:00 v1.2 application/json {"sensor": "fci", "filesystem": {"cls": "fsspec.implementations.sftp:SFTPFileSystem", "protocol": "sftp", "args": [], "host": "receiving_host"}, "uri": "ssh:///local_disk/tellicast/received/TER-1/T01-MTG-1/W_XX-EUMETSAT-Darmstadt,IMG+SAT,MTI1+FCI-1C-RRAD-FDHSI-FD--CHK-BODY--DIS-NC4E_C_EUMT_20250626073728_IDPFI_OPE_20250626073332_20250626073411_N_JLS_O_0046_0017.nc", "path": "/local_disk/tellicast/received/TER-1/T01-MTG-1/W_XX-EUMETSAT-Darmstadt,IMG+SAT,MTI1+FCI-1C-RRAD-FDHSI-FD--CHK-BODY--DIS-NC4E_C_EUMT_20250626073728_IDPFI_OPE_20250626073332_20250626073411_N_JLS_O_0046_0017.nc", "uid": "W_XX-EUMETSAT-Darmstadt,IMG+SAT,MTI1+FCI-1C-RRAD-FDHSI-FD--CHK-BODY--DIS-NC4E_C_EUMT_20250626073728_IDPFI_OPE_20250626073332_20250626073411_N_JLS_O_0046_0017.nc", "pflag": "W", "location_indicator": "XX-EUMETSAT-Darmstadt", "data_designator": "IMG+SAT", "spacecraft_id": 1, "data_source": "FCI", "coverage": "FD", "subsetting": "", "component1": "CHK", "component3": "", "purpose": "DIS", "format": "NC4E", "oflag": "C", "originator": "EUMT", "processing_time": "2025-06-26T07:37:28", "facility_or_tool": "IDPFI", "environment": "OPE", "start_time": "2025-06-26T07:33:32", "end_time": "2025-06-26T07:34:11", "processing_mode": "N", "special_compression": "JLS", "disposition_mode": "O", "repeat_cycle_in_day": 46, "count_in_repeat_cycle": 17}')  # noqa
+
+
+FCI_CONFIG_WITH_CHECK_ON_START = FCI_CONFIG + "\ncheck_existing_files_after_start: true\n"
+
+
+@patch("fsspec.filesystem")
+def test_remote_file_with_filesystem_passes_filesystem_info(filesystem):
+    """Test that remote filesystem info is passed from message when available."""
+    from posttroll.message import Message as Message_p
+    import yaml
+
+    segment_gatherer = SegmentGatherer(yaml.safe_load(FCI_CONFIG_WITH_CHECK_ON_START))
+
+    expected_uids = set()
+
+    for fci_message in [fci_remote]:
+        fci_msg = Message_p(rawstr=fci_message)
+        expected_uids.add(fci_msg.data['uid'])
+        msg = FakeMessage(fci_msg.data)
+        segment_gatherer.process(msg)
+
+    timestamp = '2025-06-26 07:30:00'
+    print(segment_gatherer.slots)
+    assert timestamp in segment_gatherer.slots
+    storage_options = dict(cls="fsspec.implementations.sftp:SFTPFileSystem",
+                           protocol="sftp",
+                           args=[],
+                           host="receiving_host")
+    filesystem.assert_called_once_with("ssh", **storage_options)
+
+    assert segment_gatherer.slots[timestamp].output_metadata["dataset"][0]["filesystem"] == storage_options
+    assert "filesystem" not in segment_gatherer.slots[timestamp].output_metadata
+    path = "/local_disk/tellicast/received/TER-1/T01-MTG-1/W_XX-EUMETSAT-Darmstadt,IMG+SAT,MTI1+FCI-1C-RRAD-FDHSI-FD--CHK-BODY--DIS-NC4E_C_EUMT_20250626073728_IDPFI_OPE_20250626073332_20250626073411_N_JLS_O_0046_0017.nc"  # noqa
+    assert segment_gatherer.slots[timestamp].output_metadata["dataset"][0]["path"] == path
+    assert "path" not in segment_gatherer.slots[timestamp].output_metadata