feat: plugins functionality for protocol conversions

canvas_sdk/effects/protocol_card/protocol_card.py

@@ -51,6 +51,7 @@ class Meta:
     recommendations: list[Recommendation] = []
     status: Status = Status.DUE
     feedback_enabled: bool = False
+    due_in: int = -1
 
     @property
     def values(self) -> dict[str, Any]:
@@ -63,6 +64,7 @@ def values(self) -> dict[str, Any]:
             ],
             "status": self.status.value,
             "feedback_enabled": self.feedback_enabled,
+            "due_in": self.due_in,
         }
 
     @property

canvas_sdk/effects/protocol_card/tests.py

@@ -25,7 +25,7 @@ def test_apply_method_succeeds_with_patient_id_and_key() -> None:
     applied = p.apply()
     assert (
         applied.payload
-        == '{"patient": "uuid", "patient_filter": null, "key": "something-unique", "data": {"title": "", "narrative": "", "recommendations": [], "status": "due", "feedback_enabled": false}}'
+        == '{"patient": "uuid", "patient_filter": null, "key": "something-unique", "data": {"title": "", "narrative": "", "recommendations": [], "status": "due", "feedback_enabled": false, "due_in": -1}}'
     )
 
 
@@ -112,7 +112,6 @@ def test_add_recommendations(
     p = ProtocolCard(**init_params)
     p.add_recommendation(**rec1_params)
     p.recommendations.append(Recommendation(**rec2_params))
-
     assert p.values == {
         "title": init_params["title"],
         "narrative": init_params["narrative"],
@@ -135,6 +134,7 @@ def test_add_recommendations(
             },
         ],
         "status": "due",
+        "due_in": -1,
         "feedback_enabled": False,
     }
 

canvas_sdk/protocols/base.py

@@ -1,5 +1,7 @@
 from typing import Any
 
+import arrow
+
 from canvas_sdk.data.client import GQL_CLIENT
 from canvas_sdk.handlers.base import BaseHandler
 

canvas_sdk/protocols/clinical_quality_measure.py

@@ -1,6 +1,13 @@
-from typing import Any
+from typing import Any, cast
 
+import arrow
+from django.db.models import Model
+
+from canvas_sdk.events import EventType
 from canvas_sdk.protocols.base import BaseProtocol
+from canvas_sdk.protocols.timeframe import Timeframe
+from canvas_sdk.v1.data.condition import Condition
+from canvas_sdk.v1.data.medication import Medication
 
 
 class ClinicalQualityMeasure(BaseProtocol):
@@ -23,6 +30,11 @@ class Meta:
         is_abstract: bool = False
         is_predictive: bool = False
 
+    def __init__(self, *args: Any, **kwargs: Any):
+        self._patient_id: str | None = None
+        self.now = arrow.utcnow()
+        super().__init__(*args, **kwargs)
+
     @classmethod
     def _meta(cls) -> dict[str, Any]:
         """
@@ -41,3 +53,47 @@ def protocol_key(cls) -> str:
         External key used to identify the protocol.
         """
         return cls.__name__
+
+    @property
+    def timeframe(self) -> Timeframe:
+        """The default Timeframe (self.timeframe) for all protocols.
+        This defaults to have a start of 1 year ago and an end time of the current time.
+        Plugin authors can override this if a different timeframe is desired.
+        """
+        end = self.now
+        return Timeframe(start=end.shift(years=-1), end=end)
+
+    # TODO: This approach should be considered against the alternative of just including the patient
+    #  ID in the event context, given that so many events will be patient-centric.
+    def patient_id_from_target(self) -> str:
+        """
+        Get and return the patient ID from an event target.
+
+        This method will attempt to obtain the patient ID from the event target for supported event
+        types. It stores the patient ID on a member variable so that it can be referenced without
+        incurring more SQL queries.
+        """
+
+        def patient_id(model: type[Model]) -> str:
+            return cast(
+                str,
+                model.objects.select_related("patient")
+                .values_list("patient__id")
+                .get(id=self.event.target)[0],
+            )
+
+        if not self._patient_id:
+            # TODO: Add cases for ProtocolOverride
+            match self.event.type:
+                case EventType.CONDITION_CREATED | EventType.CONDITION_UPDATED:
+                    self._patient_id = patient_id(Condition)
+                case (
+                    EventType.MEDICATION_LIST_ITEM_CREATED | EventType.MEDICATION_LIST_ITEM_UPDATED
+                ):
+                    self._patient_id = patient_id(Medication)
+                case _:
+                    raise AssertionError(
+                        f"Event type {self.event.type} not supported by 'patient_id_from_event'"
+                    )
+
+        return self._patient_id

canvas_sdk/protocols/timeframe.py

