oasislmf.pytools.common.event_stream

Contain all common function and attribute to help read the event stream containing the losses

Attributes

PIPE_CAPACITY
CDF_STREAM_ID
GUL_STREAM_ID
FM_STREAM_ID
LOSS_STREAM_ID
SUMMARY_STREAM_ID
ITEM_STREAM
COVERAGE_STREAM
NUM_SPECIAL_SIDX
MEAN_IDX
STD_DEV_IDX
TIV_IDX
MAX_LOSS_IDX

Classes

EventReader

Abstract class to read event stream

Functions

stream_info_to_bytes(stream_source_type, stream_agg_type)	From Stream source type and aggregation type produce the stream header
bytes_to_stream_types(stream_header)	Read the stream header and return the information on stream type
read_stream_info(stream_obj)	From open stream object return the information that characterize the stream (stream_source_type, stream_agg_type, len_sample)
get_streams_in(files_in, stack)
get_and_check_header_in(streams_in)
init_streams_in(files_in, stack)	If files_in use stdin as stream in
mv_read(byte_mv, cursor, _dtype, itemsize)	Read a certain dtype from numpy byte view starting at cursor, return the value and the index of the end of the object
reservation_overflows(idx, reservation, capacity, name)	Check whether reserving reservation more rows in a fixed-size output buffer
mv_write(→ int)	Load an object into the numpy byte view at index cursor, return the index of the end of the object
mv_write_summary_header(→ int)	Wrapper for cached write a summary header to the numpy byte view at index cursor, return the index of the end of the object
mv_write_summary_header_cached(→ int)	Cached write a summary header to the numpy byte view at index cursor, return the index of the end of the object
mv_write_item_header(→ int)	Wrapper function for cached mv_write_item_header. writes an item header to the numpy byte view,
mv_write_sidx_loss(→ int)	Write sidx and loss to the numpy byte view at index cursor, return the index of the end of the object
mv_write_sidx_loss_cached(→ int)	Cached write sidx and loss to the numpy byte view at index cursor, return the index of the end of the object
write_mv_to_stream(stream, byte_mv, cursor)	Write numpy byte array view to stream

Module Contents

oasislmf.pytools.common.event_stream.PIPE_CAPACITY = 65536

oasislmf.pytools.common.event_stream.CDF_STREAM_ID = 0

oasislmf.pytools.common.event_stream.GUL_STREAM_ID = 1

oasislmf.pytools.common.event_stream.FM_STREAM_ID = 2

oasislmf.pytools.common.event_stream.LOSS_STREAM_ID = 2

oasislmf.pytools.common.event_stream.SUMMARY_STREAM_ID = 3

oasislmf.pytools.common.event_stream.ITEM_STREAM = 1

oasislmf.pytools.common.event_stream.COVERAGE_STREAM = 2

oasislmf.pytools.common.event_stream.NUM_SPECIAL_SIDX = 5

oasislmf.pytools.common.event_stream.MEAN_IDX = -1

oasislmf.pytools.common.event_stream.STD_DEV_IDX = -2

oasislmf.pytools.common.event_stream.TIV_IDX = -3

oasislmf.pytools.common.event_stream.MAX_LOSS_IDX = -5

oasislmf.pytools.common.event_stream.stream_info_to_bytes(stream_source_type, stream_agg_type)

From Stream source type and aggregation type produce the stream header

Parameters:

• stream_source_type (np.int32) – id of the tool that produced the stream

• stream_agg_type (np.int32) – id of the aggregation level of the stream

Returns:

return bytes

oasislmf.pytools.common.event_stream.bytes_to_stream_types(stream_header)

Read the stream header and return the information on stream type

Parameters:

stream_header – bytes

Returns:

(stream source type (np.int32), stream aggregation type (np.int32))

oasislmf.pytools.common.event_stream.read_stream_info(stream_obj)

From open stream object return the information that characterize the stream (stream_source_type, stream_agg_type, len_sample)

Parameters:

stream_obj – open stream

Returns:

(stream_source_type, stream_agg_type, len_sample) as np.int32 triplet

oasislmf.pytools.common.event_stream.get_streams_in(files_in, stack)

oasislmf.pytools.common.event_stream.get_and_check_header_in(streams_in)

oasislmf.pytools.common.event_stream.init_streams_in(files_in, stack)

If files_in use stdin as stream in otherwise open each path in files_in, read the header, check that they are the same, and return the streams and their info

Parameters:

• files_in – none or a list of path

• stack – contextlib stack to add the open stream to

Returns:

list of open streams and their info

oasislmf.pytools.common.event_stream.mv_read(byte_mv, cursor, _dtype, itemsize)

Read a certain dtype from numpy byte view starting at cursor, return the value and the index of the end of the object

Parameters:

• byte_mv – numpy byte view

• cursor – index of where the object start

• _dtype – data type of the object

• itemsize – size of the data type

Returns:

(object value, end of object index)

oasislmf.pytools.common.event_stream.reservation_overflows(idx, reservation, capacity, name)

Check whether reserving reservation more rows in a fixed-size output buffer (used before reading a summary’s raw data, so its worst-case output is guaranteed to fit - see elt/manager.py and plt/manager.py’s read_buffer) would overflow it.

Parameters:

• idx – current write position in the output buffer

• reservation – worst-case number of rows this summary could add

• capacity – output buffer’s total size

• name – output type name, used only in the raised error message (e.g. “SELT”)

Returns:

True if the buffer needs to be flushed before this summary can be read.

False if there’s room.

Return type:

bool

Raises:

ValueError – if idx == 0 (buffer already empty) and it still doesn’t fit - flushing an empty buffer can never make more room, so returning True here would have the caller loop forever instead of making progress.

