refactor reprocess_files partially #34

swvo/io/dst/wdc.py

@@ -17,7 +17,7 @@
 
 import numpy as np
 import pandas as pd
-import wget
+import requests
 
 logger = logging.getLogger(__name__)
 
@@ -82,35 +82,48 @@ def download_and_process(self, start_time: datetime, end_time: datetime, reproce
         temporary_dir = Path("./temp_wdc")
         temporary_dir.mkdir(exist_ok=True, parents=True)
 
-        try:
-            file_paths, time_intervals = self._get_processed_file_list(start_time, end_time)
+        file_paths, time_intervals = self._get_processed_file_list(start_time, end_time)
 
-            for file_path, time_interval in zip(file_paths, time_intervals):
-                filename = "index.html"
-
-                URL = self.URL.replace("YYYYMM", time_interval.strftime("%Y%m"))
+        for file_path, time_interval in zip(file_paths, time_intervals):
+            filename = "index.html"
+            if file_path.exists() and not reprocess_files:
+                continue
 
-                if file_path.exists():
-                    if reprocess_files:
-                        file_path.unlink()
-                    else:
-                        continue
+            tmp_path = file_path.with_suffix(file_path.suffix + ".tmp")
 
-                logger.debug(f"Downloading file {URL + filename} ...")
+            URL = self.URL.replace("YYYYMM", time_interval.strftime("%Y%m"))
 
-                wget.download(URL + filename, str(temporary_dir))
+            if file_path.exists():
+                if reprocess_files:
+                    file_path.unlink()
+                else:
+                    continue
 
+            try:
+                logger.debug(f"Downloading file {URL} ...")
+                response = requests.get(URL)
+                response.raise_for_status()
+                data = response.text.splitlines()
+                with open(temporary_dir / filename, "w") as file:
+                    file.write("\n".join(data))
                 logger.debug("Processing file ...")
 
                 processed_df = self._process_single_file(
                     temporary_dir / filename,
                     year=time_interval.year,
                     month=time_interval.month,
                 )
-                processed_df.to_csv(file_path, index=True, header=True)
+                processed_df.to_csv(tmp_path, index=True, header=True)
+                tmp_path.replace(file_path)
+            except Exception as e:
+                logger.error(f"Failed to process {file_path}: {e}")
+                if tmp_path.exists():
+                    tmp_path.unlink()
+                    pass
+                continue
 
-        finally:
-            rmtree(temporary_dir, ignore_errors=True)
+            finally:
+                rmtree(temporary_dir, ignore_errors=True)
 
     def _get_processed_file_list(self, start_time: datetime, end_time: datetime) -> Tuple[List, List]:
         """Get list of file paths and their corresponding time intervals.

swvo/io/f10_7/swpc.py

@@ -18,7 +18,7 @@
 
 import numpy as np
 import pandas as pd
-import wget
+import requests
 
 logger = logging.getLogger(__name__)
 
@@ -95,22 +95,18 @@ def download_and_process(self) -> None:
         temp_dir = Path("./temp_f107")
         temp_dir.mkdir(exist_ok=True)
 
-        try:
-            logger.debug("Downloading F10.7 data...")
+        logger.debug("Downloading F10.7 data...")
+        self._download(temp_dir, self.NAME_F107)
 
-            wget.download(self.URL + self.NAME_F107, str(temp_dir))
+        logger.debug("Processing F10.7 data...")
 
-            if Path(temp_dir / self.NAME_F107).stat().st_size == 0:
-                msg = f"Error downloading file: {self.URL + self.NAME_F107}"
-                raise FileNotFoundError(msg)
+        new_data = self._process_single_file(temp_dir / self.NAME_F107)
 
-            logger.debug("Processing F10.7 data...")
-
-            new_data = self._process_single_file(temp_dir / self.NAME_F107)
-
-            for year, year_data in new_data.groupby(new_data.date.dt.year):
-                file_path = self.data_dir / f"SWPC_F107_{year}.csv"
+        for year, year_data in new_data.groupby(new_data.date.dt.year):
+            file_path = self.data_dir / f"SWPC_F107_{year}.csv"
+            tmp_path = file_path.with_suffix(file_path.suffix + ".tmp")
 
+            try:
                 if file_path.expanduser().exists():
                     logger.debug(f"Updating {file_path}...")
 
@@ -127,11 +123,44 @@ def download_and_process(self) -> None:
                     logger.debug(f"Creating new file for {year}")
                     combined_data = year_data
 
-                combined_data.to_csv(file_path, index=False)
+                combined_data.to_csv(tmp_path, index=False)
+                tmp_path.replace(file_path)
+
+            except Exception as e:
+                logger.error(f"Failed to process file for year {year}: {e}")
+                if tmp_path.exists():
+                    tmp_path.unlink()
+                continue
+
+            finally:
+                shutil.rmtree(temp_dir, ignore_errors=True)
+
+    def _download(self, temp_dir: Path, filename: str) -> None:
+        """Download a file from SWPC server.
+
+        Parameters
+        ----------
+        temp_dir : Path
+            Temporary directory to store the downloaded file.
+        filename : str
+            Name of the file to download.
+
+        Raises
+        ------
+        requests.HTTPError
+            If the HTTP request fails.
+        FileNotFoundError
+            If the downloaded file is empty.
+        """
+        response = requests.get(self.URL + filename)
+        response.raise_for_status()
+
+        with open(temp_dir / filename, "wb") as f:
+            f.write(response.content)
 
-        finally:
-            # ...
-            shutil.rmtree(temp_dir, ignore_errors=True)
+        if (temp_dir / filename).stat().st_size == 0:
+            msg = f"Error downloading file: {self.URL + filename}"
+            raise FileNotFoundError(msg)
 
     def _process_single_file(self, file_path: Path) -> pd.DataFrame:
         """Read and process the F10.7 data file.

swvo/io/hp/gfz.py

@@ -13,7 +13,7 @@
 
 import numpy as np
 import pandas as pd
-import wget
+import requests
 
 logger = logging.getLogger(__name__)
 
@@ -91,40 +91,69 @@ def download_and_process(
         temporary_dir = Path("./temp_hp_wget")
         temporary_dir.mkdir(exist_ok=True, parents=True)
 
-        try:
-            file_paths, time_intervals = self._get_processed_file_list(start_time, end_time)
+        file_paths, time_intervals = self._get_processed_file_list(start_time, end_time)
+
+        for file_path, time_interval in zip(file_paths, time_intervals):
+            if file_path.exists() and not reprocess_files:
+                continue
+
+            tmp_path = file_path.with_suffix(file_path.suffix + ".tmp")
 
-            for file_path, time_interval in zip(file_paths, time_intervals):
-                filenames_download = [
-                    f"Hp{self.index_number}/Hp{self.index_number}_ap{self.index_number}_{time_interval[0].year!s}.txt"
-                ]
+            filenames_download = [
+                f"Hp{self.index_number}/Hp{self.index_number}_ap{self.index_number}_{time_interval[0].year!s}.txt"
+            ]
 
-                # there is a separate nowcast file
-                if time_interval[0].year == datetime.now(timezone.utc).year:
-                    filenames_download.append(
-                        f"Hp{self.index_number}/Hp{self.index_number}_ap{self.index_number}_nowcast.txt"
-                    )
+            # there is a separate nowcast file
+            if time_interval[0].year == datetime.now(timezone.utc).year:
+                filenames_download.append(
+                    f"Hp{self.index_number}/Hp{self.index_number}_ap{self.index_number}_nowcast.txt"
+                )
 
+            try:
                 for filename_download in filenames_download:
-                    logger.debug(f"Downloading file {self.URL + filename_download} ...")
+                    self._download(temporary_dir, filename_download)
 
-                    wget.download(self.URL + filename_download, str(temporary_dir))
+                filenames_download = [x.split("/")[-1] for x in filenames_download]  # strip folder from filename
 
-                    logger.debug("Processing file ...")
+                processed_df = self._process_single_file(temporary_dir, filenames_download)
 
-                    if file_path.exists():
-                        if reprocess_files:
-                            file_path.unlink()
-                        else:
-                            continue
+                file_path.parent.mkdir(parents=True, exist_ok=True)
+                processed_df.to_csv(tmp_path, index=True, header=False)
+                tmp_path.replace(file_path)
 
-                filenames_download = [x[5:] for x in filenames_download]  # strip of folder of filename
+            except Exception as e:
+                logger.error(f"Failed to process {file_path}: {e}")
+                if tmp_path.exists():
+                    tmp_path.unlink()
+                continue
 
-                processed_df = self._process_single_file(temporary_dir, filenames_download)
-                processed_df.to_csv(file_path, index=True, header=False)
+            finally:
+                rmtree(temporary_dir, ignore_errors=True)
+
+    def _download(self, temporary_dir: Path, filename: str) -> None:
+        """Download a file from the GFZ server.
+
+        Parameters
+        ----------
+        temporary_dir : Path
+            Temporary directory to store the downloaded file.
+        filename : str
+            Full path of the file to download (including folder).
+
+        Raises
+        ------
+        requests.HTTPError
+            If the HTTP request fails.
+        """
+        logger.debug(f"Downloading file {self.URL + filename} ...")
+
+        response = requests.get(self.URL + filename)
+        response.raise_for_status()
 
-        finally:
-            rmtree(temporary_dir)
+        # Extract just the filename from the path
+        filename_only = filename.split("/")[-1]
+        with open(temporary_dir / filename_only, "wb") as f:
+            f.write(response.content)
 
     def read(self, start_time: datetime, end_time: datetime, *, download: bool = False) -> pd.DataFrame:
         """Read HpGFZ data for the given time range.

swvo/io/kp/niemegk.py

@@ -16,7 +16,7 @@
 
 import numpy as np
 import pandas as pd
-import wget
+import requests
 
 logger = logging.getLogger(__name__)
 
@@ -86,26 +86,22 @@ def download_and_process(self, start_time: datetime, end_time: datetime, reproce
         temporary_dir = Path("./temp_kp_niemegk_wget")
         temporary_dir.mkdir(exist_ok=True, parents=True)
 
-        try:
-            logger.debug(f"Downloading file {self.URL + self.NAME} ...")
+        logger.debug(f"Downloading file {self.URL + self.NAME} ...")
 
-            wget.download(self.URL + self.NAME, str(temporary_dir))
-
-            # check if download was successfull
-            if os.stat(str(temporary_dir / self.NAME)).st_size == 0:
-                raise FileNotFoundError(f"Error while downloading file: {self.URL + self.NAME}!")
-
-            logger.debug("Processing file ...")
-            processed_df = self._process_single_file(temporary_dir)
+        file_paths, time_intervals = self._get_processed_file_list(start_time, end_time)
+        for file_path, time_interval in zip(file_paths, time_intervals):
+            if file_path.exists() and not reprocess_files:
+                continue
+            tmp_path = file_path.with_suffix(file_path.suffix + ".tmp")
+            try:
+                self._download(temporary_dir)
 
-            file_paths, time_intervals = self._get_processed_file_list(start_time, end_time)
+                # check if download was successfull
+                if os.stat(str(temporary_dir / self.NAME)).st_size == 0:
+                    raise FileNotFoundError(f"Error while downloading file: {self.URL + self.NAME}!")
 
-            for file_path, time_interval in zip(file_paths, time_intervals):
-                if file_path.exists():
-                    if reprocess_files:
-                        file_path.unlink()
-                    else:
-                        continue
+                logger.debug("Processing file ...")
+                processed_df = self._process_single_file(temporary_dir)
 
                 data_single_file = processed_df[
                     (processed_df.index >= time_interval[0]) & (processed_df.index <= time_interval[1])
@@ -115,12 +111,26 @@ def download_and_process(self, start_time: datetime, end_time: datetime, reproce
                     continue
 
                 file_path.parent.mkdir(parents=True, exist_ok=True)
-                data_single_file.to_csv(file_path, index=True, header=False)
+                data_single_file.to_csv(tmp_path, index=True, header=False)
+                tmp_path.replace(file_path)
 
                 logger.debug(f"Saving processed file {file_path}")
+            except Exception as e:
+                logger.error(f"Failed to process {file_path}: {e}")
+                if tmp_path.exists():
+                    tmp_path.unlink()
+                    pass
+                continue
+
+            finally:
+                rmtree(temporary_dir, ignore_errors=True)
+
+    def _download(self, temporary_dir):
+        response = requests.get(self.URL + self.NAME, str(temporary_dir))
+        response.raise_for_status()
 
-        finally:
-            rmtree(temporary_dir)
+        with open(temporary_dir / self.NAME, "w") as f:
+            f.write(response.text)
 
     def read(self, start_time: datetime, end_time: datetime, download: bool = False) -> pd.DataFrame:
         """Read Niemegk Kp data for the specified time range.