def get_message_source()

in transcoder/source/SourceUtil.py [0:0]

• 31 lines of code
• 7 McCabe index (conditional complexity)

def get_message_source(source_loc: str,  # pylint: disable=too-many-arguments
                       source_file_encoding: str, source_file_format_type: str,
                       endian: str, skip_bytes: int = 0, skip_lines: int = 0,
                        message_skip_bytes: int = 0, prefix_length: int = 2,
                        base64: bool = False, base64_urlsafe: bool = False) -> Source:
    """Returns a Source implementation instance based on the supplied source name"""

    source: Source = None

    if source_file_format_type == PcapFileMessageSource.source_type_identifier():
        source = PcapFileMessageSource(source_loc, message_skip_bytes=message_skip_bytes)
    elif source_file_format_type == LengthDelimitedFileMessageSource.source_type_identifier():

        source = LengthDelimitedFileMessageSource(source_loc, skip_bytes=skip_bytes,
                                                  message_skip_bytes=message_skip_bytes,
                                                  prefix_length=prefix_length)

    elif source_file_format_type == LineDelimitedFileMessageSource.source_type_identifier():

        if base64 is True:
            line_encoding = LineEncoding.BASE_64
        elif base64_urlsafe is True:
            line_encoding = LineEncoding.BASE_64_URL_SAFE
        else:
            line_encoding = LineEncoding.NONE

        source = LineDelimitedFileMessageSource(source_loc,
                                                encoding=source_file_encoding,
                                                skip_lines=skip_lines,
                                                line_encoding=line_encoding,
                                                message_skip_bytes=message_skip_bytes)
    elif source_file_format_type == CmeBinaryPacketFileMessageSource.source_type_identifier():
        source = CmeBinaryPacketFileMessageSource(source_loc, endian, skip_bytes=skip_bytes,
                                                  message_skip_bytes=message_skip_bytes,
                                                  prefix_length=prefix_length)
    else:
        raise UnsupportedFileTypeError(f'Source {source_loc} is not supported')
    return source