Optionally loosen strictness for certain fields (#11)

Meet Manager fairly regards certain fields as optional which are listed
as M1 in the published spec. We should support serializing and
deserializing SDIF files where these values are absent.

Adds a new strict mode that requires we match the spec exactly, which is
off by default. Notes on the RelayName model that these fields are
Optional even though they're "m1".

Fixes #9.

src/sdif/fields.py

@@ -23,7 +23,10 @@
 @runtime_checkable
 class SdifModel(Protocol):
     # __attrs_attrs__: ClassVar  # pyright can't detect this without the benefit of plugins
-    identifier: ClassVar[str]
+
+    @property
+    def identifier(self) -> str:
+        ...
 
 
 class FieldType(Enum):
@@ -48,6 +51,7 @@ class FieldMetadata:
     len: int
     type: Optional[FieldType]
     m2: bool
+    override_m1: Optional[bool]
 
 
 def infer_type(attr_type: type, field_meta: FieldMetadata) -> FieldType:
@@ -77,11 +81,22 @@ def infer_type(attr_type: type, field_meta: FieldMetadata) -> FieldType:
 
 @attr.define(frozen=True)
 class FieldDef:
+    """
+    m1 describes whether the field is marked m1 in the spec.
+
+    optional describes whether the models can tolerate a None value.
+    In strict=True mode, missing values are not tolerated where m1=True.
+
+    In strict=False mode, missing values are tolerated where optional=True, even
+    if m1=True.
+    """
+
     name: str
     start: int
     len: int
     m1: bool
     m2: bool
+    optional: bool
     record_type: FieldType
     model_type: type
 
@@ -95,6 +110,7 @@ def record_fields(cls: type[SdifModel]) -> Iterator[FieldDef]:
         len=2,
         m1=True,
         m2=False,
+        optional=False,
         record_type=FieldType.const,
         model_type=str,
     )
@@ -106,7 +122,9 @@ def record_fields(cls: type[SdifModel]) -> Iterator[FieldDef]:
         meta = field.metadata["sdif"]
         assert isinstance(meta, FieldMetadata)
 
-        if is_optional_type(field.type):
+        field_is_optional = is_optional_type(field.type)
+
+        if field_is_optional:
             args: list[type] = [arg for arg in get_args(field.type) if arg != type(None)]
             if len(args) == 1:
                 (attr_type,) = args
@@ -116,13 +134,17 @@ def record_fields(cls: type[SdifModel]) -> Iterator[FieldDef]:
             attr_type = field.type
         attr_type = cast(type, attr_type)
 
-        m1 = not is_optional_type(field.type)
+        m1 = not field_is_optional
+        if meta.override_m1 is not None:
+            m1 = meta.override_m1
+
         yield FieldDef(
             name=field.name,
             start=meta.start,
             len=meta.len,
             m1=m1,
             m2=meta.m2,
+            optional=field_is_optional,
             record_type=infer_type(attr_type, meta),
             model_type=attr_type,
         )

src/sdif/model_meta.py

@@ -7,8 +7,24 @@
 REGISTERED_MODELS: dict[str, type[SdifModel]] = {}
 
 
-def spec(start: int, len: int, type: Optional[FieldType] = None, m2: bool = False):
-    return attr.field(metadata=dict(sdif=FieldMetadata(start=start, len=len, type=type, m2=m2)))
+def spec(
+    start: int,
+    len: int,
+    type: Optional[FieldType] = None,
+    m2: bool = False,
+    override_m1: Optional[bool] = None,
+):
+    return attr.field(
+        metadata=dict(
+            sdif=FieldMetadata(
+                start=start,
+                len=len,
+                type=type,
+                m2=m2,
+                override_m1=override_m1,
+            )
+        )
+    )
 
 
 if TYPE_CHECKING:

src/sdif/models.py

@@ -3,8 +3,6 @@
 from enum import Enum
 from typing import ClassVar, Optional
 
-from typing_extensions import Self
-
 from sdif.fields import FieldType
 from sdif.model_meta import model, spec
 from sdif.time import Time
@@ -110,7 +108,7 @@ class CourseStatusCode(Enum):
 
     short_meters_hytek_nonstandard = "S"
 