oasislmf.pytools.common.event_stream.mv_write(byte_mv, cursor, _dtype, itemsize, value) → int

Load an object into the numpy byte view at index cursor, return the index of the end of the object

Parameters:

• byte_mv – numpy byte view

• cursor – index of where the object start

• _dtype – data type of the object

• itemsize – size of the data type

• value – value to write

Returns:

end of object index

oasislmf.pytools.common.event_stream.mv_write_summary_header(byte_mv, cursor, event_id, summary_id, exposure_value) → int

Wrapper for cached write a summary header to the numpy byte view at index cursor, return the index of the end of the object

Parameters:

• byte_mv – numpy byte view

• cursor – index of where the object start

• event_id – event id

• summary_id – summary id

• exposure_value – exposure value

Returns:

end of object index

oasislmf.pytools.common.event_stream.mv_write_summary_header_cached(byte_mv, cursor, event_id, summary_id, exposure_value, event_id_type, event_id_size, summary_id_dtype, summary_id_size, exposure_value_dtype, exposure_value_size) → int

Cached write a summary header to the numpy byte view at index cursor, return the index of the end of the object :param byte_mv: numpy byte view :param cursor: index of where the object start :param event_id: event id :param summary_id: summary id :param exposure_value: exposure value :param event_id_type: type info for event id :param event_id_size: size of event id in bytes :param summary_id_dtype: type info for summary id :param summary_id_size: size of summary id :param exposure_value_dtype: type info for exposure value :param exposure_value_size: size of exposure value

Returns:

end of object index

oasislmf.pytools.common.event_stream.mv_write_item_header(byte_mv, cursor, event_id, item_id) → int

Wrapper function for cached mv_write_item_header. writes an item header to the numpy byte view, return index of the end of the object

Parameters:

• byte_mv – numpy byte view

• cursor – index of where the object start

• event_id – event id

• item_id – item id

Returns:

end of object index

oasislmf.pytools.common.event_stream.mv_write_sidx_loss(byte_mv, cursor, sidx, loss) → int

Write sidx and loss to the numpy byte view at index cursor, return the index of the end of the object

Parameters:

• byte_mv – numpy byte view

• cursor – index of where the object start

• sidx – sample id

• loss – loss

Returns:

end of object index

oasislmf.pytools.common.event_stream.mv_write_sidx_loss_cached(byte_mv, cursor, sidx, loss, sidx_type, loss_type, sidx_size, loss_size) → int

Cached write sidx and loss to the numpy byte view at index cursor, return the index of the end of the object :param byte_mv: numpy byte view :param cursor: index of where the object start :param sidx: sample id :param loss: loss :param sidx_type: sample id type info :param loss_type: loss type info :param sidx_size: sample id size in bytes :param loss_size: loss size in bytes

Returns:

end of object index

class oasislmf.pytools.common.event_stream.EventReader

Abstract class to read event stream

This class provides a generic interface to read multiple event streams using:

• selector: handle back pressure — the program is paused and doesn’t use resources if nothing is in the stream buffer

• memoryview: read a chunk (PIPE_CAPACITY) of data at a time then work on it using a numpy byte view of this buffer

These methods need to be implemented:

• __init__(self, ...): the constructor with all data structures needed to read and store the event stream

• read_buffer(self, byte_mv, cursor, valid_buff, event_id, item_id): point to a local numba.jit function named read_buffer (a template is provided below); it should implement the specific logic of where and how to store the event information

These methods may be overwritten:

• item_exit(self): specific logic to do when an item is finished (only executed once the stream is finished but no 0,0 closure was present)

• event_read_log(self): what kpi to log when a full event is read

Usage snippet:

with ExitStack() as stack:
    streams_in, (stream_type, stream_agg_type, len_sample) = init_streams_in(files_in, stack)
    reader = CustomReader(<read relevant attributes>)
    for event_id in reader.read_streams(streams_in):
        <event logic>

static register_streams_in(selector_class, streams_in)

Data from input process is generally sent by event block, meaning once a stream receive data, the complete event is going to be sent in a short amount of time. Therefore, we can focus on each stream one by one using their specific selector ‘stream_selector’.

read_streams(streams_in)

Read multiple stream input, yield each event id and load relevant value according to the specific read_buffer implemented in subclass

Parameters:

streams_in – streams to read

Yields:

int – each event id read from the streams

read_event(stream_in, main_selector, stream_selector, mv, byte_mv, cursor, valid_buff, file_idx)

Read one event from stream_in close and remove the stream from main_selector when all is read

Parameters:

• stream_in – stream to read

• main_selector – selector that contain all the streams

• stream_selector – this stream selector

• mv – buffer memoryview

• byte_mv – numpy byte view of the buffer

• cursor – current cursor of the memory view

• valid_buff – valid data in memory view

• file_idx – file index

Returns:

event_id, cursor, valid_buff

abstractmethod read_buffer(byte_mv, cursor, valid_buff, event_id, item_id, **kwargs)

item_exit()

finalize_event()

Hook called by read_streams just before yielding each event_id. Subclasses use this to perform any per-event finishing work that read_buffer cannot do on its own — e.g. sorting accumulated state. Default is a no-op.

event_read_log(event_id)

oasislmf.pytools.common.event_stream.write_mv_to_stream(stream, byte_mv, cursor)

Write numpy byte array view to stream - use select to handle forward pressure - use a while loop in case the stream is non-blocking (meaning the ammount of byte written is not guarantied to be cursor len)

Parameters:

• stream – stream to write to

• byte_mv – numpy byte view of the buffer to write

• cursor – ammount of byte to write