Merge pull request #15 from UMassCDS/new-form-retrieval

New form retrieval

Author: ginic

CHANGELOG.md

@@ -7,9 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 You should also add project tags for each release in Github, see [Managing releases in a repository](https://docs.github.com/en/repositories/releasing-projects-on-github/managing-releases-in-a-repository).
 
 ## [Unreleased]
-
 ### Added
 - Merged the MSF-OCR-Streamlit repository into this repository
+- User authenticates with DHIS2 password rather than hard coded passkey
+- Table headers are corrected based on key-value pairs from DHIS2
+- Payload with key-value pairs for DHIS2 is displayed to the user before uploading
+- Images are resized before being sent to OpenAI 
+
+### Changed
+- User must confirm each page and key-value pairs before they're allowed to upload, so upload buttons is not initially selectable
 
 ## [1.1.0] - 2024-07-26
 ### Changed

app_llm.py

@@ -1,5 +1,5 @@
 import urllib.parse
-
+import json
 import requests
 
 # Make sure these are set before trying to make requests
@@ -10,16 +10,12 @@
 # TODO It might be clearer to create a Server object class and have this be the __init__() function
 def configure_DHIS2_server(username=None, password=None, server_url=None):
     global DHIS2_SERVER_URL, DHIS2_USERNAME, DHIS2_PASSWORD
-    DHIS2_USERNAME = username
-    DHIS2_PASSWORD = password
-    DHIS2_SERVER_URL = server_url
-
-
-def checkResponseStatus(res):
-    if res.status_code == 401:
-        raise ValueError("Authentication failed. Check your username and password.")
-    res.raise_for_status()
-
+    if username is not None: 
+        DHIS2_USERNAME = username
+    if password is not None: 
+        DHIS2_PASSWORD = password
+    if server_url is not None: 
+        DHIS2_SERVER_URL = server_url
 
 def getAllUIDs(item_type, search_items):
     encoded_search_items = [urllib.parse.quote_plus(item) for item in search_items]
@@ -30,10 +26,7 @@ def getAllUIDs(item_type, search_items):
         filter_param = 'filter=' + '&filter='.join([f'name:ilike:{term}' for term in encoded_search_items])
 
     url = f'{DHIS2_SERVER_URL}/api/{item_type}?{filter_param}'
-    response = requests.get(url, auth=(DHIS2_USERNAME, DHIS2_PASSWORD))
-    checkResponseStatus(response)
-
-    data = response.json()
+    data = getResponse(url)
     items = data[item_type]
     print(f"{len(data[item_type])} matches found for {search_items}")
     if len(items) > 0:
@@ -43,6 +36,15 @@ def getAllUIDs(item_type, search_items):
 
     return uid
 
+def getResponse(url):
+    response = requests.get(url, auth=(DHIS2_USERNAME, DHIS2_PASSWORD))
+
+    if response.status_code == 401:
+        raise ValueError("Authentication failed. Check your username and password.")
+    response.raise_for_status()
+
+    data = response.json()
+    return data
 
 def getOrgUnitChildren(uid):
     """
@@ -51,11 +53,7 @@ def getOrgUnitChildren(uid):
     :return: List of (org unit child name, org unit child data sets))
     """
     url = f'{DHIS2_SERVER_URL}/api/organisationUnits/{uid}?includeChildren=true'
-    
-    response = requests.get(url, auth=(DHIS2_USERNAME, DHIS2_PASSWORD))
-    checkResponseStatus(response)
-    
-    data = response.json()
+    data = getResponse(url)
     items = data['organisationUnits']
     children = [(item['name'], item['dataSets'], item['id']) for item in items if item['id'] != uid]
 
@@ -73,71 +71,67 @@ def getDataSets(data_sets_uids):
         uid = uid_obj['id']
         url = f'{DHIS2_SERVER_URL}/api/dataSets/{uid}'
 
-        response = requests.get(url, auth=(DHIS2_USERNAME, DHIS2_PASSWORD))
-        checkResponseStatus(response)
-        
-        data = response.json()
+        data = getResponse(url)
         data_set = (data['name'], data['id'], data['periodType'])
         data_sets.append(data_set)
 
     return data_sets
 
-def getCategoryUIDs(dataSet_uid):
+def getFormJson(dataSet_uid, period, orgUnit_uid):
     """
-    Hierarchically searches DHIS2 to generate category UIDs for each dataElement. Also used for retreiving all data elements and categories present in a dataset.
-    :param data_sets_uid: UID of the dataset
-    :return: dataElement_to_id (dict[str, str]), dataElement_to_categoryCombo (dict[str, str]), categoryCombos (dict[str, str]), category_list (list[str]), dataElement_list (list[str])
-             category list eg. ['0-11m','<5y'...]
+    Gets information about all forms associated with a organisation, dataset, period combination in DHIS2.
+    :param dataset UID, time period, organisation unit UID
+    :return json response containing hierarchical information about tabs, tables, non-tabular fields
     """
-    url = f'{DHIS2_SERVER_URL}/api/dataSets/{dataSet_uid}?fields=dataSetElements'
-
-    response = requests.get(url, auth=(DHIS2_USERNAME, DHIS2_PASSWORD))
-    checkResponseStatus(response)
 
-    data = response.json()
-
-    items = data['dataSetElements']
-
-    dataElement_to_categoryCombo = {}
-    categoryCombos = {}
-    categoryOptionCombos = {}
-    for item in items:
-        if 'categoryCombo' in item:
-            dataElement_to_categoryCombo[item['dataElement']['id']] = item['categoryCombo']['id']
-            categoryCombos[item['categoryCombo']['id']] = {}
- 
-    for catCombo_id in categoryCombos:
-        url = f'{DHIS2_SERVER_URL}/api/categoryCombos/{catCombo_id}?fields=categoryOptionCombos'
-
-        response = requests.get(url, auth=(DHIS2_USERNAME, DHIS2_PASSWORD))
-        checkResponseStatus(response)
-        
-        data = response.json()
-
-        items = data['categoryOptionCombos']
-
-        for item in items:
-            url = f"{DHIS2_SERVER_URL}/api/categoryOptionCombos/{item['id']}?fields=name"
-
-            response = requests.get(url, auth=(DHIS2_USERNAME, DHIS2_PASSWORD))
-            checkResponseStatus(response)
+    # POST empty data payload to trigger form generation
+    json_export = {}
+    json_export["dataSet"] = dataSet_uid
+    json_export["period"] = period
+    json_export["orgUnit"] = orgUnit_uid
+    json_export["dataValues"] = []
+    data_payload = json.dumps(json_export)
+    posturl = f'{DHIS2_SERVER_URL}/api/dataValueSets?dryRun=true' 
+
+    response = requests.post(
+                        posturl,
+                        auth=(DHIS2_USERNAME, DHIS2_PASSWORD),
+                        headers={'Content-Type': 'application/json'},
+                        data=data_payload
+                    )  
+    response.raise_for_status()
+
+    # Get form now
+    url = f'{DHIS2_SERVER_URL}/api/dataSets/{dataSet_uid}/form.json?pe={period}&ou={orgUnit_uid}'
+    data = getResponse(url)
+    return data
 
-            data = response.json()
-
-            categoryCombos[catCombo_id][data['name']] = item['id']
-
-            if data['name'] not in categoryOptionCombos:
-                categoryOptionCombos[data['name']] = ''
-    category_list = list(categoryOptionCombos.keys())
-                                    
-    url = f'{DHIS2_SERVER_URL}/api/dataElements?filter=dataSetElements.dataSet.id:eq:{dataSet_uid}&fields=id,formName'
-    response = requests.get(url, auth=(DHIS2_USERNAME, DHIS2_PASSWORD))
-    checkResponseStatus(response)
-    data = response.json()                   
-
-    dataElement_to_id = {item["formName"]:item["id"] for item in data['dataElements']}
-    dataElement_list = [item["formName"] for item in data['dataElements']]
+def get_DE_COC_List(form):
+    """
+    Finds the list of all dataElements (row names in tables) and categoryOptionCombos (column names in tables) within a DHIS2 form
+    :param json data containing hierarchical information about tabs, tables, non-tabular fields within a organisation, dataset, period combination in DHIS2. 
+    :return List of row names found, List of column names found 
+    """
+    url = f'{DHIS2_SERVER_URL}/api/dataElements?paging=false&fields=id,formName'
+    data = getResponse(url)
+    allDataElements = {item['id']:item['formName'] for item in data['dataElements'] if 'formName' in item and 'id' in item}
+
+    url = f'{DHIS2_SERVER_URL}/api/categoryOptionCombos?paging=false&fields=id,name'
+    data = getResponse(url)
+    allCategory = {item['id']:item['name'] for item in data['categoryOptionCombos'] if 'name' in item and 'id' in item}
+
+    # Form tabs found in DHIS2
+    tabs = form['groups']
+    dataElement_list = {}
+    categoryOptionCombo_list = {}
+    for tab in tabs:
+        for field in tab['fields']:
+            DE_ID = field['dataElement']
+            COC_ID = field['categoryOptionCombo']
+            if allDataElements[DE_ID] not in dataElement_list:
+                dataElement_list[allDataElements[DE_ID]] = 1
+            if allCategory[COC_ID] not in categoryOptionCombo_list:
+                categoryOptionCombo_list[allCategory[COC_ID]] = 1
+    return list(dataElement_list.keys()), list(categoryOptionCombo_list.keys())   
 
-    return dataElement_to_id, dataElement_to_categoryCombo, categoryCombos, category_list, dataElement_list       
-       
 

src/msfocr/data/dhis2.py

@@ -9,7 +9,6 @@
 import numpy as np
 import pandas as pd
 
-from msfocr.data import dhis2
 
 
 def letter_by_letter_similarity(text1, text2):
@@ -152,8 +151,7 @@ def get_sheet_type(res):
                     period.append(date)
     return [dataSet, orgUnit, sorted(period)]
 
-
-def generate_key_value_pairs(table, dataSet_uid):
+def generate_key_value_pairs(table, form):
     """
     Generates key-value pairs in the format required to upload data to DHIS2.
     {'dataElement': data_element_id,
@@ -162,36 +160,35 @@ def generate_key_value_pairs(table, dataSet_uid):
      UIDs like data_element_id, category_id are obtained by querying the DHIS2 metadata.
     :param table: DataFrame generated from table detection
     :return: List of key value pairs as shown above.
-    """
-    # Save UIDs found in a dictionary to avoid repeated UID querying
-    id_found = {}
-    
-    # Get dataElement to UID map for all dataElements in the dataset 
-    dataElement_to_id, dataElement_to_categoryCombo, categoryCombos_to_name_to_id,_,_ = dhis2.getCategoryUIDs(dataSet_uid)
-
+    """ 
     data_element_pairs = []
+
     # Iterate over each cell in the DataFrame
     table_array = table.values
     columns = table.columns
     for row_index in range(table_array.shape[0]):
+        # Row name in tally sheet
         data_element = table_array[row_index][0]
         for col_index in range(1, table_array.shape[1]):
+            # Column name in tally sheet
             category = columns[col_index]
             cell_value = table_array[row_index][col_index]
-            if cell_value is not None:
-                if data_element not in id_found:
-                    # Retrive UIDs for dataElement and categoryOption
-                    data_element_id = dataElement_to_id[data_element]
-                    id_found[data_element] = data_element_id
-                    print(data_element, data_element_id)
-                else:
-                    data_element_id = id_found[data_element]    
-
-                # Get category_UID for each dataElement
-                categoryCombo = dataElement_to_categoryCombo[data_element_id]
-                categoryOptionCombos = categoryCombos_to_name_to_id[categoryCombo]
-                category_id = categoryOptionCombos[category]
-
+            if cell_value is not None and cell_value!="-" and cell_value!="":
+                data_element_id = None
+                category_id = None
+                # Search for the string in the "label" field of form information
+                string_search = data_element + " " + category
+                for group in form['groups']:
+                    for field in group['fields']:
+                        if field['label']==string_search:
+                            data_element_id = field['dataElement']
+                            category_id = field['categoryOptionCombo']
+                
+                # The following exceptions will be raised if the row or column name in the tally sheet is different from the names used in metadata
+                # For eg. Pop1: Resident is called Population 1 in metadata
+                # If this exception is raised the only way forward is for the user to manually change the row/column name to the one used in metadata
+                if data_element_id is None or category_id is None:
+                    raise Exception(f"Unable to find {string_search} in DHIS2 metadata")
                 # Append to the list of data elements to be push to DHIS2
                 data_element_pairs.append(
                     {"dataElement": data_element_id,
@@ -201,7 +198,6 @@ def generate_key_value_pairs(table, dataSet_uid):
 
     return data_element_pairs
 
-
 # ocr_model = ocr_predictor(det_arch='db_resnet50', reco_arch='crnn_vgg16_bn', pretrained=True)
 # document = DocumentFile.from_images("IMG_20240514_090947.jpg")
 # result = get_word_level_content(ocr_model, document)

src/msfocr/doctr/ocr_functions.py

@@ -5,12 +5,12 @@
 import base64
 import json
 from concurrent.futures.thread import ThreadPoolExecutor
+from io import BytesIO
 
 import pandas as pd
 
 from openai import OpenAI
-from PIL import Image, ExifTags
-
+from PIL import Image, ExifTags, ImageOps
 
 def get_results(uploaded_image_paths):
     """
@@ -59,6 +59,30 @@ def parse_table_data(result):
     return table_names, dataframes
 
 
+def rescale_image(img, limit, maxi=True):
+    """Rescales an image file to GPT's proportions (Max 2048 x 768).
+
+    Args:
+        img (_Image_): The image file that needs to be rescaled.
+        limit (_int_): The maximum size of the dimension in pixels.
+        maxi (bool, optional): True for resizing the largest dimension, false for smallest. Defaults to True.
+
+    Returns:
+        _Image_: Resized image file.
+    """
+    width, height = img.size
+    if maxi:
+        max_dim = max(width, height)
+    else:
+        max_dim = min(width, height)
+    if max_dim > limit:
+        scale_factor = limit / max_dim
+        new_width = int(width * scale_factor)
+        new_height = int(height * scale_factor)
+        img = img.resize((new_width, new_height))
+    return img
+
+
 def encode_image(image_path):
     """
     Encodes an image file to base64 string.
@@ -70,7 +94,13 @@ def encode_image(image_path):
     :return: Base64 encoded string of the image.
     """
     image_path.seek(0)
-    return base64.b64encode(image_path.read()).decode("utf-8")
+    with Image.open(image_path) as img:
+        img = ImageOps.exif_transpose(img)
+        img = rescale_image(img, 2048, True)
+        img = rescale_image(img, 768, False)
+        buffered = BytesIO()
+        img.save(buffered, format="PNG")
+        return base64.b64encode(buffered.getvalue()).decode("utf-8")
 
 
 def extract_text_from_image(image_path):
@@ -139,21 +169,21 @@ def correct_image_orientation(image_path):
     :param image_path: The path to the image file.
     :return: PIL.Image.Image: The image with corrected orientation.
     """
-    image = Image.open(image_path)
-    orientation = None
-    try:
-        for orientation in ExifTags.TAGS.keys():
-            if ExifTags.TAGS[orientation] == 'Orientation':
-                break
-        exif = dict(image.getexif().items())
-        if exif.get(orientation) == 3:
-            image = image.rotate(180, expand=True)
-        elif exif.get(orientation) == 6:
-            image = image.rotate(270, expand=True)
-        elif exif.get(orientation) == 8:
-            image = image.rotate(90, expand=True)
-    except (AttributeError, KeyError, IndexError):
-        pass
-    return image
+    with Image.open(image_path) as image: 
+        orientation = None
+        try:
+            for orientation in ExifTags.TAGS.keys():
+                if ExifTags.TAGS[orientation] == 'Orientation':
+                    break
+            exif = dict(image.getexif().items())
+            if exif.get(orientation) == 3:
+                image = image.rotate(180, expand=True)
+            elif exif.get(orientation) == 6:
+                image = image.rotate(270, expand=True)
+            elif exif.get(orientation) == 8:
+                image = image.rotate(90, expand=True)
+        except (AttributeError, KeyError, IndexError):
+            pass
+        return image
 
 

src/msfocr/llm/ocr_functions.py

@@ -33,13 +33,10 @@ def test_generate_key_value_pairs(test_server_config, requests_mock):
         '5-14y': [None]
     })
 
-    requests_mock.get("http://test.com/api/dataSets/10?fields=dataSetElements", json={'dataSetElements' : [{"categoryCombo": {"id": 5},"dataElement": {"id": 1},"dataSet": {"id": 10}}, {"categoryCombo": {"id": 5},"dataElement": {"id": 3},"dataSet": {"id": 10}}]})
-    requests_mock.get("http://test.com/api/categoryCombos/5?fields=categoryOptionCombos", json={"categoryOptionCombos": [{"id": 8}, {"id": 9}]})
-    requests_mock.get("http://test.com/api/categoryOptionCombos/8?fields=name", json={"name": "0-11m"})
-    requests_mock.get("http://test.com/api/categoryOptionCombos/9?fields=name", json={"name": "12-59m"})
-    requests_mock.get("http://test.com/api/dataElements?filter=dataSetElements.dataSet.id:eq:10&fields=id,formName", json={'dataElements': [{'formName': 'BCG','id':1},{'formName': 'Polio (OPV) 1 (from 6 wks)','id':3}]})
-
-    assert len(ocr_functions.generate_key_value_pairs(df, 10)) == 0
+    assert len(ocr_functions.generate_key_value_pairs(df, {'groups': [{'fields':[{"label": "Paed (0-59m) vacc target population 0-11m",
+                    "dataElement": "paedid",
+                    "categoryOptionCombo": "0to11mid",
+                    "type": "INTEGER_POSITIVE"}]}]})) == 0
 
     df = pd.DataFrame({
         '0': ['BCG', 'Polio (OPV) 0 (birth dose)', 'Polio (OPV) 1 (from 6 wks)'],
@@ -48,16 +45,24 @@ def test_generate_key_value_pairs(test_server_config, requests_mock):
         '5-14y': [None, None, None]
     })
 
-    requests_mock.get("http://test.com/api/dataElements?filter=formName:ilike:BCG", json={"dataElements":[{"id": 1, "displayName": "AVAC_002 BCG"}]})
-    requests_mock.get("http://test.com/api/categoryOptions?filter=name:ilike:0-11m", json={'categoryOptions': [{'id': 2, 'displayName': '0-11m'}]})
-    requests_mock.get("http://test.com/api/dataElements?filter=formName:ilike:Polio (OPV) 1 (from 6 wks)", json={'dataElements': [{'id': 3, 'displayName': 'AVAC_006 Polio (OPV) 1 (from 6 wks)'}]})
-    requests_mock.get("http://test.com/api/categoryOptions?filter=name:ilike:12-59m", json={'categoryOptions': [{'id': 4, 'displayName': '12-59m'}]})
-
-    answer = [{'dataElement': '', 'categoryOptions': '', 'value': '45+29'},
-              {'dataElement': '', 'categoryOptions': '', 'value': '30+18'},
-              {'dataElement': '', 'categoryOptions': '', 'value': '55+29'}]
-
-    data_element_pairs = ocr_functions.generate_key_value_pairs(df, 10)
+    answer = [{'dataElement': 'bcgid', 'categoryOptions': '0to11mid', 'value': '45+29'},
+              {'dataElement': 'polioid', 'categoryOptions': '0to11mid', 'value': '30+18'},
+              {'dataElement': 'polioid', 'categoryOptions': '5to14yid', 'value': '55+29'}]
+
+    data_element_pairs = ocr_functions.generate_key_value_pairs(df, 
+                    {'groups': [{'fields':[{"label": "BCG 0-11m",
+                    "dataElement": "bcgid",
+                    "categoryOptionCombo": "0to11mid",
+                    "type": "INTEGER_POSITIVE"}]},
+                    {'fields':[{"label": "Polio (OPV) 1 (from 6 wks) 0-11m",
+                    "dataElement": "polioid",
+                    "categoryOptionCombo": "0to11mid",
+                    "type": "INTEGER_POSITIVE"}]},
+                    {'fields':[{"label": "Polio (OPV) 1 (from 6 wks) 12-59m",
+                    "dataElement": "polioid",
+                    "categoryOptionCombo": "5to14yid",
+                    "type": "INTEGER_POSITIVE"}]}]})
+    
     assert len(data_element_pairs) == len(answer)
 
     for i in range(len(data_element_pairs)):