Added "Maximum holding time" setting to services and analyses (#2624)

* Added analytical holding time functionality

* Typo

* Redux

* Trigger recalculate_records on sample date change

* do not display the lock button if beyond holding time

* Remove unnecessary bits

Surprisingly, this snippet is not necessary because when a sample
is created via add form, the uids of services (those assigned to
profiles included) are passed on submit

* Add warning icon when analysis is beyond holding time

* Remove unnecessary bits

* Clean import

* Changelog

* Improve texts

* Better wording

* CHangelog

* better spelling

* Fix TypeError: can't compare datetime.datetime to str

* Added timezone parameter to `to_ansi` dtime function

* Add now() function in api.dtime

* Fix doctest

* Fix inconsistency with Etc/GMT timezone when datetime is used

* Rely on to_ansi with timezone for comparisons

* Fix UnknownTimeZoneError when extracting the timezone from datetime

```
Traceback (innermost last):
  Module ZPublisher.WSGIPublisher, line 176, in transaction_pubevents
  Module ZPublisher.WSGIPublisher, line 385, in publish_module
  Module ZPublisher.WSGIPublisher, line 288, in publish
  Module ZPublisher.mapply, line 85, in mapply
  Module ZPublisher.WSGIPublisher, line 63, in call_object
  Module senaite.app.listing.view, line 246, in __call__
  Module senaite.app.listing.ajax, line 113, in handle_subpath
  Module senaite.core.decorators, line 40, in decorator
  Module senaite.app.listing.decorators, line 63, in wrapper
  Module senaite.app.listing.decorators, line 50, in wrapper
  Module senaite.app.listing.decorators, line 100, in wrapper
  Module senaite.app.listing.ajax, line 383, in ajax_folderitems
  Module senaite.app.listing.decorators, line 88, in wrapper
  Module senaite.app.listing.ajax, line 259, in get_folderitems
  Module bika.lims.browser.analyses.view, line 808, in folderitems
  Module senaite.app.listing.view, line 982, in folderitems
  Module bika.lims.browser.analyses.view, line 797, in folderitem
  Module bika.lims.browser.analyses.view, line 1737, in _folder_item_holding_time
  Module senaite.core.api.dtime, line 262, in to_ansi
  Module senaite.core.api.dtime, line 382, in to_zone
  Module pytz, line 188, in timezone
UnknownTimeZoneError: '+02'
```

* Do not raise error if timezone is not valid in to_ansi

* Better wording

* Proper comparison of dates in sample add form

---------

Co-authored-by: Ramon Bartl <rb@ridingbytes.com>

CHANGES.rst

@@ -4,6 +4,7 @@ Changelog
 2.6.0 (unreleased)
 ------------------
 
+- #2624 Added "Maximum holding time" setting to services and analyses
 - #2637 Do not remove inactive services from profiles and templates
 - #2642 Fix Attribute Error in Upgrade Step 2619
 - #2641 Fix AttributeError on rejection of samples without a contact set

src/bika/lims/browser/analyses/view.py

@@ -23,6 +23,7 @@
 from copy import copy
 from copy import deepcopy
 from datetime import datetime
+from datetime import timedelta
 from operator import itemgetter
 
 from bika.lims import api
@@ -38,9 +39,11 @@
 from bika.lims.interfaces import IFieldIcons
 from bika.lims.interfaces import IReferenceAnalysis
 from bika.lims.interfaces import IRoutineAnalysis
+from bika.lims.interfaces import ISubmitted
 from bika.lims.utils import check_permission
 from bika.lims.utils import format_supsub
 from bika.lims.utils import formatDecimalMark
+from bika.lims.utils import get_fas_ico
 from bika.lims.utils import get_image
 from bika.lims.utils import get_link
 from bika.lims.utils import get_link_for
@@ -790,6 +793,8 @@ def folderitem(self, obj, item, index):
         self._folder_item_remarks(obj, item)
         # Renders the analysis conditions
         self._folder_item_conditions(obj, item)
+        # Fill maximum holding time warnings
+        self._folder_item_holding_time(obj, item)
 
         return item
 
@@ -1698,6 +1703,70 @@ def to_str(condition):
         service = item["replace"].get("Service") or item["Service"]
         item["replace"]["Service"] = "<br/>".join([service, conditions])
 
+    def _folder_item_holding_time(self, analysis_brain, item):
+        """Adds an icon to the item dictionary if no result has been submitted
+        for the analysis and the holding time has passed or is about to expire.
+        It also displays the icon if the result was recorded after the holding
+        time limit.
+        """
+        analysis = self.get_object(analysis_brain)
+        if not IRoutineAnalysis.providedBy(analysis):
+            return
+
+        # get the maximum holding time for this analysis
+        max_holding_time = analysis.getMaxHoldingTime()
+        if not max_holding_time:
+            return
+
+        # get the datetime from which the max holding time is computed
+        start_date = analysis.getDateSampled()
+        start_date = dtime.to_dt(start_date)
+        if not start_date:
+            return
+
+        # get the timezone of the start date for correct comparisons
+        timezone = dtime.get_timezone(start_date)
+
+        # calculate the maximum holding date
+        delta = timedelta(minutes=api.to_minutes(**max_holding_time))
+        max_holding_date = dtime.to_ansi(start_date + delta)
+
+        # maybe the result was captured past the holding time
+        if ISubmitted.providedBy(analysis):
+            captured = analysis.getResultCaptureDate()
+            captured = dtime.to_ansi(captured, timezone=timezone)
+            if captured > max_holding_date:
+                msg = _("The result was captured past the holding time limit.")
+                icon = get_fas_ico("exclamation-triangle",
+                                   css_class="text-danger",
+                                   title=t(msg))
+                self._append_html_element(item, "ResultCaptureDate", icon)
+            return
+
+        # not yet submitted, maybe the holding time expired
+        now = dtime.to_ansi(dtime.now(), timezone=timezone)
+        if now > max_holding_date:
+            msg = _("The holding time for this sample and analysis has "
+                    "expired. Proceeding with the analysis may compromise the "
+                    "reliability of the results.")
+            icon = get_fas_ico("exclamation-triangle",
+                               css_class="text-danger",
+                               title=t(msg))
+            self._append_html_element(item, "ResultCaptureDate", icon)
+            return
+
+        # or maybe is about to expire
+        soon = dtime.to_ansi(dtime.now(), timezone=timezone)
+        if soon > max_holding_date:
+            msg = _("The holding time for this sample and analysis is about "
+                    "to expire. Please complete the analysis as soon as "
+                    "possible to ensure data accuracy and reliability.")
+            icon = get_fas_ico("exclamation-triangle",
+                               css_class="text-warning",
+                               title=t(msg))
+            self._append_html_element(item, "ResultCaptureDate", icon)
+            return
+
     def is_method_required(self, analysis):
         """Returns whether the render of the selection list with methods is
         required for the method passed-in, even if only option "None" is

src/bika/lims/browser/analysisrequest/add2.py

@@ -21,6 +21,7 @@
 import json
 from collections import OrderedDict
 from datetime import datetime
+from datetime import timedelta
 
 import six
 import transaction
@@ -50,7 +51,9 @@
 from Products.CMFPlone.utils import safe_unicode
 from Products.Five.browser import BrowserView
 from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile
+from senaite.core.api import dtime
 from senaite.core.catalog import CONTACT_CATALOG
+from senaite.core.catalog import SETUP_CATALOG
 from senaite.core.p3compat import cmp
 from senaite.core.permissions import TransitionMultiResults
 from zope.annotation.interfaces import IAnnotations
@@ -908,6 +911,7 @@ def get_service_info(self, obj):
             "category": obj.getCategoryTitle(),
             "poc": obj.getPointOfCapture(),
             "conditions": self.get_conditions_info(obj),
+            "max_holding_time": obj.getMaxHoldingTime(),
         })
 
         dependencies = get_calculation_dependencies_for(obj).values()
@@ -1217,11 +1221,75 @@ def ajax_recalculate_records(self):
             dependencies = self.get_unmet_dependencies_info(metadata)
             metadata.update(dependencies)
 
+            # services conducted beyond the holding time limit
+            beyond = self.get_services_beyond_holding_time(record)
+            metadata["beyond_holding_time"] = beyond
+
             # Set the metadata for current sample number (column)
             out[num_sample] = metadata
 
         return out
 
+    @viewcache.memoize
+    def get_services_max_holding_time(self):
+        """Returns a dict where the key is the uid of active services and the
+        value is a dict representing the maximum holding time in days, hours
+        and minutes. The dictionary only contains uids for services that have
+        a valid maximum holding time set
+        """
+        services = {}
+        query = {
+            "portal_type": "AnalysisService",
+            "point_of_capture": "lab",
+            "is_active": True,
+        }
+        brains = api.search(query, SETUP_CATALOG)
+        for brain in brains:
+            obj = api.get_object(brain)
+            max_holding_time = obj.getMaxHoldingTime()
+            if max_holding_time:
+                uid = api.get_uid(brain)
+                services[uid] = max_holding_time.copy()
+
+        return services
+
+    def get_services_beyond_holding_time(self, record):
+        """Return a list with the uids of the services that cannot be selected
+        because would be conducted past the holding time limit
+        """
+        # get the date to start count from
+        start_date = self.get_start_holding_date(record)
+        if not start_date:
+            return []
+
+        # get the timezone of the start date for correct comparisons
+        tz = dtime.get_timezone(start_date)
+
+        uids = []
+
+        # get the max holding times grouped by service uid
+        services = self.get_services_max_holding_time()
+        for uid, max_holding_time in services.items():
+
+            # calculate the maximum holding date
+            delta = timedelta(minutes=api.to_minutes(**max_holding_time))
+            max_holding_date = start_date + delta
+
+            # TypeError: can't compare offset-naive and offset-aware datetimes
+            max_date = dtime.to_ansi(max_holding_date)
+            now = dtime.to_ansi(dtime.now(), timezone=tz)
+            if now > max_date:
+                uids.append(uid)
+
+        return uids
+
+    def get_start_holding_date(self, record):
+        """Returns the datetime used to calculate the holding time limit,
+        typically the sample collection date.
+        """
+        sampled = record.get("DateSampled")
+        return dtime.to_dt(sampled)
+
     def get_record_metadata(self, record):
         """Returns the metadata for the record passed in
         """

src/bika/lims/browser/analysisrequest/templates/ar_add2.pt

@@ -542,6 +542,16 @@
                             🔒
                           </div>
 
+                          <!-- Beyond holding time box -->
+                          <div tal:attributes="uid python:service_uid;
+                                               arnum python:arnum;
+                                               id python:'{}-{}-beyondholdingtime'.format(service_uid, arnum);
+                                               class python:'service-beyondholdingtime {}-beyondholdingtime'.format(service_uid);"
+                               title="This service cannot be selected to prevent the analysis from being conducted beyond the analytical holding time."
+                               i18n:attributes="title">
+                            &#128683;
+                          </div>
+
                           <!-- Service checkbox -->
                           <div tal:attributes="id python:'{}-{}-analysisservice'.format(service_uid, arnum);
                                                class python:'analysisservice {}-analysisservice'.format(service_uid);">

src/bika/lims/content/abstractbaseanalysis.py

@@ -19,6 +19,7 @@
 # Some rights reserved, see README and LICENSE.
 
 from AccessControl import ClassSecurityInfo
+from bika.lims import api
 from bika.lims import bikaMessageFactory as _
 from bika.lims.browser.fields import DurationField
 from bika.lims.browser.fields import UIDReferenceField
@@ -365,6 +366,30 @@
     )
 )
 
+MaxHoldingTime = DurationField(
+    'MaxHoldingTime',
+    schemata="Analysis",
+    widget=DurationWidget(
+        label=_(
+            u"label_analysis_maxholdingtime",
+            default=u"Maximum holding time"
+        ),
+        description=_(
+            u"description_analysis_maxholdingtime",
+            default=u"This service will not appear for selection on the "
+                    u"sample registration form if the elapsed time since "
+                    u"sample collection exceeds the holding time limit. "
+                    u"Exceeding this time limit may result in unreliable or "
+                    u"compromised data, as the integrity of the sample can "
+                    u"degrade over time. Consequently, any results obtained "
+                    u"after this period may not accurately reflect the "
+                    u"sample's true composition, impacting data validity. "
+                    u"Note: This setting does not affect the test's "
+                    u"availability in the 'Manage Analyses' view."
+        )
+    )
+)
+
 # The amount of difference allowed between this analysis, and any duplicates.
 DuplicateVariation = FixedPointField(
     'DuplicateVariation',
@@ -789,6 +814,7 @@
     Instrument,
     Method,
     MaxTimeAllowed,
+    MaxHoldingTime,
     DuplicateVariation,
     Accredited,
     PointOfCapture,
@@ -1079,6 +1105,20 @@ def getMaxTimeAllowed(self):
         tat = self.Schema().getField("MaxTimeAllowed").get(self)
         return tat or self.bika_setup.getDefaultTurnaroundTime()
 
+    @security.public
+    def getMaxHoldingTime(self):
+        """Returns the maximum time since it the sample was collected for this
+        test/service to become available on sample creation. Returns None if no
+        positive maximum hold time is set. Otherwise, returns a dict with the
+        keys "days", "hours" and "minutes"
+        """
+        max_hold_time = self.Schema().getField("MaxHoldingTime").get(self)
+        if not max_hold_time:
+            return {}
+        if api.to_minutes(**max_hold_time) <= 0:
+            return {}
+        return max_hold_time
+
     # TODO Remove. ResultOptionsType field was replaced by ResulType field
     def getResultOptionsType(self):
         if self.getStringResult():

src/bika/lims/utils/__init__.py

@@ -915,3 +915,13 @@ def get_client(obj):
             return obj.getClient()
 
     return None
+
+
+def get_fas_ico(icon_id, **attrs):
+    """Returns a well-formed fontawesome (fas) icon
+    """
+    css_class = attrs.pop("css_class", "")
+    if not icon_id.startswith("fa-"):
+        icon_id = "fa-%s" % icon_id
+    attrs["css_class"] = " ".join([css_class, "fas", icon_id]).strip()
+    return "<i %s/>" % render_html_attributes(**attrs)