Qualtrics integration preliminary

pyproject.toml

@@ -51,6 +51,7 @@ fastapi = "0.101"
 typing-extensions = "4.5.0"
 kaleido = "0.2.1"
 lida = "^0.0.11"
+qualtricsapi = "^0.6.1"
 
 [tool.poetry.group.dev.dependencies]
 bandit = "^1.7.1"

tests/devices/qualtrics/test_qualtrics.py

@@ -0,0 +1,21 @@
+import pytest
+import pandas as pd
+import wearipedia
+
+@pytest.mark.parametrize("real", [True, False])
+def test_qualtrics(real):
+
+    synthetic_survey = "your_survey_id_here"
+
+    device = wearipedia.get_device(
+        "qualtrics/qualtrics",
+        survey=synthetic_survey,
+    )
+
+    if real:
+        wearipedia._authenticate_device("qualtrics", device)
+
+    data = device.get_data('responses')
+
+    # Check that the data is a pandas DataFrame
+    assert isinstance(data, pd.DataFrame), f"Data is not a pandas DataFrame"

wearipedia/__init__.py

@@ -81,6 +81,7 @@ def get_all_device_names():
         "polar/vantage",
         "strava/strava",
         "myfitnesspal/myfitnesspal",
+        "qualtrics/qualtrics",
     ]
 
 

wearipedia/devices/qualtrics/__init__.py

@@ -0,0 +1 @@
+from .qualtrics import *

wearipedia/devices/qualtrics/qualtrics.py

@@ -0,0 +1,50 @@
+from ...utils import seed_everything
+from ..device import BaseDevice
+from .qualtrics_fetch import *
+from .qualtrics_gen import *
+
+from QualtricsAPI.Setup import Credentials
+
+class_name = "Qualtrics"
+
+class Qualtrics(BaseDevice):
+    def __init__(self, seed=0, survey="SV_6FDXhJO8mupUnuC"):
+        params = {
+            "seed": seed,
+            "survey": str(survey),
+        }
+
+        self._initialize_device_params(
+            [
+                "responses",
+            ],
+            params,
+            {
+                "seed": 0,
+                "synthetic_survey": "your_survey_id_here",
+            },
+        )
+
+    def _default_params(self):
+        return {
+            "survey": self.init_params["synthetic_survey"],
+        }
+
+    def _get_real(self, data_type, params):
+        return fetch_real_data(
+            params["survey"]
+        )
+
+    def _filter_synthetic(self, data, data_type, params):
+        return data
+
+    def _gen_synthetic(self):
+        seed_everything(self.init_params["seed"])
+        self.responses = create_syn_data(
+            self.init_params["synthetic_survey"],
+        )
+
+    def _authenticate(self, auth_creds):
+        token = auth_creds["token"]
+        data_center = auth_creds["data_center"]
+        Credentials().qualtrics_api_credentials(token=token, data_center=data_center)

wearipedia/devices/qualtrics/qualtrics_fetch.py

@@ -0,0 +1,6 @@
+from QualtricsAPI.Survey import Responses
+from QualtricsAPI.Setup import Credentials
+
+def fetch_real_data(survey):
+    responses = Responses().get_survey_responses(survey=survey, useLabels=True)
+    return responses

wearipedia/devices/qualtrics/qualtrics_gen.py

@@ -0,0 +1,8 @@
+import os
+import pandas as pd
+
+def create_syn_data(survey):
+    script_dir = os.path.dirname(__file__)
+    survey_file = os.path.join(script_dir, "sample_survey.csv")
+    df = pd.read_csv(survey_file)
+    return df