Go live version

packages/server/akello/db/models.py

@@ -2,7 +2,9 @@
 import json
 from typing import List, Optional
 from akello.db.connector.dynamodb import RegistryDBBaseModel
-from akello.db.types import Measurement, UserRole, FlagTypes, PatientStatysTypes, AkelloApp, TreatmentLog, EventLog, AuditLog
+from akello.db.types import Measurement, UserRole, FlagTypes, PatientStatysTypes, AkelloApp, TreatmentLog, EventLog, \
+    AuditLog
+
 
 class UserModel(RegistryDBBaseModel):
     """
@@ -69,9 +71,9 @@ class RegistryUser(RegistryDBBaseModel):
     registry_id: str
     user_id: str
     date_created: int = datetime.datetime.utcnow().timestamp()
-    first_name: str  #TODO: consider removing this
-    last_name: str   #TODO: consider removing this
-    email: str       #TODO: consider removing this
+    first_name: str  # TODO: consider removing this
+    last_name: str  # TODO: consider removing this
+    email: str  # TODO: consider removing this
     role: UserRole
     is_admin: bool = False
 
@@ -126,7 +128,6 @@ def sort_key(self) -> str:
         return 'metadata'
 
 
-
 class PatientRegistry(RegistryDBBaseModel):
     """
     Keeps a direct 1-1 mapping of a patient to a registry.

packages/server/akello/services/reports.py

@@ -1,30 +1,43 @@
-import datetime, random, uuid, json
-from decimal import Decimal
-from akello.db.models import RegistryModel, TreatmentLog, PatientRegistry
-from akello.db.types import ContactTypes
+import datetime, json, logging
+from akello.db.models import TreatmentLog, PatientRegistry
 from akello.services import BaseService
 from akello.services.registry import RegistryService
 from datetime import datetime
 
-
-import logging
 logger = logging.getLogger('mangum')
 
 
 class ReportsService(BaseService):
+    """
+    The ReportsService class is responsible for generating reports related to billing and registry dashboards
+    for a healthcare provider. It leverages static methods to access patient data and compute relevant statistics
+    for a specified period.
+    """
 
     @staticmethod
     def get_billing_report(registry_id, from_date, to_date):
+        """
+        Generates a billing report for a given registry within a specific date range.
+        The report includes patient treatment logs, summarized by minutes spent per month.
+
+        Parameters:
+        - registry_id: An identifier for the registry whose billing report is to be generated.
+        - from_date: The starting timestamp (inclusive) for filtering treatment logs.
+        - to_date: The ending timestamp (inclusive) for filtering treatment logs.
+
+        Returns:
+        - A list of dictionaries, each representing a billing entry for a patient including first name, last name,
+        medical record number (MRN), date, payer, referring provider NPI, and total minutes of treatment.
+        """
+
         logger.info('running report for registry: %s from: %s to: %s' % (registry_id, from_date, to_date))
         patients = RegistryService.get_patients(registry_id)
         patient_report = {
             'monthly': {
-
             },
             'patients': {
-
             }
-        }        
+        }
 
         for patient in patients:
             mrn = patient['patient_mrn']
@@ -35,24 +48,19 @@ def get_billing_report(registry_id, from_date, to_date):
                 }
 
             for treatment_log in patient['treatment_logs']:
-
                 t = TreatmentLog(**treatment_log) 
                 utc_dt = datetime.utcfromtimestamp(t.date / 1000)
                 if not (utc_dt.timestamp() >= from_date and utc_dt.timestamp() <= to_date):
                     continue
                 year = utc_dt.year
                 month = utc_dt.month
-                utc_dt_str = '%s-%s' % (year, month)                
-
+                utc_dt_str = '%s-%s' % (year, month)
                 if utc_dt_str not in patient_report['patients'][mrn]['minute_stats']:
-                    patient_report['patients'][mrn]['minute_stats'][utc_dt_str] = 0                
-
+                    patient_report['patients'][mrn]['minute_stats'][utc_dt_str] = 0
                 patient_report['patients'][mrn]['minute_stats'][utc_dt_str] += t.minutes                    
