Major work-in-progress on the binary header specification system.

segpy/datatypes.py

@@ -1,11 +1,12 @@
-DATA_SAMPLE_FORMAT = {1: 'ibm',
+# A mapping from data sampel format codes to SEG Y types.
+DATA_SAMPLE_FORMAT_TO_SEG_Y_TYPE = {1: 'ibm',
                       2: 'int32',
                       3: 'int16',
                       5: 'float32',
                       8: 'int8'}
 
 # A mapping from SEG Y data types to format characters used by the struct module
-CTYPES = {'int32':  'i',
+SEG_Y_TYPE_TO_CTYPE = {'int32':  'i',
           'uint32': 'I',
           'int16':  'h',
           'uint16': 'H',
@@ -15,7 +16,7 @@ CTYPES = {'int32':  'i',
           'ibm': 'ibm'}
 
 
-CTYPE_DESCRIPTION = {'ibm': 'IBM float',
+SEG_Y_TYPE_DESCRIPTION = {'ibm': 'IBM float',
                      'int32': '32 bit signed integer',
                      'uint32': '32 bit unsigned integer',
                      'int16': '16 bit signed integer',
@@ -25,7 +26,7 @@ CTYPE_DESCRIPTION = {'ibm': 'IBM float',
                      'uint8': '8 bit unsigned integer (byte)'}
 
 
-SIZES = dict(i=4,
+CTYPE_TO_SIZE = dict(i=4,
              I=4,
              h=2,
              H=2,
@@ -36,4 +37,4 @@ SIZES = dict(i=4,
 
 
 def size_in_bytes(ctype):
-    return SIZES[ctype]
+    return CTYPE_TO_SIZE[ctype]

segpy/header.py

@@ -0,0 +1,186 @@
+from collections import OrderedDict
+from weakref import WeakKeyDictionary
+from segpy.docstring import docstring_property
+from segpy.util import is_magic_name, underscores_to_camelcase, first_sentence, ensure_contains
+
+
+class FormatMeta(type):
+    """A metaclass for header format classes.
+    """
+
+    @classmethod
+    def __prepare__(mcs, name, bases):
+        return OrderedDict()
+
+    def __new__(mcs, name, bases, namespace):
+
+        # TODO: This is a good point to validate that the fields are in order and that the
+        # TODO: format specification is valid.  We shouldn't even build the class otherwise.
+
+        namespace['_ordered_field_names'] = tuple(name for name in namespace.keys()
+                                                  if not is_magic_name(name))
+
+        for attr_name, attr in namespace.items():
+
+            # This shenanigans is necessary so we can have all the following work is a useful way
+            # help(class), help(instance), help(class.property) and help(instance.property)
+
+            # Set the _name attribute of the field instance if it hasn't already been set
+            if isinstance(attr, NamedField):
+                if attr._name is None:
+                    attr._name = attr_name
+
+            # We rename the *class* and set its docstring so help() works usefully
+            # when called with a class containing such fields.
+            attr_class = attr.__class__
+            if issubclass(attr_class, NamedField) and attr_class is not NamedField:
+                attr_class.__name__ = underscores_to_camelcase(attr_name)
+                attr_class.__doc__ = attr.documentation
+
+        return super().__new__(mcs, name, bases, namespace)
+
+
+class NamedField:
+    """Instances of NamedField can be detected by the NamedDescriptorResolver metaclass."""
+
+    def __init__(self, value_type, offset, default, documentation):
+        self._name = None  # Set later by the metaclass
+        self._value_type = value_type
+        self._offset = int(offset)
+        self._default = self._value_type(default)
+        self._documentation = str(documentation)
+
+    @property
+    def name(self):
+        "The field name."
+        return self._name
+
+    @property
+    def value_type(self):
+        "The field value type (e.g. Int32)"
+        return self._value_type
+
+    @property
+    def offset(self):
+        "The offset it bytes from the beginning of the header."
+        return self._offset
+
+    @property
+    def default(self):
+        "The default value of the field. Must be convertible to value_type."
+        return self._default
+
+    @property
+    def documentation(self):
+        "A descriptive text string."
+        return self._documentation
+
+    @docstring_property(__doc__)
+    def __doc__(self):
+        return first_sentence(self._documentation)
+
+    # TODO: Uncomment this to get HELP
+    # def __repr__(self):
+    #     return first_sentence(self._documentation)
+
+    def __repr__(self):
+        return "{}()".format(self.__class__.__name__)
+
+
+def field(value_type, offset, default, documentation):
+    """
+    Args:
+        value_type: The type of the field (e.g. Int32)
+
+        offset: The offset in bytes for this field from the start of the header.
+
+        default: The default value for this field.
+
+        documentation: A docstring for the field. The first sentence should be usable
+            as a brief description.
+
+    Returns:
+        An instance of a subclass of NamedField class.
+    """
+
+    # Create a class specifically for this field. This class will later get
+    # renamed when the NamedDescriptorMangler metaclass does its job, to
+    # a class name based on the field name.
+
+    class SpecificField(NamedField):
+        pass
+
+    return SpecificField(value_type, offset, default, documentation)
+
+
+class ValueField:
+
+    def __init__(self, name, value_type, default, documentation):
+        self._name = name
+        self._value_type = value_type
+        self._default = default
+        self._documentation = documentation
+        self._instance_data = WeakKeyDictionary()
+
+    def __get__(self, instance, owner):
+        if owner is None:
+            return self
+        if instance not in self._instance_data:
+            return self._default
+        return self._instance_data[instance]
+
+    def __set__(self, instance, value):
+        try:
+            self._instance_data[instance] = self._value_type(value)
+        except ValueError as e:
+            raise ValueError("Assigned value {!r} for {} attribute must be convertible to {}: {}"
+                             .format(value, self._name, self._value_type.__name__, e)) from e
+
+    def __delete__(self, instance):
+        raise AttributeError("Can't delete {} attribute".format(self._name))
+
+    @docstring_property(__doc__)
+    def __doc__(self):
+        return self._documentation
+
+    # TODO: Get documentation of these descriptors working correctly
+
+
+class BuildFromFormat(type):
+    """A metaclass for building a data transfer object from a format definition."""
+
+    def __new__(mcs, name, bases, namespace, format_class):
+        """Create a new DTO class from a format class."""
+        if format_class.__class__ is not FormatMeta:
+            raise TypeError("Format class {} specified for class {} does not use the FormatMeta metaclass"
+                            .format(format_class.__name__, name))
+
+        bases = ensure_contains(bases, HeaderBase)
+
+        namespace['_format'] = format_class
+
+        for field_name in format_class._ordered_field_names:
+            format_field = getattr(format_class, field_name)
+            namespace[field_name] = ValueField(
+                name=field_name,
+                value_type=format_field.value_type,
+                default=format_field.default,
+                documentation=format_field.documentation)
+
+        return super().__new__(mcs, name, bases, namespace)
+
+    def __init__(mcs, name, bases, namespace, format_class):
+        super().__init__(mcs, name, bases)
+        pass
+
+
+class HeaderBase:
+
+    def __init__(self, **kwargs):
+        for keyword, value in kwargs.items():
+            setattr(self, keyword, value)
+
+    def __repr__(self):
+        field_names = self._format._ordered_field_names
+        return '{}({})'.format(self.__class__.__name__,
+                               ', '.join('{}={}'.format(name, getattr(self, name)) for name in field_names))

segpy/packer.py

@@ -0,0 +1,125 @@
+from collections import OrderedDict
+from struct import Struct
+from segpy.datatypes import SEG_Y_TYPE_TO_CTYPE
+from segpy.util import pairwise, intervals_partially_overlap, complementary_intervals
+
+
+def size_of(t):
+    return t.SIZE
+
+
+def compile_struct(header_format_class):
+    """Compile a struct description from a record.
+
+    Args:
+        header_format_class: A header_format class.
+
+    Returns:
+        A two-tuple containing in the zeroth element a format string which can be used with the struct.unpack function,
+        and in the second element containing a list-of-lists for field names.  Each item in the out list corresponds to
+        an element of the tuple of data values returned by struct.unpack(); each name associated with that index is a
+        field to which the unpacked value should be assigned.
+
+        format, allocations = compile_struct(TraceHeaderFormat)
+        values = struct.unpack(format)
+        field_names_to_values = {}
+        for field_names, value in zip(allocations, values):
+            for field_name in field_names:
+                field_names_to_values[field_name] = value
+        header = Header(**field_names_to_values)
+
+    """
+    fields = [getattr(header_format_class, name) for name in header_format_class._ordered_field_names]
+
+    sorted_fields = sorted(fields, key=lambda f: f.offset)
+
+    for a, b in pairwise(sorted_fields):
+        if intervals_partially_overlap(range(a.offset, a.offset + size_of(a.value_type)),
+                                       range(b.offset, b.offset + size_of(b.value_type))):
+            raise ValueError("Record fields {!r} at offset {} and {!r} at offset {} are distinct but overlap."
+                              .format(a.name, a.offset, b.name, b.offset))
+
+    offset_to_fields = OrderedDict()
+    for field in sorted_fields:
+        if field.offset not in offset_to_fields:
+            offset_to_fields[field.offset] = []
+        if len(offset_to_fields[field.offset]) > 0:
+            if offset_to_fields[field.offset][0].value_type is not field.value_type:
+                raise ValueError("Coincident fields {!r} and {!r} at offset {} have different types {!r} and {!r}"
+                                  .format(offset_to_fields[field.offset][0],
+                                          field,
+                                          field.offset,
+                                          offset_to_fields[0].value_type,
+                                          field.value_type))
+        offset_to_fields[field.offset].append(field)
+
+    # Create a list of ranges where each range spans the byte indexes covered by each field
+    field_spans = [range(offset, offset + size_of(fields[0].value_type))
+                   for offset, fields in offset_to_fields.items()]
+
+    gap_spans = complementary_intervals(field_spans)
+
+    # Create a format string usable with the struct module
+    format_chunks = []
+    representative_fields = (fields[0] for fields in offset_to_fields.values())
+    for field, gap_span in zip(representative_fields, gap_spans):
+        format_chunks.append(SEG_Y_TYPE_TO_CTYPE[field.value_type.SEG_Y_TYPE])
+        format_chunks.append('x' * len(gap_span))
+    format = ''.join(format_chunks)
+
+    # Create a list of mapping item index to field names.
+    # [0] -> ['field_1', 'field_2']
+    # [1] -> ['field_3']
+    # [2] -> ['field_4']
+    field_name_allocations = [[field.name for field in fields]
+                              for fields in offset_to_fields.values()]
+    return format, field_name_allocations
+
+
+class HeaderPacker:
+
+    def __init__(self, header_format_class):
+        self._header_format_class = header_format_class
+        self._format, self._field_name_allocations = compile_struct(header_format_class)
+        self._struct = Struct(self._format)
+
+    def pack(self, header):
+        """Pack a header into a buffer.
+        """
+        if header._format is not self._header_format_class:
+            raise TypeError("{}({}) cannot pack header of type {}.".format(
+                self.__class__.__name__,
+                self._header_format_class.__name__,
+                header.__class__.__name__
+            ))
+        values = [getattr(header, names[0]) for names in self._field_name_allocations]
+        return self._struct.pack(*values)
+
+    def unpack(self, buffer, header_class):
+        """Unpack a header into a header object.
+
+        Overwrites any existing header field values with new values
+        obtained from the buffer.
+
+        Returns:
+            The header object.
+        """
+        if header_class._format is not self._header_format_class:
+            raise TypeError("{}({}) cannot unpack header of type {}.".format(
+                self.__class__.__name__,
+                self._header_format_class.__name__,
+                header_class.__name__
+            ))
+
+        values = self._struct.unpack(buffer)
+
+        kwargs = {name: value
+                  for names, value in zip(self._field_name_allocations, values)
+                  for name in names }
+
+        return header_class(**kwargs)
+
+    def __repr__(self):
+        return "{}({})".format(
+            self.__class__.__name__,
+            self._header_format_class.__name__)

segpy/reader.py

@@ -3,7 +3,7 @@ from segpy.encoding import ASCII
 
 from segpy.portability import seekable
 from segpy.util import file_length, filename_from_handle
-from segpy.datatypes import DATA_SAMPLE_FORMAT, CTYPE_DESCRIPTION, CTYPES, size_in_bytes
+from segpy.datatypes import DATA_SAMPLE_FORMAT_TO_SEG_Y_TYPE, SEG_Y_TYPE_DESCRIPTION, SEG_Y_TYPE_TO_CTYPE, size_in_bytes
 from segpy.toolkit import (extract_revision,
                            bytes_per_sample,
                            read_binary_reel_header,
@@ -235,10 +235,10 @@ class SegYReader(object):
                              .format(start, stop_sample))
 
         dsf = self._binary_reel_header['DataSampleFormat']
-        ctype = DATA_SAMPLE_FORMAT[dsf]
+        ctype = DATA_SAMPLE_FORMAT_TO_SEG_Y_TYPE[dsf]
         start_pos = (self._trace_offset_catalog[trace_index]
                      + TRACE_HEADER_NUM_BYTES
-                     + start_sample * size_in_bytes(CTYPES[ctype]))
+                     + start_sample * size_in_bytes(SEG_Y_TYPE_TO_CTYPE[ctype]))
         num_samples_to_read = stop_sample - start_sample
 
         trace_values = read_binary_values(
@@ -334,13 +334,13 @@ class SegYReader(object):
         Returns:
             One of the values from datatypes.DATA_SAMPLE_FORMAT
         """
-        return DATA_SAMPLE_FORMAT[self._binary_reel_header['DataSampleFormat']]
+        return DATA_SAMPLE_FORMAT_TO_SEG_Y_TYPE[self._binary_reel_header['DataSampleFormat']]
 
     @property
     def data_sample_format_description(self):
         """A descriptive human-readable description of the data sample format
         """
-        return CTYPE_DESCRIPTION[self.data_sample_format]
+        return SEG_Y_TYPE_DESCRIPTION[self.data_sample_format]
 
     @property
     def encoding(self):

segpy/toolkit.py

@@ -12,7 +12,7 @@ from segpy import textual_reel_header_definition
 
 
 from segpy.catalog import CatalogBuilder
-from segpy.datatypes import CTYPES, size_in_bytes
+from segpy.datatypes import SEG_Y_TYPE_TO_CTYPE, size_in_bytes
 from segpy.encoding import guess_encoding, is_supported_encoding, UnsupportedEncodingError
 from segpy.binary_reel_header_definition import HEADER_DEF
 from segpy.ibm_float import IBMFloat
@@ -431,7 +431,7 @@ c
     Returns:
         A sequence containing count items.
     """
-    fmt = CTYPES[ctype]
+    fmt = SEG_Y_TYPE_TO_CTYPE[ctype]
     item_size = size_in_bytes(fmt)
     block_size = item_size * count
 
@@ -795,7 +795,7 @@ def write_binary_values(fh, values, ctype, pos=None, endian='>'):
         endian: '>' for big-endian data (the standard and default), '<'
             for little-endian (non-standard)
     """
-    fmt = CTYPES[ctype]
+    fmt = SEG_Y_TYPE_TO_CTYPE[ctype]
 
     if pos is not None:
         fh.seek(pos, os.SEEK_SET)
@@ -874,7 +874,7 @@ def compile_trace_header_format(endian='>'):
             fmt.append(str(shortfall) + 'x')  # Ignore bytes
             length += shortfall
 
-        ctype = CTYPES[record_spec.type]
+        ctype = SEG_Y_TYPE_TO_CTYPE[record_spec.type]
         fmt.append(ctype)
         length += size_in_bytes(ctype)
 

segpy/trace_header.py

@@ -1,121 +1,6 @@
-
-from collections import OrderedDict
-import textwrap
-from weakref import WeakKeyDictionary
-from segpy.docstring import docstring_property
+from segpy.header import FormatMeta, field, BuildFromFormat
 from segpy.types import Int32, Int16
 
-from segpy.util import underscores_to_camelcase, first_sentence
-
-
-def is_magic_name(name):
-    return len(name) > 4 and name.startswith('__') and name.endswith('__')
-
-
-class FormatMeta(type):
-    """A metaclass for header format classes.
-    """
-
-    @classmethod
-    def __prepare__(mcs, name, bases):
-        return OrderedDict()
-
-    def __new__(mcs, name, bases, namespace):
-
-        # TODO: This is a good point to validate that the fields are in order and that the
-        # TODO: format specification is valid.  We shouldn't even build the class otherwise.
-
-        namespace['ORDERED_FIELD_NAMES'] = tuple(name for name in namespace.keys()
-                                                if not is_magic_name(name))
-
-        for name, attr in namespace.items():
-
-            # This shenanigans is necessary so we can have all the following work is a useful way
-            # help(class), help(instance), help(class.property) and help(instance.property)
-
-            # Set the _name attribute of the field instance if it hasn't already been set
-            if isinstance(attr, NamedField):
-                if attr._name is None:
-                    attr._name = name
-
-            # We rename the *class* and set its docstring so help() works usefully
-            # when called with a class containing such fields.
-            attr_class = attr.__class__
-            if issubclass(attr_class, NamedField) and attr_class is not NamedField:
-                attr_class.__name__ = underscores_to_camelcase(name)
-                attr_class.__doc__ = attr.documentation
-
-        return super().__new__(mcs, name, bases, namespace)
-
-
-class NamedField:
-    """Instances of NamedField can be detected by the NamedDescriptorResolver metaclass."""
-
-    def __init__(self, value_type, offset, default, documentation):
-        self._name = None  # Set later by the metaclass
-        self._value_type = value_type
-        self._offset = offset
-        self._default = self._value_type(default)
-        self._documentation = documentation
-
-    @property
-    def name(self):
-        "The field name."
-        return self._name
-
-    @property
-    def value_type(self):
-        "The field value type (e.g. Int32)"
-        return self._value_type
-
-    @property
-    def offset(self):
-        "The offset it bytes from the beginning of the header."
-        return self._offset
-
-    @property
-    def default(self):
-        "The default value of the field. Must be convertible to value_type."
-        return self._default
-
-    @property
-    def documentation(self):
-        "A descriptive text string."
-        return self._documentation
-
-    @docstring_property(__doc__)
-    def __doc__(self):
-        return first_sentence(self._documentation)
-
-    def __repr__(self):
-        return first_sentence(self._documentation)
-
-
-def field(value_type, offset, default, documentation):
-    """
-    Args:
-        value_type: The type of the field (e.g. Int32)
-
-        offset: The offset in bytes for this field from the start of the header.
-
-        default: The default value for this field.
-
-        documentation: A docstring for the field. The first sentence should be usable
-            as a brief description.
-
-    Returns:
-        An instance of a subclass of NamedField class.
-    """
-
-    # Create a class specifically for this field. This class will later get
-    # renamed when the NamedDescriptorMangler metaclass does its job, to
-    # a class name based on the field name.
-
-    class SpecificField(NamedField):
-        pass
-
-    return SpecificField(value_type, offset, default, documentation)
-
 
 class TraceHeaderFormat(metaclass=FormatMeta):
 
@@ -234,7 +119,7 @@ class TraceHeaderFormat(metaclass=FormatMeta):
     )
 
     group_x = field(
-        Int32, offset=73, default=0, documentation=
+        Int32, offset=81, default=0, documentation=
         "Group coordinate - X. The xy_scalar applies to this value. The coordinate reference system should be "
         "identified through an extended header Location Data stanza. If the coordinate units are in seconds of arc, "
         "decimal degrees or DMS, the X values represent longitude. A positive value designates east of Greenwich "
@@ -242,7 +127,7 @@ class TraceHeaderFormat(metaclass=FormatMeta):
     )
 
     group_y = field(
-        Int32, offset=77, default=0, documentation=
+        Int32, offset=85, default=0, documentation=
         "Source coordinate - Y. The xy_scalar applies to this value. The coordinate reference system should be "
         "identified through an extended header Location Data stanza. If the coordinate units are in seconds of arc, "
         "decimal degrees or DMS, the Y values represent latitude. A positive value designates north of the equator and "
@@ -287,109 +172,7 @@ class TraceHeaderFormat(metaclass=FormatMeta):
     )
 
 
-class ValueField:
-
-    def __init__(self, name, value_type, default, documentation):
-        self._name = name
-        self._value_type = value_type
-        self._default = default
-        self._documentation = documentation
-        self._instance_data = WeakKeyDictionary()
-
-    def __get__(self, instance, owner):
-        if owner is None:
-            return self
-        if instance not in self._instance_data:
-            return self._default
-        return self._instance_data[instance]
-
-    def __set__(self, instance, value):
-        try:
-            self._instance_data[instance] = self._value_type(value)
-        except ValueError as e:
-            raise ValueError("Assigned value {!r} for {} attribute must be convertible to {}: {}"
-                             .format(value, self._name, self._value_type.__name__, e)) from e
-
-    def __delete__(self, instance):
-        raise AttributeError("Can't delete {} attribute".format(self._name))
-
-    @docstring_property(__doc__)
-    def __doc__(self):
-        return self._documentation
-
-    # TODO: Get documentation of these descriptors working correctly
-
-
-class BuildFromFormat(type):
-    """A metaclass for building a data transfer object from a format definition."""
-
-    def __new__(mcs, name, bases, namespace, format_class):
-        """Create a new DTO class from a format class."""
-        if not format_class.__class__ is FormatMeta:
-            raise TypeError("Format class {} specified for class {} does not use the FormatMeta metaclass"
-                            .format(format_class.__name__, name))
-
-        for name in format_class.ORDERED_FIELD_NAMES:
-            format_field = getattr(format_class, name)
-            namespace[name] = ValueField(name=name,
-                                         value_type=format_field.value_type,
-                                         default=format_field.default,
-                                         documentation=format_field.documentation)
-
-        return super().__new__(mcs, name, bases, namespace)
-
-    def __init__(mcs, name, bases, namespace, format_class):
-        super().__init__(mcs, name, bases)
-        pass
-
-
 class TraceHeader(metaclass=BuildFromFormat, format_class=TraceHeaderFormat):
     pass
 
-# This would build the TraceHeader class from something like the original TraceHeader at the bottom of this file.
-
-
-#
-#
-# def size_of(t):
-#     return t.SIZE
-#
-#
-# def compile_struct(record, num_bytes):
-#     """Compile a struct description from a record.
-#
-#     Args:
-#         record: A record containing Field descriptors from which ftype and offset attributes can be obtained.
-#
-#         num_bytes: The total number of bytes to be retrieved.
-#     """
-#     descriptors = (descriptor for descriptor in record.__class__.__dict__.values() if isinstance(descriptor, Field))
-#
-#     # Sort those descriptors where offset is not None
-#     sorted_descriptors = sorted(descriptor for descriptor in descriptors if descriptor.offset is not None)
-#
-#     # Check for overlaps
-#     for a, b in pairwise(sorted_descriptors):
-#         if a.offset + size_of(a.ftype) > b.offset:
-#             raise ValueError("Record fields {!r} at offset {} and {!r} at offset {} are distinct but overlap."
-#                              .format(a.name, a.offset, b.name, b.offset))
-#
-#     # Invert the mapping to give offset -> [fields]
-#     # Check that the types are the same
-#     offset_to_field = OrderedDict()
-#     for descriptor in descriptors:
-#         if descriptor.offset not in offset_to_field:
-#             offset_to_field[descriptor.offset] = []
-#         if len(offset_to_field) > 0:
-#             if offset_to_field[0].ftype is not descriptor.ftype:
-#                 raise ValueError("Coincident fields {!r} and {!r} at offset {} have different types {!r} and {!r}"
-#                                  .format(offset_to_field[0], descriptor, descriptor.offset, offset_to_field[0].ftype,
-#                                          descriptor.ftype ))
-#         offset_to_field[descriptor.offset].append(descriptor)
-#
-#     range_to_field = OrderedDict()
-#     for fields in offset_to_field.values():
-#         field = fields[0]
-#         range_to_field[range(field.offset, field.offset+size_of(field.ftype))] = fields
-
 

segpy/types.py

@@ -3,6 +3,7 @@ class Int16(int):
     MINIMUM = -32768
     MAXIMUM = 32767
     SIZE = 2
+    SEG_Y_TYPE = 'int16'
 
     def __new__(cls, *args, **kwargs):
         instance = super().__new__(cls, *args, **kwargs)
@@ -17,6 +18,7 @@ class Int32(int):
     MINIMUM = -2147483648
     MAXIMUM = 2147483647
     SIZE = 4
+    SEG_Y_TYPE = 'int32'
 
     def __new__(cls, *args, **kwargs):
         instance = super().__new__(cls, *args, **kwargs)

segpy/util.py

@@ -106,6 +106,30 @@ def complementary_intervals(intervals, start=None, stop=None):
 
     yield interval_type(index, stop)
 
+def intervals_partially_overlap(interval_a, interval_b):
+    """Determine whether two intervals partially overlap.
+
+    Args:
+        interval_a: A range or slice object.
+        interval_b: A range or slice object.
+
+    Returns:
+        True if interval_a partially overlaps interval_b, otherwise False if the intervals
+        are either disjoint or exactly coincident.
+    """
+    if interval_a == interval_b:
+        return False
+
+    if interval_a.start <= interval_b.start:
+        first_interval = interval_a
+        second_interval = interval_b
+    else:
+        first_interval = interval_b
+        second_interval = interval_a
+
+    return second_interval.start < first_interval.stop
+
+
 
 def roundrobin(*iterables):
     """Take items from each iterable in turn until all iterables are exhausted.
@@ -247,4 +271,16 @@ def almost_equal(x, y, epsilon):
     max_xy_one = max(1.0, abs(x), abs(y))
     e = epsilon * max_xy_one
     delta = abs(x - y)
-    return delta <= e
+    return delta <= e
+
+
+def ensure_contains(collection, item):
+    return collection if item in collection else conjoin(collection, item)
+
+
+def conjoin(collection, item):
+    return collection + type(collection)((item,))
+
+
+def is_magic_name(name):
+    return len(name) > 4 and name.startswith('__') and name.endswith('__')

test/strategies.py

@@ -16,6 +16,6 @@ def multiline_ascii_encodable_text(min_num_lines, max_num_lines):
         and characters which are encodable as printable 7-bit ASCII characters.
     """
 
-    return strategy(integers_in_range(0, 10))                                      \
+    return strategy(integers_in_range(min_num_lines, max_num_lines))                                      \
            .flatmap(lambda n: ([integers_in_range(*PRINTABLE_ASCII_RANGE)],) * n)  \
            .map(lambda xs: '\n'.join(bytes(x).decode('ascii') for x in xs))