csvpath

  39class CsvPath(ErrorCollector, Printer):  # pylint: disable=R0902, R0904
  40    """CsvPath represents a csvpath string that contains a reference to
  41    a file, scanning instructions, and rules for matching lines.
  42    """
  43
  44    # re R0902, R0904: reasonable, but not a priority
  45
  46    def __init__(  # pylint: disable=R0913
  47        self,
  48        *,
  49        csvpaths=None,
  50        delimiter=",",
  51        quotechar='"',
  52        skip_blank_lines=True,
  53        print_default=True,
  54        config=None,
  55        error_manager=None,
  56    ):
  57        # re: R0913: all reasonable pass-ins with sensible defaults
  58        #
  59        # the config.ini file loaded as a ConfigParser instance
  60        #
  61        # definitely do not want this coming from CsvPaths because
  62        # we want to be able to override config.ini specifically for
  63        # this instance, if needed; however, we do want to be able
  64        # to pass in a config object that has been configured in some
  65        # way.
  66        self._config = config
  67        self.scanner = None
  68        """ @private """
  69        self.matcher = None
  70        """ @private """
  71        #
  72        # a parent CsvPaths may manage a CsvPath instance. if so, it will enable
  73        # the use of named files and named paths, print capture, error handling,
  74        # results collection, reference handling, etc. if a CsvPaths is not present
  75        # the CsvPath instance is responsible for all its own upkeep and does not
  76        # have some of those capabilities.
  77        #
  78        self.csvpaths = csvpaths
  79        """ @private """
  80        #
  81        # there are two logger components one for CsvPath and one for CsvPaths.
  82        # the default levels are set in config.ini. to change the levels pass LogUtility
  83        # your component instance and the logging level. e.g.:
  84        # LogUtility.logger(csvpath, "debug")
  85        #
  86        self.logger = LogUtility.logger(self)
  87        """ @private """
  88        self.logger.info("initialized CsvPath")
  89        #
  90        # if we don't have a csvpaths these will both be None
  91        #
  92        self.named_paths_name = None
  93        """ @private """
  94        self.named_file_name = None
  95        """ @private """
  96        #
  97        # all errors come to our manager if they occur during matching. we use
  98        # the CsvPaths manager if possible. Otherwise, we just make our own that
  99        # only knows how to collect errors, not distribute them.
 100        #
 101        self.ecoms = ErrorCommunications(csvpath=self)
 102        """ @private """
 103        self.error_manager = ErrorManager(csvpath=self)
 104        if csvpaths is not None:
 105            self.error_manager.add_internal_listener(csvpaths.error_manager)
 106        #
 107        # modes are set in external comments
 108        #
 109        self.modes = ModeController(self)
 110        """ @private """
 111        #
 112        # captures the number of lines up front and tracks line stats as the
 113        # run progresses
 114        #
 115        self._line_monitor = None
 116        #
 117        # the scanning part of the csvpath. e.g. $test.csv[*]
 118        #
 119        self.scan = None
 120        """ @private """
 121        #
 122        # the matching part of the csvpath. e.g. [yes()]
 123        #
 124        self.match = None
 125        """ @private """
 126        #
 127        # when True the lines that do not match are returned from next()
 128        # and collect(). this effectively switches CsvPath from being an
 129        # create an OR expression in this case. in the default, we say:
 130        #     are all of these things true?
 131        # but when collect_when_not_matched is True we ask:
 132        #     are any of these things not true?
 133        #
 134        # self._when_not_matched = False
 135        self._headers = None
 136        self.variables: Dict[str, Any] = {}
 137        self.delimiter = delimiter
 138        self.quotechar = quotechar
 139        #
 140        # a blank line has no headers. it has no data. physically it is 2 \n with
 141        # nothing but whitespace between them. any data or any delimiters would make
 142        # the line non-blank.
 143        #
 144        self.skip_blank_lines = skip_blank_lines
 145        #
 146        # in the case of a [*] scan where the last line is blank we would miss firing
 147        # last() unless we take steps. instead, we allow that line to match, but we
 148        # do not return a line to the caller of next() and we freeze the variables
 149        # there is room for side effects make changes, but that a reasonable compromise
 150        # between missing last and allowing unwanted changes. we definitely do not
 151        # freeze is_valid or stop, which can be useful signaling, even in an
 152        # inconsistent state.
 153        #
 154        self._freeze_path = False
 155        #
 156        # counts are 1-based
 157        #
 158        self.scan_count = 0
 159        self.match_count = 0
 160        #
 161        # used by stop() and advance(). a stopped CsvPath halts without finishing
 162        # its run. an advancing CsvPath doesn't consider the match part of the
 163        # csvpath and does not incur any side effects as it progresses through the
 164        # rows the advance skips. the skip() function has the same effect as
 165        # advance(1) but without any guarantee that the other match components on
 166        # the line will be considered before skipping ahead. there are likely
 167        # corner cases where an onmatch qualifier or some other constraint will
 168        # trigger match components that would otherwise be skipped so the ability
 169        # to shortcut some of the match should not be relied on for anything
 170        # critical.
 171        #
 172        self.stopped = False
 173        self._advance = 0
 174        #
 175        # the lines var will hold a reference to a LineSpooler instance during a
 176        # run, if lines are being collected by the collect() method. if the user
 177        # is using this CsvPath instance directly the LineSpooler is only there
 178        # as a proxy to the list of lines being collected -- we don't spool to
 179        # disk, at least atm.
 180        #
 181        self.lines = None
 182        """ @private """
 183        #
 184        # set by fail()
 185        #
 186        self._is_valid = True
 187        #
 188        # basic timing for the CsvPath instance only. if the CsvPath is managed
 189        # by a CsvPaths the timings for a run may include time spent by other
 190        # CsvPath instances.
 191        #
 192        self.last_row_time = -1
 193        """ @private """
 194        self.rows_time = -1
 195        """ @private """
 196        self.total_iteration_time = -1
 197        """ @private """
 198        #
 199        # limiting collection means returning fewer headers (values in the
 200        # line, a.k.a columns) then are available. limiting headers returned
 201        # can impact named results, reset_headers(), and other considerations.
 202        #
 203        self._limit_collection_to = []
 204        #
 205        # error collecting is at the CsvPath instance by default. CsvPath
 206        # instances that are managed by a CsvPaths have their errors collected
 207        # by their Results as well. Result handles persistence.
 208        #
 209        # errors policies are set in config.ini at CsvPath and CsvPaths levels.
 210        #
 211        self._errors: List[Error] = []
 212        #
 213        # saves the scan and match parts of paths for reference. mainly helpful
 214        # for testing the CsvPath library itself; not used end users. the run
 215        # name becomes the file name of the saved path parts.
 216        #
 217        self._save_scan_dir = None
 218        self._save_match_dir = None
 219        self._run_name = None
 220        #
 221        # metadata is collected from "outer" csvpath comments. outer comments
 222        # separate from the comments within the match part of the csvpath.
 223        # the keys are words with colons. e.g. ~ name: my new csvpath ~
 224        #
 225        self.metadata: Dict[str, Any] = {}
 226        #
 227        # holds the current match count while we're in the middle of a match
 228        # so that anyone who wants to can increase the match count using
 229        # raise_match_count_if(). it is important to do the raise asap so that
 230        # components that are onmatched have the right match count available.
 231        #
 232        self._current_match_count = 0
 233        #
 234        # printers receive print lines from the print function. the default
 235        # printer prints to standard out. a CsvPath that is managed by a
 236        # CsvPaths has its Results as a printer, as well as having
 237        # the default printer.
 238        #
 239        self.printers = []
 240        """ @private """
 241        if print_default:
 242            self.printers.append(StdOutPrinter())
 243        #
 244        # _function_times_match collects the time a function spends doing its matches()
 245        #
 246        self._function_times_match = {}
 247        #
 248        # _function_times_value collects the time a function spends doing its to_value()
 249        #
 250        self._function_times_value = {}
 251        self._created_at = datetime.now(timezone.utc)
 252        self._run_started_at = None
 253        self._collecting = False
 254        #
 255        # holds the unmatched lines when lines are being collected and
 256        # _unmatched_available is True. it is analogous to the lines returned
 257        # by collect(), but is the lines not returned by collect().
 258        #
 259        self._unmatched = None
 260
 261    #
 262    # this method saves and reloads the config. if you don't want that use
 263    # CsvPath.config.save_to_config().
 264    #
 265    def add_to_config(self, section, key, value) -> None:
 266        """@private"""
 267        self.config.add_to_config(section=section, key=key, value=value)
 268        self.config.save_config()
 269        self.config.reload()
 270
 271    @property
 272    def data_from_preceding(self) -> bool:
 273        """@private"""
 274        return self.modes.source_mode.value
 275
 276    @data_from_preceding.setter
 277    def data_from_preceding(self, dfp: bool) -> None:
 278        """@private"""
 279        self.modes.source_mode.value = dfp
 280
 281    @property
 282    def unmatched(self) -> list[list[Any]]:
 283        """@private"""
 284        return self._unmatched
 285
 286    @unmatched.setter
 287    def unmatched(self, lines: list[list[Any]]) -> None:
 288        """@private"""
 289        self._unmatched = lines
 290
 291    @property
 292    def collecting(self) -> bool:
 293        """@private"""
 294        return self._collecting
 295
 296    @collecting.setter
 297    def collecting(self, c: bool) -> None:
 298        """@private"""
 299        self._collecting = c
 300
 301    @property
 302    def unmatched_available(self) -> bool:
 303        """@private"""
 304        return self.modes.unmatched_mode.value
 305
 306    @unmatched_available.setter
 307    def unmatched_available(self, ua: bool) -> None:
 308        """@private"""
 309        self.modes.unmatched_mode.value = ua
 310
 311    @property
 312    def created_at(self) -> datetime:
 313        """@private"""
 314        return self._created_at
 315
 316    @property
 317    def run_started_at(self) -> datetime:
 318        """@private"""
 319        return self._run_started_at
 320
 321    @property
 322    def will_run(self) -> bool:
 323        """@private"""
 324        return self.modes.run_mode.value
 325
 326    @will_run.setter
 327    def will_run(self, mode) -> None:
 328        """@private"""
 329        self.modes.run_mode.value = mode
 330
 331    #
 332    # increases the total accumulated time spent doing c.matches() by t
 333    #
 334    def up_function_time_match(self, c, t) -> None:
 335        """@private"""
 336        if c not in self.function_times_match:
 337            self.function_times_match[c] = 0
 338        st = self.function_times_match[c]
 339        st += t
 340        self.function_times_match[c] = st
 341
 342    @property
 343    def function_times_match(self) -> int:
 344        """@private"""
 345        return self._function_times_match
 346
 347    #
 348    # increases the total accumulated time spent doing c.to_value() by t
 349    #
 350    def up_function_time_value(self, c, t) -> None:
 351        """@private"""
 352        if c not in self.function_times_value:
 353            self.function_times_value[c] = 0
 354        st = self.function_times_value[c]
 355        st += t
 356        self.function_times_value[c] = st
 357
 358    @property
 359    def function_times_value(self) -> int:
 360        """@private"""
 361        return self._function_times_value
 362
 363    def do_i_raise(self) -> bool:
 364        """@private"""
 365        return self.ecoms.do_i_raise()
 366
 367    @property
 368    def advance_count(self) -> int:  # pragma: no cover
 369        """@private"""
 370        return self._advance
 371
 372    @advance_count.setter
 373    def advance_count(self, lines: int) -> None:
 374        """@private"""
 375        self._advance = lines
 376
 377    @property
 378    def headers(self) -> List[str]:
 379        """@private"""
 380        if self._headers is None:
 381            self.get_total_lines_and_headers()
 382        return self._headers
 383
 384    @headers.setter
 385    def headers(self, headers: List[str]) -> None:
 386        """@private"""
 387        self._headers = headers
 388
 389    @property
 390    def line_monitor(self) -> LineMonitor:
 391        """@private"""
 392        if self._line_monitor is None:
 393            self.get_total_lines_and_headers()
 394        return self._line_monitor
 395
 396    @line_monitor.setter
 397    def line_monitor(self, lm) -> None:
 398        """@private"""
 399        self._line_monitor = lm
 400
 401    @property
 402    def AND(self) -> bool:  # pylint: disable=C0103
 403        return self.modes.logic_mode.value
 404
 405    @AND.setter
 406    def AND(self, a: bool) -> bool:  # pylint: disable=C0103
 407        self.modes.logic_mode.value = a
 408
 409    @property
 410    def OR(self) -> bool:  # pylint: disable=C0103
 411        return not self.modes.logic_mode.value
 412
 413    @OR.setter
 414    def OR(self, a: bool) -> bool:  # pylint: disable=C0103
 415        self.modes.logic_mode.value = not a
 416
 417    @property
 418    def identity(self) -> str:
 419        """returns id or name if found in metadata.
 420
 421        the id or name gets into metadata primarily if found
 422        in an "external" comment in the csvpath. "external"
 423        meaning outside the []s. comments are keyword:comment.
 424        we take id, Id, ID and name, Name, NAME.
 425
 426        id is preferred over name. E.g. in:
 427        ~ name: my path description: an example id: this value wins ~
 428        the id becomes the identity of the instance.
 429
 430        we prefer in this order: all-lower most, Initial-caps,
 431        ALL-CAPS least
 432
 433        the ordering is relied on in Result and possibly
 434        elsewhere.
 435        """
 436        ret = None
 437        if not self.metadata:
 438            ret = ""
 439        if "NAME" in self.metadata:
 440            ret = self.metadata["NAME"]
 441        if "Name" in self.metadata:
 442            ret = self.metadata["Name"]
 443        if "name" in self.metadata:
 444            ret = self.metadata["name"]
 445        if "ID" in self.metadata:
 446            ret = self.metadata["ID"]
 447        if "Id" in self.metadata:
 448            ret = self.metadata["Id"]
 449        if "id" in self.metadata:
 450            ret = self.metadata["id"]
 451        return ret
 452
 453    @property
 454    def config(self) -> Config:  # pylint: disable=C0116
 455        """@private"""
 456        if not self._config:
 457            self._config = Config()
 458        return self._config
 459
 460    # ==========================
 461    # Errors
 462    # <thinking> if we have a csvpaths people should look at the result to find errors
 463    # but we give access to metadata, vars, etc. from the csvpath, so we should
 464    # give errors too. that means we need to have our own listener. ultimately we'd
 465    # just be adding pointers, not dup the original error data.
 466    #
 467    def metadata_update(self, mdata: Metadata) -> None:
 468        """@private"""
 469        if isinstance(mdata, Error):
 470            self.collect_error(mdata)
 471
 472    @property
 473    def errors(self) -> List[Error]:  # pylint: disable=C0116
 474        return self._errors
 475
 476    @property
 477    def errors_count(self) -> int:  # pylint: disable=C0116
 478        return len(self.errors)
 479
 480    def collect_error(self, error: Error) -> None:  # pylint: disable=C0116
 481        """@private"""
 482        self.errors.append(error)
 483
 484    def has_errors(self) -> bool:
 485        return self.errors_count > 0
 486
 487    @property
 488    def stop_on_validation_errors(self) -> bool:
 489        """@private"""
 490        return self.modes.validation_mode.stop_on_validation_errors
 491
 492    @property
 493    def fail_on_validation_errors(self) -> bool:
 494        """@private"""
 495        return self.modes.validation_mode.fail_on_validation_errors
 496
 497    @property
 498    def print_validation_errors(self) -> bool:
 499        """@private"""
 500        return self.modes.validation_mode.print_validation_errors
 501
 502    @property
 503    def log_validation_errors(self) -> bool:
 504        """@private"""
 505        return self.modes.validation_mode.log_validation_errors
 506
 507    @property
 508    def raise_validation_errors(self) -> bool:
 509        """@private"""
 510        return self.modes.validation_mode.raise_validation_errors
 511
 512    @property
 513    def match_validation_errors(self) -> bool:
 514        """@private"""
 515        return self.modes.validation_mode.match_validation_errors
 516
 517    @property
 518    def collect_validation_errors(self) -> bool:
 519        """@private"""
 520        return self.modes.validation_mode.collect_validation_errors
 521
 522    def add_printer(self, printer) -> None:  # pylint: disable=C0116
 523        """@private"""
 524        if printer not in self.printers:
 525            self.printers.append(printer)
 526
 527    def set_printers(self, printers: List) -> None:  # pylint: disable=C0116
 528        """@private"""
 529        self.printers = printers
 530
 531    @property
 532    def has_default_printer(self) -> bool:
 533        """@private"""
 534        if not self.printers:
 535            self.printers = []
 536        for p in self.printers:
 537            if isinstance(p, StdOutPrinter):
 538                return True
 539        return False
 540
 541    def print(self, string: str) -> None:  # pylint: disable=C0116
 542        """@private"""
 543        for p in self.printers:
 544            p.print(string)
 545
 546    def print_to(self, name: str, string: str) -> None:
 547        """@private"""
 548        for p in self.printers:
 549            p.print_to(name, string)
 550
 551    @property
 552    def last_line(self):
 553        """@private
 554        this method only returns the default printer's last_line"""
 555        if not self.printers or len(self.printers) == 0:
 556            return None
 557        return self.printers[0].last_line
 558
 559    @property
 560    def lines_printed(self) -> int:
 561        """@private
 562        this method only returns the default printer's lines printed"""
 563        if not self.printers or len(self.printers) == 0:
 564            return -1
 565        return self.printers[0].lines_printed
 566
 567    @property
 568    def is_frozen(self) -> bool:
 569        """@private
 570        True if the instance is matching on its last row only to
 571        allow last()s to run; in which case, no variable updates
 572        are allowed, along with other limitations."""
 573        return self._freeze_path
 574
 575    @is_frozen.setter
 576    def is_frozen(self, freeze: bool) -> None:
 577        """@private"""
 578        self._freeze_path = freeze
 579
 580    @property
 581    def explain(self) -> bool:
 582        """@private
 583        when this property is True CsvPath dumps a match explaination
 584        to INFO. this can be expensive. a 25% performance hit wouldn't
 585        be unexpected.
 586        """
 587        return self.modes.explain_mode.value
 588
 589    @explain.setter
 590    def explain(self, yesno: bool) -> None:
 591        """@private"""
 592        self.modes.explain_mode.value = yesno
 593
 594    @property
 595    def collect_when_not_matched(self) -> bool:
 596        """@private
 597        when this property is True CsvPath returns the lines that do not
 598        match the matchers match components"""
 599        return self.modes.return_mode.collect_when_not_matched
 600
 601    @collect_when_not_matched.setter
 602    def collect_when_not_matched(self, yesno: bool) -> None:
 603        """@private
 604        when c ollect_when_not_matched is True we return the lines that failed
 605        to match, rather than the default behavior of returning the matches.
 606        """
 607        self.modes.return_mode.collect_when_not_matched = yesno
 608
 609    def parse(self, csvpath, disposably=False):
 610        """@private
 611        displosably is True when a Matcher is needed for some purpose other than
 612        the run we were created to do. could be that a match component wanted a
 613        parsed csvpath for its own purposes. when True, we create and return the
 614        Matcher, but then forget it ever existed.
 615
 616        when disposably is False we build the scanner and return that
 617        """
 618        #
 619        # strip off any comments and collect any metadata
 620        # CsvPaths will do this earlier but it stripped off
 621        # the comments so we won't find them again
 622        #
 623        csvpath = MetadataParser(self).extract_metadata(instance=self, csvpath=csvpath)
 624        self.update_settings_from_metadata()
 625        #
 626        #
 627        # exp!
 628        if disposably is False:
 629            csvpath = self._update_file_path(csvpath)
 630        #
 631        #
 632        #
 633        s, mat = self._find_scan_and_match_parts(csvpath)
 634        #
 635        # a disposable matcher still needs the match part
 636        #
 637        self.match = mat
 638        if disposably:
 639            pass
 640        else:
 641            self.scan = s
 642            self.scanner = Scanner(csvpath=self)
 643            self.scanner.parse(s)
 644        #
 645        # we build a matcher to see if it builds without error.
 646        # in principle we could keep this as the actual matcher.
 647        # atm, tho, just create a dry-run copy. in some possible
 648        # unit tests we may not have a parsable match part.
 649        #
 650        if disposably:
 651            matcher = None
 652            if mat:
 653                matcher = Matcher(csvpath=self, data=mat, line=None, headers=None)
 654            #
 655            # if the matcher was requested for some reason beyond our own needs
 656            # we just return it and forget it existed.
 657            #
 658            return matcher
 659        if self.scanner.filename is None:
 660            raise FileException("Cannot proceed without a filename")
 661        self.get_total_lines_and_headers()
 662        return self
 663
 664    def update_settings_from_metadata(self) -> None:
 665        """@private"""
 666        #
 667        # settings:
 668        #   - logic-mode: AND | OR
 669        #   - return-mode: matches | no-matches
 670        #   - print-mode: default | no-default
 671        #   - validation-mode: (no-)print | log | (no-)raise | quiet | (no-)match
 672        #   - run-mode: no-run | run
 673        #   - unmatched-mode: no-keep | keep
 674        #   - source-mode: preceding | origin
 675        #   - files-mode: all | no-data | no-unmatched | no-printouts | data | unmatched | errors | meta | vars | printouts
 676        #
 677        self.modes.update()
 678
 679    # =====================
 680    # in principle the modes should come through the mode controller like:
 681    #      self.modes.transfer_mode.value
 682    # not wading into that today. low value.
 683    #
 684    @property
 685    def transfer_mode(self) -> str:
 686        """@private"""
 687        return self.metadata.get("transfer-mode")
 688
 689    @property
 690    def source_mode(self) -> str:
 691        """@private"""
 692        return self.metadata.get("source-mode")
 693
 694    @property
 695    def error_mode(self) -> str:
 696        """@private"""
 697        return self.metadata.get("error-mode")
 698
 699    @property
 700    def files_mode(self) -> str:
 701        """@private"""
 702        return self.metadata.get("files-mode")
 703
 704    @property
 705    def validation_mode(self) -> str:
 706        """@private"""
 707        return self.metadata.get("validation-mode")
 708
 709    @property
 710    def run_mode(self) -> str:
 711        """@private"""
 712        return self.metadata.get("run-mode")
 713
 714    @property
 715    def logic_mode(self) -> str:
 716        """@private"""
 717        return self.metadata.get("logic-mode")
 718
 719    @property
 720    def return_mode(self) -> str:
 721        """@private"""
 722        return self.modes.get("return-mode")
 723
 724    @property
 725    def explain_mode(self) -> str:
 726        """@private"""
 727        return self.metadata.get("explain-mode")
 728
 729    @property
 730    def print_mode(self) -> str:
 731        """@private"""
 732        return self.metadata.get("print-mode")
 733
 734    @property
 735    def unmatched_mode(self) -> str:
 736        """@private"""
 737        return self.metadata.get("unmatched-mode")
 738
 739    # =====================
 740
 741    @property
 742    def transfers(self) -> list[tuple[str, str]]:
 743        """@private"""
 744        return self.modes.transfer_mode.transfers
 745
 746    @property
 747    def all_expected_files(self) -> list[str]:
 748        """@private"""
 749        return self.modes.files_mode.all_expected_files
 750
 751    @all_expected_files.setter
 752    def all_expected_files(self, efs: list[str]) -> None:
 753        """@private"""
 754        self.modes.files_mode.all_expected_files = efs
 755
 756    def _pick_named_path(self, name, *, specific=None) -> str:
 757        """@private"""
 758        if not self.csvpaths:
 759            raise CsvPathsException("No CsvPaths object available")
 760        np = self.csvpaths.paths_manager.get_named_paths(name)
 761        if not np:
 762            raise CsvPathsException(f"Named-paths '{name}' not found")
 763        if len(np) == 0:
 764            raise CsvPathsException(f"Named-paths '{name}' has no csvpaths")
 765        if len(np) == 1:
 766            return np[0]
 767        if specific is None:
 768            self.logger.warning(
 769                "Parse_named_path %s has %s csvpaths. Using just the first one.",
 770                name,
 771                len(np),
 772            )
 773            return np[0]
 774        for p in np:
 775            # this ends up being redundant to the caller. we do it 1x so it's not
 776            # a big lift and is consistent.
 777            c = CsvPath(csvpaths=self.csvpaths)
 778            MetadataParser(c).extract_metadata(instance=c, csvpath=p)
 779            if c.identity == specific:
 780                return p
 781        self.logger.error(
 782            "Cannot find csvpath identified as %s in named-paths %s", specific, name
 783        )
 784        raise ParsingException(f"Cannot find path '{specific}' in named-paths '{name}'")
 785
 786    def parse_named_path(self, name, *, disposably=False, specific=None):
 787        """@private
 788        disposably is True when a Matcher is needed for some purpose other than
 789        the run we were created to do. could be that a match component wanted a
 790        parsed csvpath for its own purposes. import() uses this method.
 791        when True, we create and return the Matcher, but then forget it ever existed.
 792        also note: the path must have a name or full filename. $[*] is not enough.
 793        """
 794        if not self.csvpaths:
 795            raise CsvPathsException("No CsvPaths object available")
 796
 797        path = self._pick_named_path(name, specific=specific)
 798        c = CsvPath(csvpaths=self.csvpaths)
 799        path = MetadataParser(c).extract_metadata(instance=c, csvpath=path)
 800        #
 801        # exp. oddly this seems to be superfluous
 802        # if disposably is False:
 803        #    path = c._update_file_path(path)
 804        #
 805        dis = c.parse(path, disposably=disposably)
 806        if disposably is True:
 807            return dis
 808        return None
 809
 810    def _update_file_path(self, data: str):
 811        """@private
 812        this method replaces a name (i.e. name in: $name[*[][yes()]) with
 813        a file system path, if that name is registered with csvpaths's file
 814        manager. if there is no csvpaths no replace happens. if there is a
 815        csvpaths but the file manager doesn't know the name, no replace
 816        happens.
 817        """
 818        if data is None:
 819            raise InputException("The csvpath string cannot be None")
 820        if self.csvpaths is None:
 821            return data
 822        name = self._get_name(data)
 823        path = self.csvpaths.file_manager.get_named_file(name)
 824        if path is None:
 825            return data
 826        if path == name:
 827            return data
 828        return data.replace(name, path)
 829
 830    def _get_name(self, data: str):
 831        if self.csvpaths is None:
 832            return data
 833        data = data.strip()
 834        if data[0] == "$":
 835            name = data[1 : data.find("[")]
 836            return name
 837        raise FormatException(f"Must start with '$', not {data[0]}")
 838
 839    def _find_scan_and_match_parts(self, data):
 840        if data is None or not isinstance(data, str):
 841            raise InputException("Not a csvpath string")
 842        scan = ""
 843        matches = ""
 844        data = data.strip()
 845        i = data.find("]")
 846        if i < 0:
 847            raise InputException(f"Cannot find the scan part of this csvpath: {data}")
 848        if i == len(data) - 1:
 849            raise InputException(
 850                f"The scan part of this csvpath cannot be last: {data}"
 851            )
 852
 853        scan = data[0 : i + 1]
 854        scan = scan.strip()
 855
 856        ndata = data[i + 1 :]
 857        ndata = ndata.strip()
 858
 859        if ndata == "":
 860            raise InputException(f"There must be a match part of this csvpath: {data}")
 861        if ndata[0] != "[":
 862            raise InputException(f"Cannot find the match part of this csvpath: {data}")
 863        if ndata[len(ndata) - 1] != "]":
 864            raise InputException(f"The match part of this csvpath is incorrect: {data}")
 865        matches = ndata
 866        #
 867        # if we're given directory(s) to save to, save the parts
 868        #
 869        self._save_parts_if(scan, matches)
 870        return scan, matches
 871
 872    def _save_parts_if(self, scan, match):
 873        if self._save_scan_dir and self._run_name:
 874            with open(
 875                os.path.join(self._save_scan_dir, f"{self._run_name}.txt"),
 876                "w",
 877                encoding="utf-8",
 878            ) as f:
 879                f.write(scan)
 880        if self._save_match_dir and self._run_name:
 881            with open(
 882                os.path.join(self._save_match_dir, f"{self._run_name}.txt"),
 883                "w",
 884                encoding="utf-8",
 885            ) as f:
 886                f.write(match)
 887
 888    def __str__(self):
 889        return f"""
 890            path: {self.scanner.path if self.scanner else None}
 891            identity: {self.identity}
 892            parsers: [scanner=Ply, matcher=Lark, print=Lark]
 893            from_line: {self.scanner.from_line if self.scanner else None}
 894            to_line: {self.scanner.to_line if self.scanner else None}
 895            all_lines: {self.scanner.all_lines if self.scanner else None}
 896            these: {self.scanner.these if self.scanner else None}
 897            matcher: {self.matcher}
 898            variables: {len(self.variables)}
 899            metadata: {len(self.metadata)}
 900        """
 901
 902    @property
 903    def is_valid(self) -> bool:  # pragma: no cover
 904        """Csvpaths can flag a CSV file as invalid using the fail() function"""
 905        return self._is_valid
 906
 907    @is_valid.setter
 908    def is_valid(self, tf: bool) -> None:
 909        self._is_valid = tf
 910
 911    @property
 912    def completed(self) -> bool:
 913        if not self.scanner or not self.line_monitor:
 914            return False
 915        if self.scanner.is_last(self.line_monitor.physical_line_number):
 916            return True
 917        return False
 918
 919    @property
 920    def from_line(self):  # pragma: no cover pylint: disable=C0116
 921        """@private"""
 922        if self.scanner is None:
 923            raise ParsingException("No scanner available. Have you parsed a csvpath?")
 924        return self.scanner.from_line
 925
 926    @property
 927    def to_line(self):  # pragma: no cover pylint: disable=C0116
 928        """@private"""
 929        if self.scanner is None:
 930            raise ParsingException("No scanner available. Have you parsed a csvpath?")
 931        return self.scanner.to_line
 932
 933    @property
 934    def all_lines(self):  # pragma: no cover pylint: disable=C0116
 935        """@private"""
 936        if self.scanner is None:
 937            raise ParsingException("No scanner available. Have you parsed a csvpath?")
 938        return self.scanner.all_lines
 939
 940    @property
 941    def path(self):  # pragma: no cover pylint: disable=C0116
 942        """@private"""
 943        if self.scanner is None:
 944            raise ParsingException("No scanner available. Have you parsed a csvpath?")
 945        return self.scanner.path
 946
 947    @property
 948    def these(self):  # pragma: no cover pylint: disable=C0116
 949        """@private"""
 950        if self.scanner is None:
 951            raise ParsingException("No scanner available. Have you parsed a csvpath?")
 952        return self.scanner.these
 953
 954    @property
 955    def limit_collection_to(self) -> List[int]:
 956        """@private
 957        returns the list of headers to collect when a line matches. by default
 958        this list is empty and all headers are collected.
 959        """
 960        return self._limit_collection_to
 961
 962    @limit_collection_to.setter
 963    def limit_collection_to(self, indexes: List[int]) -> None:
 964        """@private"""
 965        self._limit_collection_to = indexes
 966        self.logger.warning("Setting a limit on headers collected: %s", indexes)
 967
 968    def stop(self) -> None:
 969        """@private"""
 970        self.stopped = True
 971
 972    #
 973    #
 974    # collect(), fast_forward(), and next() are the central methods of CsvPath.
 975    #
 976    # remember that since next() is a public method and collect() and fast_forward()
 977    # rely on it, we have to cut off exceptions at next(). users will not see
 978    # fast_forward and collect in the stack trace. probably that's not a big deal.
 979    # if it seems confusing we can work around the problem.
 980    #
 981    def collect(
 982        self, csvpath: str = None, *, nexts: int = -1, lines=None
 983    ) -> List[List[Any]] | LineSpooler:
 984        """Runs the csvpath forward and returns the matching lines seen as
 985        a list of lists. this method does not holds lines locally, not as
 986        accessible attributes. lines are not kept after the run completes
 987        and the collected lines are returned.
 988
 989        the optional lines argument may be an instance of any class that has an append(obj)
 990        method. if lines is None, a list is returned.
 991        """
 992        if self.scanner is None and csvpath is not None:
 993            self.parse(csvpath)
 994        if nexts < -1:
 995            raise ProcessingException(
 996                "Input must be >= -1. -1 means collect to the end of the file."
 997            )
 998        self.collecting = True
 999        #