@@ -0,0 +1,39 @@
+import arrow
+
+
+class Timeframe:
+    """A class representing a timeframe with a start and and end."""
+
+    def __init__(self, start: arrow.Arrow, end: arrow.Arrow):
+        self.start = start
+        self.end = end
+
+    def __str__(self) -> str:
+        return f"<Timeframe start={self.start}, end={self.end}>"
+
+    @property
+    def duration(self) -> int:
+        """Returns the number of days in the timeframe."""
+        return (self.end - self.start).days
+
+    def increased_by(self, years: int = 0, months: int = 0, days: int = 0) -> "Timeframe":
+        """Returns a new Timeframe object increased by the years, months, days in the arguments."""
+        start = self.start
+        end = self.end
+
+        if years > 0:
+            end = end.shift(years=years)
+        elif years < 0:
+            start = start.shift(years=years)
+
+        if months > 0:
+            end = end.shift(months=months)
+        elif months < 0:
+            start = start.shift(months=months)
+
+        if days > 0:
+            end = end.shift(days=days)
+        elif days < 0:
+            start = start.shift(days=days)
+
+        return Timeframe(start=start, end=end)

canvas_sdk/v1/data/__init__.py

@@ -1,3 +1,4 @@
+from .billing import BillingLineItem
 from .condition import Condition, ConditionCoding
 from .medication import Medication, MedicationCoding
 from .patient import Patient

canvas_sdk/v1/data/base.py

@@ -1,10 +1,12 @@
+from abc import abstractmethod
 from collections.abc import Container
-from typing import TYPE_CHECKING, Self, Type, cast
+from typing import TYPE_CHECKING, Any, Dict, Protocol, Type, cast
 
 from django.db import models
 from django.db.models import Q
 
 if TYPE_CHECKING:
+    from canvas_sdk.protocols.timeframe import Timeframe
     from canvas_sdk.value_set.value_set import ValueSet
 
 
@@ -29,8 +31,38 @@ def for_patient(self, patient_id: str) -> "Self":
         return self.filter(patient__id=patient_id)
 
 
-class ValueSetLookupQuerySet(CommittableQuerySet):
-    """A QuerySet that can filter objects based on a ValueSet."""
+class BaseQuerySet(models.QuerySet):
+    """A base QuerySet inherited from Django's model.Queryset."""
+
+    pass
+
+
+class QuerySetProtocol(Protocol):
+    """A typing protocol for use in mixins into models.QuerySet-inherited classes."""
+
+    def filter(self, *args: Any, **kwargs: Any) -> models.QuerySet[Any]:
+        """Django's models.QuerySet filter method."""
+        ...
+
+
+class ValueSetLookupQuerySetProtocol(QuerySetProtocol):
+    """A typing protocol for use in mixins using value set lookup methods."""
+
+    @staticmethod
+    @abstractmethod
+    def codings(value_set: Type["ValueSet"]) -> tuple[tuple[str, set[str]]]:
+        """A protocol method for defining codings."""
+        raise NotImplementedError
+
+    @staticmethod
+    @abstractmethod
+    def q_object(system: str, codes: Container[str]) -> Q:
+        """A protocol method for defining Q objects for value set lookups."""
+        raise NotImplementedError
+
+
+class ValueSetLookupQuerySetMixin(ValueSetLookupQuerySetProtocol):
+    """A QuerySet mixin that can filter objects based on a ValueSet."""
 
     def find(self, value_set: Type["ValueSet"]) -> "Self":
         """
@@ -54,7 +86,7 @@ def find(self, value_set: Type["ValueSet"]) -> "Self":
     @staticmethod
     def codings(value_set: Type["ValueSet"]) -> tuple[tuple[str, set[str]]]:
         """Provide a sequence of tuples where each tuple is a code system URL and a set of codes."""
-        values_dict = value_set.values
+        values_dict = cast(Dict, value_set.values)
         return cast(
             tuple[tuple[str, set[str]]],
             tuple(
@@ -72,7 +104,7 @@ def q_object(system: str, codes: Container[str]) -> Q:
         return Q(codings__system=system, codings__code__in=codes)
 
 
-class ValueSetLookupByNameQuerySet(ValueSetLookupQuerySet):
+class ValueSetLookupByNameQuerySetMixin(ValueSetLookupQuerySetMixin):
     """
     QuerySet for ValueSet lookups using code system name rather than URL.
 
@@ -85,7 +117,7 @@ def codings(value_set: Type["ValueSet"]) -> tuple[tuple[str, set[str]]]:
         """
         Provide a sequence of tuples where each tuple is a code system name and a set of codes.
         """
-        values_dict = value_set.values
+        values_dict = cast(Dict, value_set.values)
         return cast(
             tuple[tuple[str, set[str]]],
             tuple(
@@ -94,3 +126,51 @@ def codings(value_set: Type["ValueSet"]) -> tuple[tuple[str, set[str]]]:
                 if i[0] in values_dict
             ),
         )
