Add skip lot inspections with fixed levels (#166)

* Based on combination of consignment, select compliance level.
* Each compliance level has an associated proportion of inspected consignments.
* A sampling fraction is how many of consignments should be inspected.
* Inspection is requested based on a random challenge.
* Uses sampling fraction rather than ratio inspected.
* Addresses phase 1 of #164.

docs/inspections.md

@@ -12,6 +12,50 @@ and a full version of the _Cut Flower Release Program_.
 
 Only one program can be specified at a time.
 
+### Skip Lot Program
+
+List of compliance levels and their associated ratios of inspected consignments:
+
+```yaml
+release_programs:
+  fixed_skip_lot:
+    levels:
+      - name: 1
+        sampling_fraction: 1
+      - name: 2
+        sampling_fraction: 0.5
+      - name: 3
+        sampling_fraction: 0
+```
+
+List of consignment properties (attributes) which determine into
+which compliance level a consignment belongs to:
+
+```yaml
+    track:
+      - origin
+      - commodity
+```
+
+List which assigns values of tracked consignment properties to compliance levels:
+
+```yaml
+    consignment_records:
+      - origin: Netherlands
+        commodity: Hyacinthus
+        compliance_level: 2
+      - origin: Mexico
+        commodity: Gerbera
+        compliance_level: 3
+```
+
+Default compliance level when a consignment does not have a compliance level defined:
+
+```yaml
+    default_level: 1
+```
+
+
 ### Naive Cut Flower Release Program
 
 A prototype implementation of a simple theoretical release program

popsborder/inputs.py

@@ -620,3 +620,26 @@ def load_cfrp_schedule(filename, date_format=None):
                 schedule[combo] = set()
             schedule[combo].add(date)
     return schedule
+
+
+def load_skip_lot_consignment_records(filename, tracked_properties):
+    """Load records associating consignment with skip lot compliance levels.
+
+    Only the *tracked_properties* are considered and are assumed to uniquely
+    distinguish consignment records with distinct compliance levels.
+    If the level can be converted to number, it is converted.
+    """
+    records = {}
+    # Read as CSV
+    with open(filename) as file:
+        # Import file- or format-specific items only when need.
+        # pylint: disable=import-outside-toplevel
+        import csv
+
+        for row in csv.DictReader(file):
+            combo = []
+            for tracked_property in tracked_properties:
+                combo.append(row[tracked_property])
+            level = row["compliance_level"]
+            records[tuple(combo)] = text_to_value(level)
+    return records

popsborder/skipping.py

@@ -21,8 +21,9 @@
 """
 
 import functools
+import random
 
-from .inputs import load_cfrp_schedule
+from .inputs import load_cfrp_schedule, load_skip_lot_consignment_records
 
 
 def get_inspection_needed_function(config):
@@ -39,6 +40,8 @@ def get_inspection_needed_function(config):
             # or raise exception if there is an unknown program name.
             if name == "cfrp":
                 return CutFlowerReleaseProgram(config["release_programs"][name])
+            if name == "fixed_skip_lot":
+                return FixedComplianceLevelSkipLot(config["release_programs"][name])
             elif name == "naive_cfrp":
                 return functools.partial(
                     naive_cfrp, config["release_programs"][name], name
@@ -116,3 +119,88 @@ def __call__(self, consignment, date):
                 return True, self._program_name  # inspect, is FotD
             return False, self._program_name  # release, in CFRP, but not FotD
         return True, None  # inspect, not in CFRP
+
+
+class FixedComplianceLevelSkipLot:
+    """A skip lot program which uses predefined compliance levels for consignments"""
+
+    def __init__(self, config, consignment_records=None):
+        """Creates internal consignment records, levels, and defaults.
+
+        Consignment records are read from config under key 'consignment_records',
+        from a file linked in config under consignment_records/file_name,
+        or directly from the consignment_records parameter.
+        While the records in configuration are in a list, the parameter is a dictionary
+        with consignment property values as keys and compliance level names as values.
+        """
+        self._program_name = config.get("name", "fixed_skip_lot")
+        self._tracked_properties = config.get("track")
+        self._default_level = config.get("default_level")
+
+        levels = config.get("levels")
+        self._levels = {}
+        for level in levels:
+            if "name" not in level:
+                raise ValueError("Each level needs to have 'name'")
+            if "sampling_fraction" not in level:
+                raise ValueError("Each level needs to have 'sampling_fraction'")
+            self._levels[level["name"]] = level
+
+        if consignment_records:
+            self._consignment_records = consignment_records.copy()
+        else:
+            records_config = config["consignment_records"]
+            if "file_name" in records_config:
+                self._consignment_records = load_skip_lot_consignment_records(
+                    consignment_records["file_name"],
+                    tracked_properties=self._tracked_properties,
+                )
+            elif isinstance(records_config, list):
+                self._consignment_records = {}
+                for record in records_config:
+                    key = []
+                    for tracked_property in self._tracked_properties:
+                        key.append(record[tracked_property])
+                    self._consignment_records[tuple(key)] = record["compliance_level"]
+            else:
+                raise ValueError(
+                    "The 'consignment_records' config needs to be a list "
+                    "or contain 'file_name'"
+                )
+
+    def compliance_level_for_consignment(self, consignment):
+        """Get compliance level associated with a given consignment.
+
+        The level is selected based on consignment properties.
+        """
+        key = []
+        for name in self._tracked_properties:
+            try:
+                property_value = getattr(consignment, name)
+            except AttributeError as error:
+                raise ValueError(
+                    f"Consignment does not have a property '{name}'"
+                ) from error
+            key.append(property_value)
+        key = tuple(key)
+        if key not in self._consignment_records:
+            self._consignment_records[key] = self._default_level
+            return self._default_level
+        return self._consignment_records[key]
+
+    def sampling_fraction_for_level(self, level):
+        """Get ratio of items or boxes to inspect associated with a compliance level"""
+        return self._levels[level]["sampling_fraction"]
+
+    def __call__(self, consignment, date):
+        """Decide whether the consignment should be inspected or not.
+
+        Returns boolean (True for inspect) and this program name (always because it
+        is always applied even to unknown consignments because there is a default
+        compliance level).
+        """
+        level = self.compliance_level_for_consignment(consignment)
+        sampling_fraction = self.sampling_fraction_for_level(level)
+        if random.random() <= sampling_fraction:
+            return True, self._program_name
+        return False, self._program_name

tests/test_cfrp.py

@@ -1,4 +1,4 @@
-"""Test functions for skipping inspections directly"""
+"""Test functions for skipping with CFRP"""
 
 import datetime
 

tests/test_fixed_skip_lot.py

@@ -0,0 +1,188 @@
+"""Test fixed skip lot functionality"""
+
+import pytest
+
+from popsborder.consignments import Consignment, get_consignment_generator
+from popsborder.inputs import (
+    load_configuration_yaml_from_text,
+    load_skip_lot_consignment_records,
+)
+from popsborder.simulation import random_seed
+from popsborder.skipping import (
+    FixedComplianceLevelSkipLot,
+    get_inspection_needed_function,
+    inspect_always,
+)
+
+BASE_CONSIGNMENT_CONFIG = """\
+consignment:
+  generation_method: parameter_based
+  items_per_box:
+    default: 200
+    air:
+      default: 200
+    maritime:
+      default: 200
+  parameter_based:
+    boxes:
+      min: 1
+      max: 100
+    origins:
+      - Netherlands
+      - Mexico
+      - Israel
+    flowers:
+      - Hyacinthus
+      - Rosa
+      - Gerbera
+    ports:
+      - NY JFK CBP
+      - FL Miami Air CBP
+      - HI Honolulu CBP
+"""
+
+CONFIG = """\
+release_programs:
+  fixed_skip_lot:
+    name: Skip Lot
+    track:
+      - origin
+      - commodity
+    levels:
+      - name: 1
+        sampling_fraction: 1
+      - name: 2
+        sampling_fraction: 0.5
+      - name: 3
+        sampling_fraction: 0
+    default_level: 1
+    consignment_records:
+      - origin: Netherlands
+        commodity: Hyacinthus
+        compliance_level: 2
+      - origin: Mexico
+        commodity: Gerbera
+        compliance_level: 3
+"""
+
+RECORDS_CSV_TEXT = """\
+origin,commodity,compliance_level
+Netherlands,Hyacinthus,2
+Mexico,Gerbera,3
+"""
+
+
+def simple_consignment(flower, origin, date=None, port="FL Miami Air CBP"):
+    """Get consignment with some default values"""
+    return Consignment(
+        flower=flower,
+        num_items=0,
+        items=0,
+        items_per_box=0,
+        num_boxes=0,
+        date=date,
+        boxes=[],
+        pathway="airport",
+        port=port,
+        origin=origin,
+    )
+
+
+def test_fixed_skip_lot():
+    """Check that fixed skip lot program is accepted and gives expected results"""
+    consignment_generator = get_consignment_generator(
+        load_configuration_yaml_from_text(BASE_CONSIGNMENT_CONFIG)
+    )
+    is_needed_function = get_inspection_needed_function(
+        load_configuration_yaml_from_text(CONFIG)
+    )
+    # The following assumes what is the default returned by the get function,
+    # i.e., it relies on its internals, not the interface.
+    # pylint: disable=comparison-with-callable
+    assert is_needed_function != inspect_always
+    for seed in range(10):
+        # We run with different, but fixed seeded so we can know which seed fails.
+        random_seed(seed)
+        consignment = consignment_generator.generate_consignment()
+        inspect, program = is_needed_function(consignment, consignment.date)
+        assert isinstance(inspect, bool)
+        # Testing custom name (program name should be always present for skip lot)
+        assert program == "Skip Lot"
+
+
+def test_load_consignment_records(tmp_path):
+    """Check that schedule loads from a CSV file with custom date format"""
+    file = tmp_path / "records.csv"
+    file.write_text(RECORDS_CSV_TEXT)
+    records = load_skip_lot_consignment_records(
+        file, tracked_properties=["origin", "commodity"]
+    )
+    assert records == {("Netherlands", "Hyacinthus"): 2, ("Mexico", "Gerbera"): 3}
+
+
+def test_sometimes_inspect_in_program():
+    """Inspection is requested at least sometimes when consignment is in the program"""
+    program = FixedComplianceLevelSkipLot(
+        load_configuration_yaml_from_text(CONFIG)["release_programs"]["fixed_skip_lot"]
+    )
+    consignment = simple_consignment(flower="Hyacinthus", origin="Netherlands")
+    inspected = 0
+    for seed in range(10):
+        random_seed(seed)
+        inspect, program_name = program(consignment, consignment.date)
+        inspected += int(inspect)
+        assert program_name == "Skip Lot"
+    assert inspected, "With the seeds, we expect at least one inspection to happen"
+
+
+def test_never_inspect_in_program():
+    """Inspection is not requested when consignment is in a zero inspections level"""
+    program = FixedComplianceLevelSkipLot(
+        load_configuration_yaml_from_text(CONFIG)["release_programs"]["fixed_skip_lot"]
+    )
+    consignment = simple_consignment(flower="Gerbera", origin="Mexico")
+    for seed in range(10):
+        random_seed(seed)
+        inspect, program_name = program(consignment, consignment.date)
+        assert (
+            not inspect
+        ), "We disabled inspections completely for this inspection level"
+        assert program_name == "Skip Lot"
+
+
+def test_inspect_not_in_program():
+    """Check inspection is requested when consignment is not in the program"""
+    program = FixedComplianceLevelSkipLot(
+        load_configuration_yaml_from_text(CONFIG)["release_programs"]["fixed_skip_lot"]
+    )
+    consignment = simple_consignment(flower="Rosa", origin="Netherlands")
+    for seed in range(10):
+        random_seed(seed)
+        inspect, program_name = program(consignment, consignment.date)
+        assert inspect
+        assert program_name == "Skip Lot"
+
+
+@pytest.mark.parametrize(["level", "fraction"], [(1, 1), (2, 0.5), (3, 0)])
+def test_fraction(level, fraction):
+    """Correct fraction is returned for a level"""
+    program = FixedComplianceLevelSkipLot(
+        load_configuration_yaml_from_text(CONFIG)["release_programs"]["fixed_skip_lot"]
+    )
+    assert program.sampling_fraction_for_level(level) == fraction
+
+
+@pytest.mark.parametrize(
+    ["consignment", "level"],
+    [
+        (simple_consignment(flower="Hyacinthus", origin="Netherlands"), 2),
+        (simple_consignment(flower="Gerbera", origin="Mexico"), 3),
+        (simple_consignment(flower="Rosa", origin="Israel"), 1),
+    ],
+)
+def test_level(consignment, level):
+    """Correct level is returned for a shipment"""
+    program = FixedComplianceLevelSkipLot(
+        load_configuration_yaml_from_text(CONFIG)["release_programs"]["fixed_skip_lot"]
+    )
+    assert program.compliance_level_for_consignment(consignment) == level