ldms_blob_file_to_http_post.Readline0 Class Reference

Inheritance diagram for ldms_blob_file_to_http_post.Readline0:

Collaboration diagram for ldms_blob_file_to_http_post.Readline0:

Public Member Functions
None	__init__ (self, typing.Union[typing.BinaryIO, int] file_, bytes separator, int blocksize)
typing.Tuple[typing.List[bytes], bool, bytes]	handle_field_pairs (cls, typing.List[typing.Tuple[bytes, bytes]] field_pairs)
None	get_fields (self)
typing.Iterator[bytes]	sequence (self)

Public Attributes
	file_
	blocksize
	have_fraction
	fraction
	separator
	yieldno
	bang
	metapattern
	buffer_
	pattern
	at_eof
	fields

Detailed Description

Yield a series of blocks, separated by separator.

Constructor & Destructor Documentation

__init__()

None ldms_blob_file_to_http_post.Readline0.__init__	(		self,
		typing.Union[typing.BinaryIO, int]	file_,
		bytes	separator,
		int	blocksize
	)

Initialize.

    def __init__(self, file_: typing.Union[typing.BinaryIO, int], separator: bytes, blocksize: int) -> None:
        """Initialize."""
        self.file_ = file_
        self.blocksize = blocksize
 
        self.have_fraction = False
        self.fraction = b''
 
        self.separator = separator
 
        self.fields: typing.List[bytes] = []
 
        self.yieldno = 0
 
        self.bang = b'!'
        self.metapattern = b'([^!]*)!|([^!]+)$'
        self.buffer_ = b''
        self.separator = separator
 
        # bytes objects have a split method, but it doesn't work, at least not in Python 3.1.2.  But the re module
        # works with bytes, so we use that.
 
        self.pattern = re.sub(self.bang, self.separator, self.metapattern)
 
        self.at_eof = False
 

Member Function Documentation

get_fields()

None ldms_blob_file_to_http_post.Readline0.get_fields

(

self

)

Read a block, chop it up into fields - taking into account any leftover partial field.

    def get_fields(self) -> None:
        """Read a block, chop it up into fields - taking into account any leftover partial field."""
        if isinstance(self.file_, int):
            tail_block: bytes = os.read(self.file_, self.blocksize)
        else:
            # assume we have a file-like object
            tail_block = self.file_.read(self.blocksize)
 
        if tail_block:
            self.at_eof = False
        else:
            self.at_eof = True
 
        if self.have_fraction:
            block = self.fraction + tail_block
            self.fraction = b''
            self.have_fraction = False
        else:
            block = tail_block
 
        field_pairs = re.findall(self.pattern, block)
        regular_fields, self.have_fraction, self.fraction = self.handle_field_pairs(field_pairs)
 
        # we put the fields in reverse order so we can repeatedly pop efficiently
        regular_fields.reverse()
 
        self.fields = regular_fields
 

handle_field_pairs()

typing.Tuple[typing.List[bytes], bool, bytes] ldms_blob_file_to_http_post.Readline0.handle_field_pairs	(		cls,
		typing.List[typing.Tuple[bytes, bytes]]	field_pairs
	)

Pick apart the pairs from our regex split and return the correct values.

    ) -> typing.Tuple[typing.List[bytes], bool, bytes]:
        """Pick apart the pairs from our regex split and return the correct values."""
        regular_fields = []
        have_fraction = False
        fraction = b''
 
        for field_pair in field_pairs:
            if field_pair[0]:
                if field_pair[1]:
                    # They're both not zero length - that's an error
                    raise AssertionError('Both field_pair[0] and field_pair[1] are non-empty')
                else:
                    # The first is not zero length, the second is zero length
                    regular_fields.append(field_pair[0])
            else:
                if field_pair[1]:
                    # the first is zero length, the second is not zero length
                    if have_fraction:
                        raise AssertionError('Already have a fraction')
                    fraction = field_pair[1]
                    have_fraction = True
                else:
                    # they're both zero length - this is legal for !! - yield one or the other but not both
                    regular_fields.append(field_pair[0])
 
        return regular_fields, have_fraction, fraction
 

sequence()

typing.Iterator[bytes] ldms_blob_file_to_http_post.Readline0.sequence

(

self

)

Generate each field (line) in turn.

    def sequence(self) -> typing.Iterator[bytes]:
        """Generate each field (line) in turn."""
        while True:
            if not self.fields:
                self.get_fields()
            while self.fields:
                yield self.fields.pop()
            if self.at_eof:
                if self.have_fraction:
                    yield self.fraction
                break
 

Member Data Documentation

at_eof

ldms_blob_file_to_http_post.Readline0.at_eof

bang

ldms_blob_file_to_http_post.Readline0.bang

blocksize

ldms_blob_file_to_http_post.Readline0.blocksize

buffer_

ldms_blob_file_to_http_post.Readline0.buffer_

fields

ldms_blob_file_to_http_post.Readline0.fields

file_

ldms_blob_file_to_http_post.Readline0.file_

fraction

ldms_blob_file_to_http_post.Readline0.fraction

have_fraction

ldms_blob_file_to_http_post.Readline0.have_fraction

metapattern

ldms_blob_file_to_http_post.Readline0.metapattern

pattern

ldms_blob_file_to_http_post.Readline0.pattern

separator

ldms_blob_file_to_http_post.Readline0.separator

yieldno

ldms_blob_file_to_http_post.Readline0.yieldno

---

The documentation for this class was generated from the following file:

• src/adctk/ldms_scripts/ldms_blob_file_to_http_post.py