+
+
+class TimeframeLookupQuerySetProtocol(QuerySetProtocol):
+    """A typing protocol for use in TimeframeLookupQuerySetMixin."""
+
+    @property
+    @abstractmethod
+    def timeframe_filter_field(self) -> str:
+        """A protocol method for timeframe_filter_field."""
+        raise NotImplementedError
+
+
+class TimeframeLookupQuerySetMixin(TimeframeLookupQuerySetProtocol):
+    """A class that adds queryset functionality to filter using timeframes."""
+
+    @property
+    def timeframe_filter_field(self) -> str:
+        """Returns the field that should be filtered on. Can be overridden for different models."""
+        return "note__datetime_of_service"
+
+    def within(self, timeframe: "Timeframe") -> models.QuerySet:
+        """A method to filter a queryset for datetimes within a timeframe."""
+        return self.filter(
+            **{
+                f"{self.timeframe_filter_field}__range": (
+                    timeframe.start.datetime,
+                    timeframe.end.datetime,
+                )
+            }
+        )
+
+
+class ValueSetLookupQuerySet(BaseQuerySet, ValueSetLookupQuerySetMixin):
+    """A class that includes methods for looking up value sets."""
+
+    pass
+
+
+class ValueSetLookupByNameQuerySet(BaseQuerySet, ValueSetLookupByNameQuerySetMixin):
+    """A class that includes methods for looking up value sets by name."""
+
+    pass
+
+
+class ValueSetTimeframeLookupQuerySet(ValueSetLookupQuerySet, TimeframeLookupQuerySetMixin):
+    """A class that includes methods for looking up value sets and using timeframes."""
+
+    pass

canvas_sdk/v1/data/billing.py

@@ -0,0 +1,59 @@
+from typing import TYPE_CHECKING, Type
+
+from django.db import models
+
+from canvas_sdk.v1.data.base import ValueSetTimeframeLookupQuerySet
+from canvas_sdk.v1.data.note import Note
+from canvas_sdk.v1.data.patient import Patient
+from canvas_sdk.value_set.value_set import CodeConstants
+
+if TYPE_CHECKING:
+    from canvas_sdk.value_set.value_set import ValueSet
+
+
+class BillingLineItemQuerySet(ValueSetTimeframeLookupQuerySet):
+    """A class that adds functionality to filter BillingLineItem objects."""
+
+    def find(self, value_set: Type["ValueSet"]) -> models.QuerySet:
+        """
+        This method is overridden to use for BillingLineItem CPT codes.
+        The codes are saved as string values in the BillingLineItem.cpt field,
+        which differs from other coding models.
+        """
+        values_dict = value_set.values
+        return self.filter(cpt__in=values_dict.get(CodeConstants.HCPCS, []))
+
+
+class BillingLineItemStatus(models.TextChoices):
+    """Billing line item status."""
+
+    ACTIVE = "active", "Active"
+    REMOVED = "removed", "Removed"
+
+
+class BillingLineItem(models.Model):
+    """BillingLineItem."""
+
+    class Meta:
+        managed = False
+        app_label = "canvas_sdk"
+        db_table = "canvas_sdk_data_api_billinglineitem_001"
+
+    # objects = BillingLineItemQuerySet.as_manager()
+    objects = models.Manager().from_queryset(BillingLineItemQuerySet)()
+
+    id = models.UUIDField()
+    dbid = models.BigIntegerField(primary_key=True)
+    created = models.DateTimeField()
+    modified = models.DateTimeField()
+    note = models.ForeignKey(Note, on_delete=models.DO_NOTHING, related_name="billing_line_items")
+    patient = models.ForeignKey(
+        Patient, on_delete=models.DO_NOTHING, related_name="billing_line_items"
+    )
+    cpt = models.CharField()
+    charge = models.DecimalField()
+    description = models.CharField()
+    units = models.IntegerField()
+    command_type = models.CharField()
+    command_id = models.IntegerField()
+    status = models.CharField(choices=BillingLineItemStatus.choices)

canvas_sdk/v1/data/condition.py

@@ -1,4 +1,5 @@
 from django.db import models
+from django.db.models import TextChoices
 
 from canvas_sdk.v1.data.base import (
     CommittableModelManager,
@@ -9,6 +10,16 @@
 from canvas_sdk.v1.data.user import CanvasUser
 
 
+class ClinicalStatus(TextChoices):
+    """Condition clinical status."""
+
+    ACTIVE = "active", "active"
+    RELAPSE = "relapse", "relapse"
+    REMISSION = "remission", "remission"
+    RESOLVED = "resolved", "resolved"
+    INVESTIGATIVE = "investigative", "investigative"
+
+
 class ConditionQuerySet(ValueSetLookupQuerySet):
     """ConditionQuerySet."""
 
@@ -29,6 +40,7 @@ class Meta:
     dbid = models.BigIntegerField(primary_key=True)
     onset_date = models.DateField()
     resolution_date = models.DateField()
+    clinical_status = models.CharField(choices=ClinicalStatus.choices)
     deleted = models.BooleanField()
     entered_in_error = models.ForeignKey(CanvasUser, on_delete=models.DO_NOTHING)
     committer = models.ForeignKey(CanvasUser, on_delete=models.DO_NOTHING)