1000        # we're going to use the passed in lines object. if it exists
1001        # it is a LineSpooler (we presume). we'll associate it with the
1002        # csvpath temporarily so anyone interested can get it. if it
1003        # doesn't exist we create a list for the lines var. we create a
1004        # list LinesSpooler to handle any inquiries during the run we
1005        # do all our line collecting busness with the local lines var
1006        # which has an append method regardless of if it is list or
1007        # LineSpooler. when we're done we break the link with the csvpath
1008        # and return the local variable. if it is a list it is going to
1009        # a csvpath user. if it is a LineSpooler it is going to a
1010        # CsvPaths instance.
1011        #
1012        self.lines = lines
1013        if lines is None:
1014            lines = []
1015            self.lines = ListLineSpooler(lines=lines)
1016        for _ in self.next():
1017            _ = _[:]
1018            self.lines.append(_)
1019            if nexts == -1:
1020                continue
1021            if nexts > 1:
1022                nexts -= 1
1023            else:
1024                break
1025        # we don't want to hold on to data more than needed. but
1026        # we do want to return data if we're not spooling. the
1027        # way we do that is to keep the local var available with the
1028        # list and/or the spooler. the caller needs to be aware of
1029        # both possibilities, but both offer __len__ and append.
1030        #
1031        # we keep the self.lines if it is not a list because that
1032        # makes it available to the runtime data collector so we can
1033        # see the line count in the metadata, saving opening a
1034        # potentially large data.csv to find out how many lines.
1035        if isinstance(self.lines, list):
1036            self.lines = None
1037        return lines
1038
1039    def fast_forward(self, csvpath=None):
1040        """Scans to the end of the CSV file. All scanned rows will be
1041        considered for match and variables and side effects will happen,
1042        but no rows will be returned or stored. -1 means to the end of
1043        the file. If you do not pass the csvpath string here you must first
1044        use the parse method."""
1045        if self.scanner is None and csvpath is not None:
1046            self.parse(csvpath)
1047        for _ in self.next():
1048            pass
1049        return self
1050
1051    #
1052    # dont_raise is for the use of fast_forward and collect who
1053    # may want to handle raising errors themselves.
1054    #
1055    def next(self, csvpath=None):
1056        """A generator function that steps through the CSV file returning
1057        matching rows."""
1058        try:
1059            if self.scanner is None and csvpath is not None:
1060                self.parse(csvpath)
1061            start = time.time()
1062            if self.will_run is True:
1063                for line in self._next_line():
1064                    b = self._consider_line(line)
1065                    if b:
1066                        line = self.limit_collection(line)
1067                        if line is None:
1068                            msg = "Line cannot be None"
1069                            self.logger.error(msg)
1070                            raise MatchException(msg)
1071                        if len(line) == 0:
1072                            msg = "Line cannot be len() == 0"
1073                            self.logger.error(msg)
1074                            raise MatchException(msg)
1075                        yield line
1076                    elif self.collecting and self.unmatched_available:
1077                        if self.unmatched is None:
1078                            self.unmatched = []
1079                        line = self.limit_collection(line)
1080                        # we aren't None and 0 checking as above. needed?
1081                        self.unmatched.append(line)
1082                    if self.stopped:
1083                        self.logger.info(
1084                            "CsvPath has been stopped at line %s",
1085                            self.line_monitor.physical_line_number,
1086                        )
1087                        break
1088            else:
1089                self.logger.warning(
1090                    "Csvpath identified as {self.identity} is disabled by run-mode:no-run"
1091                )
1092            self.finalize()
1093            end = time.time()
1094            self.total_iteration_time = end - start
1095            self.logger.info("Run against %s is complete.", self.scanner.filename)
1096            self.logger.info(
1097                "Iteration time was %s", round(self.total_iteration_time, 2)
1098            )
1099            self.logger.info(
1100                "%s per line",
1101                round(
1102                    self.total_iteration_time
1103                    / self.line_monitor.physical_end_line_count,
1104                    2,
1105                ),
1106            )
1107        except Exception as e:
1108            if not self.ecoms.do_i_quiet():
1109                self.logger.error(e, exc_info=True)
1110            if self.ecoms.do_i_raise():
1111                raise
1112
1113    def _next_line(self) -> List[Any]:
1114        """@private"""
1115        self.logger.info("beginning to scan file: %s", self.scanner.filename)
1116        #
1117        # this exception will blow up a standalone CsvPath but should be
1118        # caught and handled if there is a CsvPaths.
1119        #
1120        # but when would it happen? shouldn't we just let Python's exception
1121        # handle it should it really occur?
1122        #
1123        if self.scanner.filename is None:
1124            raise FileException("There is no filename")
1125        #
1126        # DataFileReader is abstract. instantiating it results in a concrete subclass.
1127        # pylint doesn't like that just because it doesn't see what we're doing.
1128        # otoh, is this a bad way to do it? not sure but it works fine.
1129        #
1130        reader = DataFileReader(  # pylint: disable=E0110
1131            self.scanner.filename, delimiter=self.delimiter, quotechar=self.quotechar
1132        )
1133        for line in reader.next():
1134            self.track_line(line=line)
1135            yield line
1136        self.finalize()
1137
1138    def finalize(self) -> None:
1139        """@private
1140        clears caches, etc. this is an internal method, but not _ because
1141        it is part of the lifecycle and we might find a reason to call it
1142        from higher up.
1143        """
1144        # this method can run multiple times w/np, but that
1145        # shouldn't happen anyway.
1146        self._freeze_path = True
1147        if self.matcher:
1148            self.matcher.clear_caches()
1149
1150    def track_line(self, line) -> None:
1151        """@private
1152        csvpaths needs to handle some of the iteration logic, and we don't want
1153        to lose track of line number monitoring or repeat the code up there,
1154        so we need this method to give csvpaths a way to tap in.
1155        """
1156        last_line = None
1157        if self.matcher:
1158            last_line = self.matcher.line
1159        self.line_monitor.next_line(last_line=last_line, data=line)
1160        if self.line_monitor.physical_line_number == 0:
1161            self._run_started_at = datetime.now(timezone.utc)
1162
1163    def _consider_line(self, line):  # pylint: disable=R0912, R0911
1164        """@private"""
1165        # re: R0912: this method has already been refactored but maybe
1166        # there is more we can do?
1167        #
1168        # we always look at the last line so that last() has a
1169        # chance to run
1170        #
1171        # if we're empty, but last, we need to make sure the
1172        # matcher runs a final time so that any last() can run.
1173        #
1174        if self.line_monitor.is_last_line_and_blank(line):
1175            # if self.line_monitor.is_last_line_and_empty(line):
1176            self.logger.info("last line is empty. freezing, matching, returning false")
1177            self._freeze_path = True
1178            self.matches(line)
1179            return False
1180        if self.skip_blank_lines and len(line) == 0:
1181            self.logger.info(
1182                "Skipping line %s because blank", self.line_monitor.physical_line_number
1183            )
1184            return False
1185        if self.scanner.includes(self.line_monitor.physical_line_number):
1186            self.logger.debug("Scanner includes line")
1187            self.scan_count = self.scan_count + 1
1188            matches = None
1189            self._current_match_count = self.match_count
1190            if self.advance_count > 0:
1191                self.advance_count -= 1
1192                matches = False
1193                self.logger.debug(
1194                    "Advancing one line with {self.advance_count} more skips to go"
1195                )
1196            else:
1197                self.logger.debug("Starting matching")
1198                startmatch = time.perf_counter_ns()
1199                matches = self.matches(line)
1200                endmatch = time.perf_counter_ns()
1201                t = (endmatch - startmatch) / 1000000
1202                self.last_row_time = t
1203                self.rows_time += t
1204                self.logger.debug(
1205                    "CsvPath.matches:703: %s: matches: %s", self.identity, matches
1206                )
1207            #
1208            # if we are done scanning we can stop
1209            #
1210            if self.scanner.is_last(self.line_monitor.physical_line_number):
1211                self.stop()
1212            if matches is True:
1213                #
1214                # _current_match_count is a placeholder that
1215                # allows anyone to call a match early and update
1216                # the count. this is important when there is
1217                # an onmatch component that needs to use the
1218                # match_count. e.g. an onmatch print statement.
1219                # we would want the onmatch to propagate asap. we
1220                # can accept that there could be a variable set to
1221                # match count prior to the onmatch upping the
1222                # count. that wouldn't be great for explainability,
1223                # but order is important -- match components
1224                # impact each other left to right, top to bottom.
1225                #
1226                self.raise_match_count_if()
1227                if self.collect_when_not_matched:
1228                    return False
1229                return True
1230            if self.collect_when_not_matched:
1231                return True
1232            return False
1233        return False
1234
1235    def raise_match_count_if(self):
1236        """@private
1237        if the match count has already been raised earlier in the matching
1238        process than the caller we don't raise it; otherwise, we raise."""
1239        if self._current_match_count == self.match_count:
1240            self.match_count += 1
1241        else:
1242            self.logger.debug("Match count was already raised, so not doing it again")
1243
1244    def limit_collection(self, line: List[Any]) -> List[Any]:
1245        """@private
1246        this method creates a line based on the given line that holds only the headers
1247        that the csvpath says to collect. headers for collection are indicated using
1248        the collect() function.
1249        """
1250        if len(self.limit_collection_to) == 0:
1251            return line
1252        ls = []
1253        for k in self.limit_collection_to:
1254            if k is None or k >= len(line):
1255                raise InputException(
1256                    f"[{self.identity}] Line {self.line_monitor.physical_line_number}: unknown header name: {k}"
1257                )
1258            ls.append(line[k])
1259        return ls
1260
1261    def advance(self, ff: int = -1) -> None:
1262        """@private
1263        Advances the iteration by ff rows. The rows will be seen but not matched."""
1264        if ff is None:
1265            raise InputException("Input to advance must not be None")
1266        if self.line_monitor.physical_end_line_number is None:
1267            raise ProcessingException(
1268                "The last line number must be known (physical_end_line_number)"
1269            )
1270        if ff == -1:
1271            a = self.advance_count
1272            a = (
1273                self.line_monitor.physical_end_line_number
1274                - self.line_monitor.physical_line_number
1275                - a
1276            )
1277            self.advance_count = a
1278        else:
1279            self.advance_count += ff
1280        self.advance_count = min(
1281            self.advance_count, self.line_monitor.physical_end_line_number
1282        )
1283
1284    def get_total_lines(self) -> int:  # pylint: disable=C0116
1285        """@private"""
1286        if (
1287            self.line_monitor.physical_end_line_number is None
1288            or self.line_monitor.physical_end_line_number == 0
1289        ):
1290            self.get_total_lines_and_headers()
1291        return self.line_monitor.physical_end_line_number
1292
1293    def get_total_lines_and_headers(self) -> None:  # pylint: disable=C0116
1294        """@private"""
1295        if not self.scanner or not self.scanner.filename:
1296            self.logger.error(
1297                "Csvpath identified as %s has no filename. Since we could be error handling an exception is not raised.",
1298                self.identity,
1299            )
1300            # exp
1301            return
1302            # end exp
1303        #
1304        # there are times, e.g. when using Lambda, when it may be better to
1305        # not use a cache. in the case of Lambda the reason is to avoid working
1306        # around the read-only filesystem.
1307        #
1308        use_cache = self.csvpaths is not None
1309        if use_cache:
1310            uc = self.csvpaths.config.get(section="cache", name="use_cache")
1311            use_cache = uc is None or uc.strip().lower() != "no"
1312        if self.csvpaths and use_cache is True:
1313            self.line_monitor = self.csvpaths.file_manager.cacher.get_new_line_monitor(
1314                self.scanner.filename
1315            )
1316            self.headers = self.csvpaths.file_manager.cacher.get_original_headers(
1317                self.scanner.filename
1318            )
1319        else:
1320            lc = LineCounter(self)
1321            lm, headers = lc.get_lines_and_headers(self.scanner.filename)
1322            self.line_monitor = lm
1323            self.headers = headers
1324
1325    @property
1326    def current_scan_count(self) -> int:  # pylint: disable=C0116
1327        """@private"""
1328        return self.scan_count
1329
1330    @property
1331    def current_match_count(self) -> int:  # pylint: disable=C0116
1332        """@private"""
1333        return self.match_count
1334
1335    def matches(self, line) -> bool:  # pylint: disable=C0116
1336        """@private"""
1337        if not self.match:
1338            return True
1339        #
1340        # when we first consider a line we don't have a matcher. we build
1341        # it on the fly. later, we just reset the matcher for the new lines.
1342        #
1343        # when we originally call parse we're just parsing for the scanner:
1344        #
1345        #   path = CsvPath()
1346        #   path.parse ("$file[*][yes()]")
1347        #   path.fast_forward()
1348        #
1349        # "find_file" would be a more intuitive method name. we don't create
1350        # the path's matcher until the 3rd line. by then we're on the 3rd parser
1351        # and 4 parse.
1352        #
1353        if self.matcher is None:
1354            h = hashlib.sha256(self.match.encode("utf-8")).hexdigest()
1355            self.logger.info("Loading matcher with data. match part hash: %s", h)
1356            self.matcher = Matcher(
1357                csvpath=self, data=self.match, line=line, headers=self.headers, myid=h
1358            )
1359            self.matcher.AND = self.AND
1360            #
1361            # we need to register all the Expressions as error listeners. not
1362            # sure it matters if we do it here or allow the Matcher to do it.
1363            # since the Matcher is responsible for its Expressions, has a handle
1364            # this CsvPath, and through it has the error_manager let's let it
1365            # register the expressions.
1366            #
1367        else:
1368            self.logger.debug("Resetting and reloading matcher")
1369            self.matcher.reset()
1370            self.matcher.line = line
1371        matched = self.matcher.matches()
1372        return matched
1373
1374    def set_variable(self, name: str, *, value: Any, tracking: Any = None) -> None:
1375        """@private
1376        sets a variable and the tracking variable as a key within
1377        it, if a tracking value is provided."""
1378        if self._freeze_path:
1379            self.logger.warning(
1380                "Run is ending, variables are frozen. Cannot set %s to %s.", name, value
1381            )
1382            return
1383        if not name:
1384            raise VariableException(
1385                f"Name cannot be None: name: {name}, tracking: {tracking}, value: {value}"
1386            )
1387        if name.strip() == "":
1388            raise VariableException(
1389                f"""Name cannot be the empty string:
1390                    name: {name}, tracking: {tracking}, value: {value}"""
1391            )
1392        if tracking is not None and f"{tracking}".strip() == "":
1393            raise VariableException(
1394                f"""Tracking value cannot be empty.
1395                    name: {name}, tracking: {tracking}, value: {value}"""
1396            )
1397        if tracking is not None:
1398            if name not in self.variables:
1399                self.variables[name] = {}
1400            instances = self.variables[name]
1401            instances[tracking] = value
1402        else:
1403            self.variables[name] = value
1404
1405    def get_variable(  # pylint: disable=R0912
1406        self, name: str, *, tracking: Any = None, set_if_none: Any = None
1407    ) -> Any:
1408        """@private
1409        gets a variable by name. uses the tracking value as a key to get
1410        the value if the variable is a dictionary."""
1411        #
1412        # re: R0912: totally true. this is a scary method. plan to refactor.
1413        #
1414        if not name:
1415            raise VariableException("Name cannot be None")
1416        if self._freeze_path:
1417            #
1418            # run is ending, no more changes
1419            #
1420            set_if_none = None
1421        thevalue = None
1422        if tracking is not None:
1423            thedict = None
1424            thevalue = None
1425            if name in self.variables:
1426                thedict = self.variables[name]
1427                if not thedict:
1428                    thedict = {}
1429                    self.variables[name] = thedict
1430                    thedict[tracking] = set_if_none
1431            else:
1432                thedict = {}
1433                thedict[tracking] = set_if_none
1434                self.variables[name] = thedict
1435            if isinstance(thedict, dict):
1436                thevalue = thedict.get(tracking)
1437            if not thevalue and set_if_none is not None:
1438                thedict[tracking] = set_if_none
1439                thevalue = set_if_none
1440        else:
1441            if name not in self.variables:
1442                if set_if_none is not None:
1443                    self.variables[name] = set_if_none
1444                    thevalue = set_if_none
1445            else:
1446                thevalue = self.variables[name]
1447        if self._freeze_path:
1448            if isinstance(thevalue, list):
1449                #
1450                # run is ending, no more changes
1451                #
1452                thevalue = tuple(thevalue[:])
1453                self.logger.debug(
1454                    "Returning %s for frozen variable %s.%s", thevalue, name, tracking
1455                )
1456        return thevalue
1457
1458    def line_numbers(self) -> Iterator[int | str]:
1459        """@private
1460        returns all the line numbers the scanner will scan during
1461        the run of a csvpath"""
1462        these = self.scanner.these
1463        from_line = self.scanner.from_line
1464        to_line = self.scanner.to_line
1465        all_lines = self.scanner.all_lines
1466        return self._line_numbers(
1467            these=these, from_line=from_line, to_line=to_line, all_lines=all_lines
1468        )
1469
1470    def _line_numbers(
1471        self,
1472        *,
1473        these: List[int] = None,
1474        from_line: int = None,
1475        to_line: int = None,
1476        all_lines: bool = None,
1477    ) -> Iterator[int | str]:
1478        """@private"""
1479        if these is None:
1480            these = []
1481        if len(these) > 0:
1482            yield from these
1483        else:
1484            if from_line is not None and to_line is not None and from_line > to_line:
1485                yield from range(to_line, from_line + 1)
1486            elif from_line is not None and to_line is not None:
1487                yield from range(from_line, to_line + 1)
1488            elif from_line is not None:
1489                if all_lines:
1490                    yield f"{from_line}..."
1491                else:
1492                    yield from_line
1493            elif to_line is not None:
1494                yield f"0..{to_line}"
1495
1496    def collect_line_numbers(self) -> List[int | str]:  # pylint: disable=C0116
1497        """@private"""
1498        if self.scanner is None:
1499            raise ParsingException("No scanner available. Have you parsed a csvpath?")
1500        these = self.scanner.these
1501        from_line = self.scanner.from_line
1502        to_line = self.scanner.to_line
1503        all_lines = self.scanner.all_lines
1504        return self._collect_line_numbers(
1505            these=these, from_line=from_line, to_line=to_line, all_lines=all_lines
1506        )
1507
1508    def _collect_line_numbers(
1509        self,
1510        *,
1511        these: List[int] = None,
1512        from_line: int = None,
1513        to_line: int = None,
1514        all_lines: bool = None,
1515    ) -> List[int | str]:
1516        """@private"""
1517        collect = []
1518        if these is None:
1519            these = []
1520        for i in self._line_numbers(
1521            these=these, from_line=from_line, to_line=to_line, all_lines=all_lines
1522        ):
1523            collect.append(i)
1524        return collect
1525
1526    def header_index(self, name: str) -> int:  # pylint: disable=C0116
1527        """@private"""
1528        if not self.headers:
1529            return None
1530        for i, n in enumerate(self.headers):
1531            if n == name:
1532                return i
1533        return None