-    def normalize(self) -> Self:
+    def normalize(self) -> "CourseStatusCode":
         return {
             self.short_meters_int: self.short_meters,
             self.long_meters_int: self.long_meters,
@@ -451,6 +449,9 @@ class RelayName:
 
     relay_team_name:  one alpha char to concatenate with the team abbreviation in
     record C1 -- creates such names as "Dolphins A"
+
+    prelim_order and swimoff_order are M1 in the spec, but not emitted by
+    Meet Manager (https://github.com/tdsmith/sdif/issues/9).
     """
 
     identifier: ClassVar[str] = "F0"
@@ -463,8 +464,8 @@ class RelayName:
     birthdate: Optional[date] = spec(66, 8, m2=True)
     age_or_class: Optional[str] = spec(74, 2)
     sex: SexCode = spec(76, 1)
-    prelim_order: OrderCode = spec(77, 1)
-    swimoff_order: OrderCode = spec(78, 1)
+    prelim_order: Optional[OrderCode] = spec(77, 1, override_m1=True)
+    swimoff_order: Optional[OrderCode] = spec(78, 1, override_m1=True)
     finals_order: OrderCode = spec(79, 1)
     leg_time: Optional[TimeT] = spec(80, 8)
     course: Optional[CourseStatusCode] = spec(88, 1)

src/sdif/records.py

@@ -14,9 +14,9 @@
 RECORD_SEP: Final = "\r\n"
 
 
-def encode_value(field: FieldDef, value: Any) -> str:
+def encode_value(field: FieldDef, value: Any, strict: bool) -> str:
     if value is None:
-        if field.m1:
+        if (strict and field.m1) or (not strict and not field.optional):
             raise ValueError(f"No value provided for mandatory field {field=}")
         return " " * field.len
 
@@ -91,25 +91,25 @@ def encode_value(field: FieldDef, value: Any) -> str:
     assert_never(field_type)
 
 
-def encode_record(record: fields.SdifModel) -> str:
+def encode_record(record: fields.SdifModel, strict: bool) -> str:
     buf = [" "] * RECORD_CONTENT_LEN
     for field in fields.record_fields(type(record)):
         value = getattr(record, field.name)
-        encoded = encode_value(field, value)
+        encoded = encode_value(field, value, strict)
         assert len(encoded) == field.len
         buf[field.start - 1 : field.start - 1 + field.len] = encoded
     return "".join(buf)
 
 
-def encode_records(records: Iterable[fields.SdifModel]) -> str:
-    return RECORD_SEP.join(encode_record(i) for i in records)
+def encode_records(records: Iterable[fields.SdifModel], strict: bool = False) -> str:
+    return RECORD_SEP.join(encode_record(i, strict) for i in records)
 
 
-def decode_value(field: FieldDef, value: str) -> Any:
+def decode_value(field: FieldDef, value: str, strict: bool) -> Any:
     field_type = field.record_type
     stripped = value.strip()
     if stripped == "":
-        if field.m1:
+        if (strict and field.m1) or (not strict and not field.optional):
             raise ValueError(f"Blank value for mandatory field; {field=}")
         return None
     if field_type in (
@@ -163,20 +163,20 @@ def decode_value(field: FieldDef, value: str) -> Any:
 M = TypeVar("M", bound=fields.SdifModel)
 
 
-def decode_record(record: str, record_type: type[M]) -> M:
+def decode_record(record: str, record_type: type[M], strict: bool) -> M:
     kwargs = {}
     for field in fields.record_fields(record_type):
         if field.name == "identifier":
             continue
         value = record[field.start - 1 : field.start - 1 + field.len]
-        decoded = decode_value(field, value)
+        decoded = decode_value(field, value, strict)
         kwargs[field.name] = decoded
     return record_type(**kwargs)
 
 
-def decode_records(records: Iterable[str]) -> Iterable[SdifModel]:
+def decode_records(records: Iterable[str], strict: bool = False) -> Iterable[SdifModel]:
     if isinstance(records, str):
         records = records.split(RECORD_SEP)
     for record in records:
         cls = model_meta.REGISTERED_MODELS[record[:2]]
-        yield decode_record(record, cls)
+        yield decode_record(record, cls, strict)

tests/test_models.py

@@ -1,8 +1,15 @@
 import sdif.fields as fields
+import sdif.models as models
 import sdif.model_meta as model_meta
 
 
 def test_m1_m2_exclusive():
     for cls in model_meta.REGISTERED_MODELS.values():
         for field in fields.record_fields(cls):
             assert not (field.m1 and field.m2)
+
+
+def test_optional_m1_metadata():
+    (field,) = [f for f in fields.record_fields(models.RelayName) if f.name == "prelim_order"]
+    assert field.m1 == True
+    assert field.optional == True

tests/test_records.py

@@ -57,11 +57,12 @@ def test_round_trip_value(field_type: FieldType, len: int, value: Any, expected:
         len=len,
         m1=False,
         m2=False,
+        optional=True,
         record_type=field_type,
         model_type=type(value),
     )
-    assert encode_value(field_def, value) == expected
-    assert decode_value(field_def, expected) == value
+    assert encode_value(field_def, value, strict=True) == expected
+    assert decode_value(field_def, expected, strict=True) == value
 
 
 @pytest.mark.parametrize(
@@ -83,11 +84,12 @@ def test_round_trip_ish_value(
         len=len,
         m1=False,
         m2=False,
+        optional=True,
         record_type=field_type,
         model_type=type(value),
     )
-    assert encode_value(field_def, value) == expected
-    assert decode_value(field_def, expected) == roundtrip
+    assert encode_value(field_def, value, strict=True) == expected
+    assert decode_value(field_def, expected, strict=True) == roundtrip
 
 
 def test_round_trip_record():
@@ -113,3 +115,34 @@ def test_round_trip_hytek_signon():
     (record,) = decode_records([orig])
     serialized = encode_records([record])
     assert orig == serialized
+
+
+def test_round_trip_m1_optional_fields():
+    m = models.RelayName(
+        organization=None,
+        team_code="ABC",
+        relay_team_name=None,
+        swimmer_name="Joe Bloggs",
+        uss_number=None,
+        citizen=None,
+        birthdate=None,
+        age_or_class=None,
+        sex=models.SexCode.male,
+        prelim_order=None,
+        swimoff_order=None,
+        finals_order=models.OrderCode.alternate,
+        leg_time=None,
+        course=None,
+        takeoff_time=None,
+        uss_number_new=None,
+        preferred_first_name=None,
+    )
+    serialized = encode_records([m])
+    (recovered,) = decode_records(serialized)
+    assert m == recovered
+
+    with pytest.raises(ValueError):
+        encode_records([m], strict=True)
+
+    with pytest.raises(ValueError):
+        list(decode_records(serialized, strict=True))