Merge pull request #515 from NHSDigital/feature/eja-eli-579-adding-derivation-for-previous-dose-calculations

Feature/eja eli 579 adding derivation for previous dose calculations

Author: eddalmond1

poetry.lock

@@ -0,0 +1,28 @@
+from eligibility_signposting_api.services.processors.derived_values.add_days_handler import AddDaysHandler
+from eligibility_signposting_api.services.processors.derived_values.base import (
+    DerivedValueContext,
+    DerivedValueHandler,
+)
+from eligibility_signposting_api.services.processors.derived_values.registry import (
+    DerivedValueRegistry,
+    get_registry,
+)
+
+__all__ = [
+    "AddDaysHandler",
+    "DerivedValueContext",
+    "DerivedValueHandler",
+    "DerivedValueRegistry",
+    "get_registry",
+]
+
+# Register default handlers
+DerivedValueRegistry.register_default(
+    AddDaysHandler(
+        default_days=91,
+        vaccine_type_days={
+            "COVID": 91,  # 91 days between COVID vaccinations
+            # Add other vaccine-specific configurations here as needed.
+        },
+    )
+)

src/eligibility_signposting_api/services/processors/derived_values/__init__.py

@@ -0,0 +1,180 @@
+from datetime import UTC, datetime, timedelta
+from typing import ClassVar
+
+from eligibility_signposting_api.services.processors.derived_values.base import (
+    DerivedValueContext,
+    DerivedValueHandler,
+)
+
+
+class AddDaysHandler(DerivedValueHandler):
+    """Handler for adding days to a date value.
+
+    This handler calculates derived dates by adding a configurable number of days
+    to a source date attribute. It supports:
+    - Default days value for all vaccine types
+    - Vaccine-specific days configuration
+    - Configurable mapping of derived attributes to source attributes
+
+    Example token: [[TARGET.COVID.NEXT_DOSE_DUE:ADD_DAYS(91)]]
+    This would add 91 days to COVID's LAST_SUCCESSFUL_DATE to calculate NEXT_DOSE_DUE.
+
+    The number of days can be specified in three ways (in order of precedence):
+    1. In the token itself: :ADD_DAYS(91)
+    2. In the vaccine_type_days configuration
+    3. Using the default_days value
+    """
+
+    function_name: str = "ADD_DAYS"
+
+    # Mapping of derived attribute names to their source attributes
+    DERIVED_ATTRIBUTE_SOURCES: ClassVar[dict[str, str]] = {
+        "NEXT_DOSE_DUE": "LAST_SUCCESSFUL_DATE",
+    }
+
+    def __init__(
+        self,
+        default_days: int = 91,
+        vaccine_type_days: dict[str, int] | None = None,
+    ) -> None:
+        """Initialize the AddDaysHandler.
+
+        Args:
+            default_days: Default number of days to add when not specified
+                        in token or vaccine_type_days. Defaults to 91.
+            vaccine_type_days: Dictionary mapping vaccine types to their
+                        specific days values. E.g., {"COVID": 91, "FLU": 365}
+        """
+        self.default_days = default_days
+        self.vaccine_type_days = vaccine_type_days or {}
+
+    def get_source_attribute(self, target_attribute: str, function_args: str | None = None) -> str:
+        """Get the source attribute for a derived attribute.
+
+        Check if source is provided in function args (e.g., ADD_DAYS(91, SOURCE_FIELD)).
+        If not, fall back to mapping or return target_attribute as default.
+
+        Args:
+            target_attribute: The derived attribute name (e.g., 'NEXT_DOSE_DUE')
+            function_args: Optional arguments from token (e.g., '91, LAST_SUCCESSFUL_DATE')
+
+        Returns:
+            The source attribute name (e.g., 'LAST_SUCCESSFUL_DATE')
+        """
+        if function_args and "," in function_args:
+            # Extract source from args if present (second argument)
+            parts = [p.strip() for p in function_args.split(",")]
+            if len(parts) > 1 and parts[1]:
+                return parts[1].upper()
+
+        return self.DERIVED_ATTRIBUTE_SOURCES.get(target_attribute, target_attribute)
+
+    def calculate(self, context: DerivedValueContext) -> str:
+        """Calculate a date with added days.
+
+        Args:
+            context: DerivedValueContext containing:
+                - person_data: List of attribute dictionaries
+                - attribute_name: Vaccine type (e.g., 'COVID')
+                - source_attribute: The source date attribute
+                - function_args: Optional days override from token
+                - date_format: Optional output date format
+
+        Returns:
+            The calculated date as a formatted string
+
+        Raises:
+            ValueError: If source date is not found or invalid
+        """
+        source_date = self._find_source_date(context)
+        if not source_date:
+            return ""
+
+        days_to_add = self._get_days_to_add(context)
+        calculated_date = self._add_days_to_date(source_date, days_to_add)
+
+        return self._format_date(calculated_date, context.date_format)
+
+    def _find_source_date(self, context: DerivedValueContext) -> str | None:
+        """Find the source date value from person data.
+
+        Args:
+            context: The derived value context
+
+        Returns:
+            The source date string or None if not found
+        """
+        source_attr = context.source_attribute
+        if not source_attr:
+            return None
+
+        for attribute in context.person_data:
+            if attribute.get("ATTRIBUTE_TYPE") == context.attribute_name:
+                return attribute.get(source_attr)
+
+        return None
+
+    def _get_days_to_add(self, context: DerivedValueContext) -> int:
+        """Determine the number of days to add.
+
+        Priority:
+        1. Function argument from token (e.g., :ADD_DAYS(91))
+        2. Vaccine-specific configuration
+        3. Default days
+
+        Args:
+            context: The derived value context
+
+        Returns:
+            Number of days to add
+        """
+        # Priority 1: Token argument (if non-empty)
+        if context.function_args:
+            args = context.function_args.split(",")[0].strip()
+            if args:
+                try:
+                    return int(args)
+                except ValueError as e:
+                    message = f"Invalid days argument '{args}' for ADD_DAYS function. Expected an integer."
+                    raise ValueError(message) from e
+
+        # Priority 2: Vaccine-specific configuration
+        if context.attribute_name in self.vaccine_type_days:
+            return self.vaccine_type_days[context.attribute_name]
+
+        # Priority 3: Default
+        return self.default_days
+
+    def _add_days_to_date(self, date_str: str, days: int) -> datetime:
+        """Parse a date string and add days.
+
+        Args:
+            date_str: Date in YYYYMMDD format
+            days: Number of days to add
+
+        Returns:
+            The calculated datetime
+
+        Raises:
+            ValueError: If date format is invalid
+        """
+        try:
+            date_obj = datetime.strptime(date_str, "%Y%m%d").replace(tzinfo=UTC)
+            return date_obj + timedelta(days=days)
+        except ValueError as e:
+            message = f"Invalid date format: {date_str}"
+            raise ValueError(message) from e
+
+    def _format_date(self, date_obj: datetime, date_format: str | None) -> str:
+        """Format a datetime object.
+
+        Args:
+            date_obj: The datetime to format
+            date_format: Optional strftime format string
+
+        Returns:
+            Formatted date string. If no format specified, returns YYYYMMDD.
+        """
+        if date_format:
+            return date_obj.strftime(date_format)
+        return date_obj.strftime("%Y%m%d")