-
-
+
         # flatten the data
         r = []
-
         for mrn in patient_report['patients']:
             p = patient_report['patients'][mrn]['info']
             for s in patient_report['patients'][mrn]['minute_stats']:                
@@ -71,6 +79,20 @@ def get_billing_report(registry_id, from_date, to_date):
 
     @staticmethod
     def get_registry_dashboard(registry_id, from_date, to_date):
+        """
+        Generates a dashboard report for a given registry within a specific date range, including information
+        on payer distribution, patient statuses, and treatment performance metrics such as average, median,
+        and maximum treatment weeks.
+
+        Parameters:
+        - registry_id: An identifier for the registry whose dashboard report is to be generated.
+        - from_date: The starting timestamp (inclusive) for considering patient treatments.
+        - to_date: The ending timestamp (inclusive) for considering patient treatments.
+
+        Returns:
+        - A dictionary containing structured information about the treatment performance, screening scores,
+        payer distribution, and patient status distribution for the specified registry and date range.
+        """
 
         patients = RegistryService.get_patients(registry_id)
 

packages/server/akello/services/screeners.py

@@ -1,40 +1,22 @@
-import os
-import json
-import logging
-import yaml
-
+import os, logging, yaml
 from akello.services import BaseService
 from akello.db.models import Measurement
 
-
 logger = logging.getLogger('mangum')
 
 
 class ScreenerService(BaseService):
-
-
+
     @staticmethod
-    #def get_measurements():
     def get_screeners():
         measurement_list = []
-        measurements = os.listdir('akello/screeners/measurements')        
+        measurements = os.listdir('akello/screeners/measurements')
         for measurement in measurements:
-
             if measurement.endswith('.yaml'):
                 with open(f'akello/screeners/measurements/{measurement}') as f:
-
-                    try:                                                
+                    try:
                         mobj = Measurement(**yaml.safe_load(f))
-                        measurement_list.append(mobj)                                                 
+                        measurement_list.append(mobj)
                     except yaml.YAMLError as exc:
-                        print(exc)                    
-
+                        print(exc)
         return measurement_list
-
-    @staticmethod
-    def get_screener_by_id(screener_id):
-        pass
-
-    def generate_questionnaire_response(self, questionnaire_response):
-        # Example: https://build.fhir.org/ig/HL7/sdc/QuestionnaireResponse-questionnaireresponse-sdc-profile-example-PHQ9.json.html
-        pass

packages/server/akello/services/tests/test_user_service.py

@@ -4,13 +4,30 @@
 from akello.services.tests import mock_env_configs
 from akello.services.user import UserService
 
+
 @mock.patch.dict(os.environ, mock_env_configs)
 class TestUserService(TestCase):
 
-    @patch('akello.db.connector.dynamodb.registry_db.query')    
+    @patch('akello.db.connector.dynamodb.registry_db.query')
     def test_service_get_user(self, mock_query):
 
         db_data = [{'user-id': 1}]
         mock_query.return_value = {'Items': db_data}
         user_response = UserService.get_user('mock-user-id')
-        self.assertEqual(user_response, db_data)
+        self.assertEqual(user_response, db_data)
+
+    @patch('akello.db.connector.dynamodb.registry_db.query')
+    def test_registry_access_exists(self, mock_query):
+        db_data = [{'user-id': 1}]
+        mock_query.return_value = {'Items': db_data}
+        response = UserService.check_registry_access('mock-user-id', 'mock-registry-id')
+        self.assertEqual(response, {'user-id': 1})
+
+    @patch('akello.db.connector.dynamodb.registry_db.query')
+    def test_registry_access_does_not_exist(self, mock_query):
+        db_data = []
+        mock_query.return_value = {'Items': db_data}
+        try:
+            UserService.check_registry_access('mock-user-id', 'mock-registry-id')
+        except Exception as e:
+            self.assertEqual(str(e), 'The user is not authorized')