Warning
This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy.datatypes.dataproviders.line
"""
Dataproviders that iterate over lines from their sources.
"""
import collections
import logging
import re
from . import base
log = logging.getLogger(__name__)
_TODO = """
line offsets (skip to some place in a file) needs to work more efficiently than simply iterating till we're there
capture tell() when provider is done
def stop( self ): self.endpoint = source.tell(); raise StopIteration()
a lot of the hierarchy here could be flattened since we're implementing pipes
"""
[docs]class FilteredLineDataProvider(base.LimitedOffsetDataProvider):
"""
Data provider that yields lines of data from its source allowing
optional control over which line to start on and how many lines
to return.
"""
DEFAULT_COMMENT_CHAR = "#"
settings = {
"strip_lines": "bool",
"strip_newlines": "bool",
"provide_blank": "bool",
"comment_char": "str",
}
[docs] def __init__(
self,
source,
strip_lines=True,
strip_newlines=False,
provide_blank=False,
comment_char=DEFAULT_COMMENT_CHAR,
**kwargs,
):
"""
:param strip_lines: remove whitespace from the beginning an ending
of each line (or not).
Optional: defaults to True
:type strip_lines: bool
:param strip_newlines: remove newlines only
(only functions when ``strip_lines`` is false)
Optional: defaults to False
:type strip_lines: bool
:param provide_blank: are empty lines considered valid and provided?
Optional: defaults to False
:type provide_blank: bool
:param comment_char: character(s) that indicate a line isn't data (a comment)
and should not be provided.
Optional: defaults to '#'
:type comment_char: str
"""
super().__init__(source, **kwargs)
self.strip_lines = strip_lines
self.strip_newlines = strip_newlines
self.provide_blank = provide_blank
self.comment_char = comment_char
[docs] def filter(self, line):
"""
Determines whether to provide line or not.
:param line: the incoming line from the source
:type line: str
:returns: a line or `None`
"""
if line is not None:
# ??: shouldn't it strip newlines regardless, if not why not use on of the base.dprovs
if self.strip_lines:
line = line.strip()
elif self.strip_newlines:
line = line.strip("\n")
if not self.provide_blank and line == "":
return None
elif self.comment_char and line.startswith(self.comment_char):
return None
return super().filter(line)
[docs]class RegexLineDataProvider(FilteredLineDataProvider):
"""
Data provider that yields only those lines of data from its source
that do (or do not when `invert` is True) match one or more of the given list
of regexs.
.. note:: the regex matches are effectively OR'd (if **any** regex matches
the line it is considered valid and will be provided).
"""
settings = {
"regex_list": "list:escaped",
"invert": "bool",
}
[docs] def __init__(self, source, regex_list=None, invert=False, **kwargs):
"""
:param regex_list: list of strings or regular expression strings that will
be `match`ed to each line
Optional: defaults to `None` (no matching)
:type regex_list: list (of str)
:param invert: if `True` will provide only lines that **do not match**.
Optional: defaults to False
:type invert: bool
"""
super().__init__(source, **kwargs)
self.regex_list = regex_list if isinstance(regex_list, list) else []
self.compiled_regex_list = [re.compile(regex) for regex in self.regex_list]
self.invert = invert
# NOTE: no support for flags
[docs] def filter(self, line):
# NOTE: filter_fn will occur BEFORE any matching
line = super().filter(line)
if line is not None and self.compiled_regex_list:
line = self.filter_by_regex(line)
return line
[docs] def filter_by_regex(self, line):
matches = any(regex.match(line) for regex in self.compiled_regex_list)
if self.invert:
return line if not matches else None
return line if matches else None
# ============================================================================= MICELLAINEOUS OR UNIMPLEMENTED
# ----------------------------------------------------------------------------- block data providers
[docs]class BlockDataProvider(base.LimitedOffsetDataProvider):
"""
Class that uses formats where multiple lines combine to describe a single
datum. The data output will be a list of either map/dicts or sub-arrays.
Uses FilteredLineDataProvider as its source (kwargs **not** passed).
e.g. Fasta, GenBank, MAF, hg log
Note: mem intensive (gathers list of lines before output)
"""
[docs] def __init__(self, source, new_block_delim_fn=None, block_filter_fn=None, **kwargs):
"""
:param new_block_delim_fn: T/F function to determine whether a given line
is the start of a new block.
:type new_block_delim_fn: function
:param block_filter_fn: function that determines if a block is valid and
will be provided.
Optional: defaults to `None` (no filtering)
:type block_filter_fn: function
"""
# composition - not inheritance
# TODO: not a fan of this:
(filter_fn, limit, offset) = (kwargs.pop("filter_fn", None), kwargs.pop("limit", None), kwargs.pop("offset", 0))
line_provider = FilteredLineDataProvider(source, **kwargs)
super().__init__(line_provider, filter_fn=filter_fn, limit=limit, offset=offset)
self.new_block_delim_fn = new_block_delim_fn
self.block_filter_fn = block_filter_fn
self.init_new_block()
[docs] def init_new_block(self):
"""
Set up internal data for next block.
"""
# called in __init__ and after yielding the prev. block
self.block_lines = collections.deque([])
def __iter__(self):
"""
Overridden to provide last block.
"""
parent_gen = super().__iter__()
yield from parent_gen
if (last_block := self.handle_last_block()) is not None:
self.num_data_returned += 1
yield last_block
[docs] def filter(self, line):
"""
Line filter here being used to aggregate/assemble lines into a block
and determine whether the line indicates a new block.
:param line: the incoming line from the source
:type line: str
:returns: a block or `None`
"""
line = super().filter(line)
# TODO: HACK
self.num_data_read -= 1
if line is None:
return None
block_to_return = None
if self.is_new_block(line):
# if we're already in a block, return the prev. block and add the line to a new block
if self.block_lines:
block_to_return = self.assemble_current_block()
block_to_return = self.filter_block(block_to_return)
self.num_data_read += 1
self.init_new_block()
self.add_line_to_block(line)
return block_to_return
[docs] def is_new_block(self, line):
"""
Returns True if the given line indicates the start of a new block
(and the current block should be provided) or False if not.
"""
if self.new_block_delim_fn:
return self.new_block_delim_fn(line)
return True
# NOTE:
# some formats have one block attr per line
# some formats rely on having access to multiple lines to make sensible data
# So, building the block from the lines can happen in either:
# add_line_to_block AND/OR assemble_current_block
[docs] def add_line_to_block(self, line):
"""
Integrate the given line into the current block.
Called per line.
"""
# here either:
# consume the line (using it to add attrs to self.block)
# save the line (appending to self.block_lines) for use in assemble_current_block
self.block_lines.append(line)
[docs] def assemble_current_block(self):
"""
Build the current data into a block.
Called per block (just before providing).
"""
# empty block_lines and assemble block
return list(self.block_lines.popleft() for i in range(len(self.block_lines)))
[docs] def filter_block(self, block):
"""
Is the current block a valid/desired datum.
Called per block (just before providing).
"""
if self.block_filter_fn:
return self.block_filter_fn(block)
return block
[docs] def handle_last_block(self):
"""
Handle any blocks remaining after the main loop.
"""
if self.limit is not None and self.num_data_returned >= self.limit:
return None
last_block = self.assemble_current_block()
self.num_data_read += 1
last_block = self.filter_block(last_block)
if last_block is not None:
self.num_valid_data_read += 1
return last_block