src/eligibility_signposting_api/services/processors/derived_values/add_days_handler.py

@@ -0,0 +1,67 @@
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass
+class DerivedValueContext:
+    """Context object containing all data needed for derived value calculation.
+
+    Attributes:
+        person_data: List of person attribute dictionaries
+        attribute_name: The condition/vaccine type (e.g., 'COVID', 'RSV')
+        source_attribute: The source attribute to derive from (e.g., 'LAST_SUCCESSFUL_DATE')
+        function_args: Arguments passed to the function (e.g., number of days)
+        date_format: Optional date format string for output formatting
+    """
+
+    person_data: list[dict[str, Any]]
+    attribute_name: str
+    source_attribute: str | None
+    function_args: str | None
+    date_format: str | None
+
+
+class DerivedValueHandler(ABC):
+    """Abstract base class for derived value handlers.
+
+    Derived value handlers compute values that don't exist directly in the data
+    but are calculated from existing attributes. Each handler is responsible for
+    a specific type of calculation (e.g., adding days to a date).
+
+    To create a new derived value handler:
+    1. Subclass DerivedValueHandler
+    2. Set the `function_name` class attribute to the token function name (e.g., 'ADD_DAYS')
+    3. Implement the `calculate` method
+    4. Register the handler with the DerivedValueRegistry
+    """
+
+    function_name: str = ""
+
+    @abstractmethod
+    def calculate(self, context: DerivedValueContext) -> str:
+        """Calculate the derived value.
+
+        Args:
+            context: DerivedValueContext containing all necessary data
+
+        Returns:
+            The calculated value as a string
+
+        Raises:
+            ValueError: If the calculation cannot be performed
+        """
+
+    @abstractmethod
+    def get_source_attribute(self, target_attribute: str, function_args: str | None = None) -> str:
+        """Get the source attribute name needed for this derived value.
+
+        For example, NEXT_DOSE_DUE derives from LAST_SUCCESSFUL_DATE.
+
+        Args:
+            target_attribute: The target derived attribute name (e.g., 'NEXT_DOSE_DUE')
+            function_args: Optional arguments from the token function call
+
+        Returns:
+            The source attribute name to use for calculation
+        """