1.0.4.dev3 (Testing and logic improvements) (#75)

Author: mikeqfu

pyrcs/_base.py

@@ -376,7 +376,7 @@ def _make_file_pathname(self, data_name, ext=".pkl", data_dir=None, sub_dir=None
         :param ext: The file extension (including the leading dot), defaults to ``".pkl"``.
         :type ext: str
         :param data_dir: The directory where the file will be saved; defaults to ``None``.
-        :type data_dir: str | None
+        :type data_dir: str | os.PathLike | None
         :param sub_dir: A subdirectory name or a list of subdirectory names; defaults to ``None``.
         :type sub_dir: str | list | None
         :return: The pathname for saving the data file.
@@ -403,7 +403,7 @@ def _make_file_pathname(self, data_name, ext=".pkl", data_dir=None, sub_dir=None
             sub_dir_ = []
 
         if data_dir:
-            self.current_data_dir = validate_dir(path_to_dir=data_dir)
+            self.current_data_dir = validate_dir(path_to_dir=None if data_dir is True else data_dir)
             file_pathname = os.path.join(self.current_data_dir, *sub_dir_, filename)
 
         else:  # data_dir is None or data_dir == ""
@@ -431,7 +431,7 @@ def _save_data_to_file(self, data, data_name, ext=".pkl", dump_dir=None, sub_dir
         :type ext: str | bool
         :param dump_dir: The directory where the file should be saved;
             if ``None`` (default) a default directory within the class is used.
-        :type dump_dir: str | None
+        :type dump_dir: str | os.PathLike | None
         :param sub_dir: A subdirectory name or a list of subdirectory names; defaults to ``None``.
         :type sub_dir: str | list | None
         :param verbose: Whether to print detailed information to the console; defaults to ``False``.

pyrcs/collector.py

@@ -354,6 +354,7 @@ def update(self, confirmation_required=True, verbose=False, interval=5, init_upd
 
                 # Stations
                 print(f"\n{self.Stations.NAME}:")
+                _ = self.Stations.fetch_catalogue(**update_args)
                 _ = self.Stations.fetch_locations(**update_args)
 
                 time.sleep(interval)

pyrcs/data/.metadata

@@ -5,7 +5,7 @@
     "Author": "Qian Fu",
     "Affiliation": "University of Birmingham",
     "Email": "q.fu@bham.ac.uk",
-    "Version": "1.0.4.dev2",
+    "Version": "1.0.4.dev3",
     "License": "MIT",
     "First release": "August 2019"
 }

pyrcs/line_data/elr_mileage.py

@@ -3,7 +3,6 @@
 `Engineer's Line References (ELRs) <http://www.railwaycodes.org.uk/elrs/elr0.shtm>`_.
 """
 
-import copy
 import functools
 import itertools
 import os
@@ -473,46 +472,57 @@ def fetch_elr(self, initial=None, update=False, dump_dir=None, verbose=False, **
 
             data = {self.KEY: data_, self.KEY_TO_LAST_UPDATED_DATE: latest_update_date}
 
-            if dump_dir is not None:
+            if dump_dir:
                 self._save_data_to_file(
                     data=data, data_name=self.NAME, dump_dir=dump_dir, verbose=verbose)
 
         return data
 
-    def _mileage_file_dump_names(self, elr):
-        data_name = remove_punctuation(elr).lower()
+    def _dump_mileage_file(self, mileage_file, dump_dir=None, verbose=False):
+        """
+        Dump the collected mileage file data.
 
-        if data_name == "prn":
-            data_name += "_"
+        :param mileage_file: Data of the mileage file.
+        :type mileage_file: dict
+        :param dump_dir: The path to a directory where the mileage file data is saved;
+            defaults to ``False``.
+        :type dump_dir: str | os.PathLike | bool | None
+        :param verbose: Whether to print relevant information to the console; defaults to ``False``.
+        :type verbose: bool | int
+        """
 
-        dump_dir = self._cdd("mileage-files", data_name[0])
+        if dump_dir is False:
+            return None
 
-        return data_name, dump_dir
+        data_name = mileage_file['ELR'].lower()
+        data_name += ("_" if data_name == "prn" else "")
 
-    def _dump_mileage_file(self, elr, mileage_file, dump_it, verbose):
-        if dump_it:
-            data_name, dump_dir = self._mileage_file_dump_names(elr)
+        if dump_dir is None:
+            sub_dir = data_name[0]
+            target_dir = self._cdd("mileage-files", sub_dir)
+        else:
+            target_dir = dump_dir
 
-            self._save_data_to_file(
-                data=mileage_file, data_name=data_name, ext=".pkl", dump_dir=dump_dir,
-                verbose=verbose)
+        self._save_data_to_file(
+            data=mileage_file, data_name=data_name, ext=".pkl", dump_dir=target_dir,
+            verbose=verbose)
 
-    def _handle_err404(self, elr, notes_dat, parsed, dump_it, verbose):
+    def _handle_err404(self, elr, notes_dat, parsed, dump_dir=False, verbose=False):
         elr_alt = re.search(r'(?<= )[A-Z]{3}(\d)?', notes_dat).group(0)
         mileage_file_alt = self.collect_mileage_file(
-            elr=elr_alt, parsed=parsed, confirmation_required=False, dump_it=False,
+            elr=elr_alt, parsed=parsed, confirmation_required=False, dump_dir=False,
             verbose=verbose)
 
         if notes_dat.startswith('Now'):
-            mileage_file_former = copy.copy(mileage_file_alt)
+            mileage_file_former = mileage_file_alt.copy()
 
             mileage_file_alt.update({'Formerly': elr})
             self._dump_mileage_file(
-                elr_alt, mileage_file=mileage_file_alt, dump_it=dump_it, verbose=verbose)
+                mileage_file=mileage_file_alt, dump_dir=dump_dir, verbose=verbose)
 
-            mileage_file_former.update(({'Now': elr_alt}))
+            mileage_file_former.update(({'ELR': elr, 'Now': elr_alt}))
             self._dump_mileage_file(
-                elr, mileage_file=mileage_file_former, dump_it=dump_it, verbose=verbose)
+                mileage_file=mileage_file_former, dump_dir=dump_dir, verbose=verbose)
 
         return mileage_file_alt
 
@@ -753,16 +763,13 @@ def _parse_mileage_and_notes(self, content):
 
         return mileage_data, notes_data
 
-    def _collect_mileage_file(self, source, elr, parsed=True, dump_it=False, verbose=False):
+    def _collect_mileage_file(self, source, elr, parsed=True, dump_dir=False, verbose=False):
         soup = bs4.BeautifulSoup(markup=source.content, features='html.parser')
 
-        line_name = soup.find(name='h3').text
+        line_name = soup.find(name='h3').get_text(strip=True)
 
         sub_line_name_ = soup.find(name='h4')
-        if sub_line_name_ is not None:
-            sub_line_name = sub_line_name_.get_text()
-        else:
-            sub_line_name = ''
+        sub_line_name = sub_line_name_.get_text().strip() if sub_line_name_ is not None else ''
 
         err404 = {'"404" error: page not found', '404 error: page not found'}
         if any(x in err404 for x in {line_name, sub_line_name}):
@@ -771,7 +778,9 @@ def _collect_mileage_file(self, source, elr, parsed=True, dump_it=False, verbose
 
             notes_dat = elr_data['Notes'].iloc[0]
             if re.match(r'(Now( part of)? |= |See )[A-Z]{3}(\d)?$', notes_dat):
-                return self._handle_err404(elr, notes_dat, parsed, dump_it, verbose)
+                mileage_file_alt = self._handle_err404(
+                    elr=elr, notes_dat=notes_dat, parsed=parsed, dump_dir=dump_dir, verbose=verbose)
+                return mileage_file_alt
 
             else:
                 line_name, content = self._get_parsed_contents(elr_data, notes_dat)
@@ -809,12 +818,11 @@ def _collect_mileage_file(self, source, elr, parsed=True, dump_it=False, verbose
         if verbose in {True, 1}:
             print("Done.")
 
-        self._dump_mileage_file(
-            elr=elr, mileage_file=mileage_file, dump_it=dump_it, verbose=verbose)
+        self._dump_mileage_file(mileage_file=mileage_file, dump_dir=dump_dir, verbose=verbose)
 
         return mileage_file
 
-    def collect_mileage_file(self, elr, parsed=True, confirmation_required=True, dump_it=False,
+    def collect_mileage_file(self, elr, parsed=True, confirmation_required=True, dump_dir=False,
                              verbose=False, raise_error=False):
         """
         Collects the mileage file for a specific ELR from the source web page.
@@ -828,8 +836,9 @@ def collect_mileage_file(self, elr, parsed=True, confirmation_required=True, dum
             if ``confirmation_required=True`` (default), prompts the user for confirmation
             before proceeding with data collection.
         :type confirmation_required: bool
-        :param dump_it: Whether to save the collected data as a pickle file; defaults to ``False``.
-        :type dump_it: bool
+        :param dump_dir: The path to a directory where the mileage file data is saved;
+            if ``False`` (default), the data will not be dumped.
+        :type dump_dir: str | os.PathLike | bool | None
         :param verbose: Whether to print relevant information to the console; defaults to ``False``.
         :type verbose: bool | int
         :param raise_error: Whether to raise the provided exception;
@@ -843,7 +852,7 @@ def collect_mileage_file(self, elr, parsed=True, confirmation_required=True, dum
             - In some cases, mileages may be unknown and thus left blank
               (e.g. ``'ANI2, Orton Junction with ROB (~3.05)'``).
             - Mileages in parentheses are not on that ELR but are included for reference
-              (e.g., ``'ANL, (8.67) NORTHOLT [London Underground]'``).
+              (e.g. ``'ANL, (8.67) NORTHOLT [London Underground]'``).
             - As with the main ELR list, mileages preceded by a tilde (~) are approximate.
 
         **Examples**::
@@ -940,32 +949,32 @@ def collect_mileage_file(self, elr, parsed=True, confirmation_required=True, dum
             [4 rows x 8 columns]
         """
 
-        elr_ = remove_punctuation(elr).upper()
+        target_elr = remove_punctuation(elr).upper()
 
-        if elr_:
-            if confirmed(f"To collect mileage file of \"{elr_}\"\n?", confirmation_required):
+        if target_elr:
+            if confirmed(f'To collect mileage file of "{target_elr}"\n?', confirmation_required):
                 if verbose in {True, 1}:
                     message_ = "Collecting the mileage file"
                     if not confirmation_required:
-                        message_ += f' of "{elr_}"'
+                        message_ += f' of "{target_elr}"'
                     print(message_, end=" ... ")
 
                 try:
-                    url = homepage_url() + f'/elrs/_mileages/{elr_[0]}/{elr_}.shtm'.lower()
+                    url = urllib.parse.urljoin(
+                        homepage_url(),
+                        f'/elrs/_mileages/{target_elr[0]}/{target_elr}.shtm'.lower())
                     source = requests.get(url=url, headers=fake_requests_headers())
-
+                    source.raise_for_status()
                 except Exception as e:
                     print_instance_connection_error(verbose=verbose, e=e)
                     return None
 
                 try:
                     return self._collect_mileage_file(
-                        source=source, elr=elr_, parsed=parsed, dump_it=dump_it,
+                        source=source, elr=target_elr, parsed=parsed, dump_dir=dump_dir,
                         verbose=verbose)
-
                 except Exception as e:
-                    _print_failure_message(
-                        e=e, prefix="Errors:", verbose=verbose, raise_error=raise_error)
+                    _print_failure_message(e, "Errors:", verbose=verbose, raise_error=raise_error)
 
     def fetch_mileage_file(self, elr, update=False, dump_dir=None, verbose=False,
                            raise_error=False):
@@ -979,7 +988,7 @@ def fetch_mileage_file(self, elr, update=False, dump_dir=None, verbose=False,
         :type update: bool
         :param dump_dir: Path to the directory where the data file will be saved;
             defaults to ``None``.
-        :type dump_dir: str | None
+        :type dump_dir: str | os.PathLike | None
         :param verbose: Whether to print relevant information to the console; defaults to ``False``.
         :type verbose: bool | int
         :param raise_error: Whether to raise the provided exception;
@@ -992,9 +1001,12 @@ def fetch_mileage_file(self, elr, update=False, dump_dir=None, verbose=False,
         **Examples**::
 
             >>> from pyrcs.line_data import ELRMileages  # from pyrcs import ELRMileages
+            >>> import tempfile
+            >>> import pathlib
+            >>> tmp_path = pathlib.Path(tempfile.TemporaryDirectory().name)
             >>> em = ELRMileages()
             >>> # Get the mileage file of 'AAL' (Now 'NAJ3')
-            >>> aal_mileage_file = em.fetch_mileage_file(elr='AAL')
+            >>> aal_mileage_file = em.fetch_mileage_file(elr='AAL', dump_dir=tmp_path)
             >>> type(aal_mileage_file)
             dict
             >>> list(aal_mileage_file.keys())
@@ -1020,7 +1032,7 @@ def fetch_mileage_file(self, elr, update=False, dump_dir=None, verbose=False,
             12  18.0638               ...        DCL             81.12
             [13 rows x 8 columns]
             >>> # Get the mileage file of 'MLA'
-            >>> mla_mileage_file = em.fetch_mileage_file(elr='MLA')
+            >>> mla_mileage_file = em.fetch_mileage_file(elr='MLA', dump_dir=tmp_path)
             >>> type(mla_mileage_file)
             dict
             >>> list(mla_mileage_file.keys())
@@ -1044,25 +1056,28 @@ def fetch_mileage_file(self, elr, update=False, dump_dir=None, verbose=False,
             3  0.1606                      0.73  ...         None
             [4 rows x 8 columns]
             >>> # Get the mileage file of 'LCG'
-            >>> mla_mileage_file = em.fetch_mileage_file(elr='LCG')
+            >>> mla_mileage_file = em.fetch_mileage_file(elr='LCG', dump_dir=tmp_path)
         """
 
         try:
-            elr_ = remove_punctuation(elr)
-            data_name, _ = self._mileage_file_dump_names(elr_)
-            ext = ".pkl"
-            path_to_pickle = self._cdd("mileage-files", data_name[0], data_name + ext, mkdir=False)
+            target_elr = remove_punctuation(elr)
+
+            data_name = target_elr.lower()
+            data_name += ("_" if data_name == "prn" else "")
+            sub_dir, ext = data_name[0], ".pkl"
+
+            path_to_file = self._cdd("mileage-files", sub_dir, f"{data_name}{ext}", mkdir=False)
 
-            if os.path.isfile(path_to_pickle) and not update:
-                mileage_file = load_data(path_to_pickle)
+            if os.path.isfile(path_to_file) and not update:
+                mileage_file = load_data(path_to_file)
 
             else:
                 verbose_ = get_collect_verbosity_for_fetch(data_dir=dump_dir, verbose=verbose)
                 mileage_file = self.collect_mileage_file(
-                    elr=elr_, parsed=True, confirmation_required=False, dump_it=True,
+                    elr=target_elr, parsed=True, confirmation_required=False, dump_dir=None,
                     verbose=verbose_)
 
-            if dump_dir is not None:
+            if dump_dir not in {False, None}:
                 self._save_data_to_file(
                     data=mileage_file, data_name=data_name, ext=ext, dump_dir=dump_dir,
                     verbose=verbose)

pyrcs/other_assets/tunnel.py

@@ -90,40 +90,41 @@ def _parse_length(x):
         """
 
         if '✖' in x:
-            x, note_ = x.split('✖')
-            note_ = ' (' + note_ + ')'
+            x, note_suffix = x.split('✖')
+            note_suffix = f' ({note_suffix.strip()})'
         else:
-            note_ = ''
+            note_suffix = ''
 
-        if re.match(r'[Uu]nknown', x):
-            length = np.nan
-            note = 'Unknown'
+        if x == '':
+            length, note = np.nan, 'Unavailable'
 
-        elif x == '':
-            length = np.nan
-            note = 'Unavailable'
+        elif re.match(r'[Uu]nknown', x):
+            length, note = np.nan, 'Unknown'
 
         elif re.match(r'\d+m \d+yd?(- | to )?.*\d+m \d+yd?.*', x):
             miles_a, yards_a, miles_b, yards_b = re.findall(r'\d+', x)
             length_a = float(miles_a) * 1609.344 + float(yards_a) * 0.9144
-            # measurement.measures.Distance(mi=miles_a).m + measurement.measures.Distance(yd=yards_a).m
+            # from measurement.measures import Distance
+            # Distance(mi=miles_a).m + Distance(yd=yards_a).m
             length_b = float(miles_b) * 1609.344 + float(yards_b) * 0.9144
-            # measurement.measures.Distance(mi=miles_b).m + measurement.measures.Distance(yd=yards_b).m
+            # Distance(mi=miles_b).m + Distance(yd=yards_b).m
             length = (length_a + length_b) / 2
             note = '-'.join([str(round(length_a, 2)), str(round(length_b, 2))]) + ' metres'
 
         else:
-            if re.match(r'(formerly )?c?≈?\d+m ?\d+yd?|ch', x):
+            if re.match(r'(formerly )?c?≈?\d+m ?\d+(?:yd?|ch)', x):
                 miles, yards = re.findall(r'\d+', x)
                 if re.match(r'.*\d+ch$', x):  # "yards" is "chains"
-                    yards = yards * 22  # measurement.measures.Distance(chain=yards).yd
+                    yards = float(yards) * 22  # measurement.measures.Distance(chain=yards).yd
 
                 if re.match(r'^c.*|^≈', x):
                     note = 'Approximate'
                 elif re.match(r'\d+y$', x):
                     note = re.search(r'(?<=\dy).*$', x).group(0)
                 elif re.match(r'^(formerly).*', x):
                     note = 'Formerly'
+                elif re.match(r'.*(?:yd?|ch)\s*(.+)$', x):
+                    note = re.search(r'(?:yd?|ch)\s*(.+)$', x).group(1)
                 else:
                     note = ''
 
@@ -139,7 +140,7 @@ def _parse_length(x):
             # measurement.measures.Distance(mi=miles).m + measurement.measures.Distance(yd=yards).m
 
             if note != '':
-                note = note + note_
+                note = note.strip() + note_suffix
 
         return length, note