Merge 42b968d into 5258b7c

src/AkvoResponseGrouper/tests/test_lib_02_utils.py

@@ -1,11 +1,13 @@
 import os
+import pandas as pd
 from ..utils import (
     group_by_category_output,
     get_intersection,
     generate_data_as_json_file,
     get_valid_list,
     flatten_list,
     validate_number,
+    get_counted_category,
 )
 
 """
@@ -110,3 +112,28 @@ def test_valid_number():
     opt5 = {"number": {"less_than_equal": 10}}
     res5 = validate_number(q=opt5, answer=answer)
     assert res5 is True
+
+
+def test_get_counted_category():
+    categories = [
+        {
+            "id": 1,
+            "data": 1,
+            "form": 1,
+            "name": "Water",
+            "category": "Limited",
+        },
+        {
+            "id": 1,
+            "data": 2,
+            "form": 1,
+            "name": "Water",
+            "category": "Basic",
+        },
+    ]
+    df = pd.DataFrame(categories)
+    grouped = get_counted_category(df=df)
+    assert grouped == [
+        {"category": "Basic", "count": 1, "form": 1, "name": "Water"},
+        {"category": "Limited", "count": 1, "form": 1, "name": "Water"},
+    ]

src/AkvoResponseGrouper/utils.py

@@ -1,4 +1,5 @@
 import json
+import pandas as pd
 from itertools import groupby
 
 
@@ -17,7 +18,7 @@ def group_by_category_output(data):
                 for o in list(value)
             ],
         }
-        for key, value in groupby(g, key=lambda x:(x['name'], x['form']))
+        for key, value in groupby(g, key=lambda x: (x["name"], x["form"]))
     ]
     return res
 
@@ -110,3 +111,53 @@ def get_valid_list(opt, c, category):
     if sorted(valid) == sorted(validator):
         category = c["name"]
     return category
+
+
+def get_category(opt: dict, file_path: str = "./.category.json"):
+    with open(f"{file_path}") as config_file:
+        configs = json.load(config_file)
+    category = False
+    for config in configs:
+        for c in config["categories"]:
+            category = get_valid_list(opt, c, category)
+    return category
+
+
+def transform_categories_to_df(
+    categories: list, file_path: str = "./.category.json"
+):
+    df = pd.DataFrame(categories)
+    results = df.to_dict("records")
+    for d in results:
+        d.update({"category": get_category(opt=d["opt"], file_path=file_path)})
+    res = pd.DataFrame(results)
+    if list(res) != ["id", "data", "form", "name", "opt", "category"]:
+        return pd.DataFrame(
+            columns=[
+                "id",
+                "data",
+                "form",
+                "name",
+                "category",
+            ]
+        )
+    res = pd.concat(
+        [res.drop("opt", axis=1), pd.DataFrame(df["opt"].tolist())], axis=1
+    )
+    return res[
+        [
+            "id",
+            "data",
+            "form",
+            "name",
+            "category",
+        ]
+    ]
+
+
+def get_counted_category(df):
+    return (
+        df.groupby(["name", "category", "form"])["category"]
+        .agg("count")
+        .reset_index(name="count")
+    ).to_dict("records")

src/AkvoResponseGrouper/views.py

@@ -1,20 +1,7 @@
-import json
-import pandas as pd
 from typing import List, Optional
 from sqlalchemy.orm import Session
 from .models import Category, CategoryDict
-from .utils import get_valid_list
-
-
-def get_category(opt: dict):
-    file_path = "./.category.json"
-    with open(f"{file_path}") as config_file:
-        configs = json.load(config_file)
-    category = False
-    for config in configs:
-        for c in config["categories"]:
-            category = get_valid_list(opt, c, category)
-    return category
+from .utils import transform_categories_to_df, get_counted_category
 
 
 def get_data_categories(session: Session):
@@ -25,33 +12,7 @@ def get_data_categories(session: Session):
 
 def get_results(session: Session):
     categories = get_data_categories(session=session)
-    df = pd.DataFrame(categories)
-    results = df.to_dict("records")
-    for d in results:
-        d.update({"category": get_category(d["opt"])})
-    res = pd.DataFrame(results)
-    if list(res) != ["id", "data", "form", "name", "opt", "category"]:
-        return pd.DataFrame(
-            columns=[
-                "id",
-                "data",
-                "form",
-                "name",
-                "category",
-            ]
-        )
-    res = pd.concat(
-        [res.drop("opt", axis=1), pd.DataFrame(df["opt"].tolist())], axis=1
-    )
-    return res[
-        [
-            "id",
-            "data",
-            "form",
-            "name",
-            "category",
-        ]
-    ]
+    return transform_categories_to_df(categories=categories)
 
 
 def get_categories(
@@ -92,12 +53,7 @@ def get_group_by_category(
     if len(queries):
         queries = " & ".join(queries)
         res = res.query(queries)
-    res = (
-        res.groupby(["name", "category", "form"])["category"]
-        .agg("count")
-        .reset_index(name="count")
-    )
-    return res.to_dict("records")
+    return get_counted_category(df=res)
 
 
 def refresh_view(session: Session):

tests/test_category_data_frame.py

@@ -0,0 +1,56 @@
+import unittest
+from AkvoResponseGrouper.utils import (
+    generate_data_as_json_file,
+    transform_categories_to_df,
+)
+
+
+class TestCategoryDataFrame(unittest.TestCase):
+    def test_df_to_get_limited_category(self):
+        category = [
+            {
+                "name": "Water",
+                "form": 1,
+                "categories": [
+                    {
+                        "name": "Basic",
+                        "questions": [
+                            {
+                                "id": 567800081,
+                                "text": "Saperate Toilet",
+                                "options": ["Yes"],
+                                "else": {"name": "Limited"},
+                            }
+                        ],
+                    }
+                ],
+            }
+        ]
+        fc = generate_data_as_json_file(data=category)
+        categories = [
+            {
+                "id": 1,
+                "data": 1,
+                "form": 1,
+                "name": "Water",
+                "opt": {
+                    "567800081": ["No"],
+                },
+            }
+        ]
+        records = transform_categories_to_df(
+            categories=categories, file_path=fc
+        ).to_dict("records")
+        assert records == [
+            {
+                "id": 1,
+                "data": 1,
+                "form": 1,
+                "name": "Water",
+                "category": "Limited",
+            }
+        ]
+
+
+if __name__ == "__main__":
+    unittest.main()