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,104 @@
+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):
+    """
+    A class representing a Qualtrics survey device.
+
+    This class is responsible for initializing the device with the necessary
+    parameters, generating synthetic data, and fetching real data from Qualtrics 
+    using the unofficial QualtricsAPI.
+    """
+    def __init__(self, seed=0, survey="your_survey_id_here"):
+        """
+        Initializes the Qualtrics device with the given seed and survey ID from Qualtrics.
+
+        Parameters:
+            seed (int, optional): The seed for random number generation. 
+            survey (str, optional): The ID of the survey to be used. Retrieve this from Qualtrics.
+        """
+        params = {
+            "seed": seed,
+            "survey": str(survey),
+        }
+
+        self._initialize_device_params(
+            [
+                "responses",
+            ],
+            params,
+            {
+                "seed": 0,
+                "synthetic_survey": "your_survey_id_here",
+            },
+        )
+
+    def _default_params(self):
+        """
+        Returns the default parameters for the Qualtrics device.
+
+        Returns:
+            dict: A dictionary containing the default parameters.
+        """
+        return {
+            "survey": self.init_params["synthetic_survey"],
+        }
+
+    def _get_real(self, data_type, params):
+        """
+        Fetches real data from Qualtrics based on the provided parameters.
+
+        Parameters:
+            data_type (str): The type of data to fetch.
+            params (dict): A dictionary containing the parameters for fetching data.
+
+        Returns:
+            dict: A dictionary containing the fetched real data.
+        """
+        return fetch_real_data(
+            params["survey"]
+        )
+
+    def _filter_synthetic(self, data, data_type, params):
+        """
+        Filters the synthetic data based on the provided parameters. This method is not implemented.
+
+        Parameters:
+            data (dict): The synthetic data to filter.
+            data_type (str): The type of data being filtered.
+            params (dict): A dictionary containing the parameters for filtering data.
+
+        Returns:
+            dict: The filtered synthetic data. This method returns the data as is.
+        """
+        return data
+
+    def _gen_synthetic(self):
+        """
+        Generates synthetic survey data based on the initial parameters.
+
+        This method sets the `responses` attribute with the generated synthetic data.
+        """
+        seed_everything(self.init_params["seed"])
+        self.responses = create_syn_data(
+            self.init_params["synthetic_survey"],
+        )
+
+    def _authenticate(self, auth_creds):
+        """
+        Authenticates the device using the provided credentials.
+
+        Parameters:
+            auth_creds (dict): A dictionary containing authentication credentials.
+                - token (str): The API token for authentication.
+                - data_center (str): The data center to connect to.
+        """
+        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,18 @@
+from QualtricsAPI.Survey import Responses
+from QualtricsAPI.Setup import Credentials
+
+def fetch_real_data(survey):
+    """
+    Fetches real survey data from Qualtrics.
+
+    This function uses QualtricsAPI to fetch survey responses based on the given survey ID.
+    The responses are retrieved with labels (actual user responses) instead of numeric codes.
+
+    Parameters:
+        survey (str): The ID of the survey from which to fetch responses.
+
+    Returns:
+        dict: A dictionary containing the survey responses.
+    """
+    responses = Responses().get_survey_responses(survey=survey, useLabels=True)
+    return responses

wearipedia/devices/qualtrics/qualtrics_gen.py

@@ -0,0 +1,21 @@
+import os
+import pandas as pd
+
+def create_syn_data(survey):
+    """
+    Creates synthetic survey data based on a sample survey CSV file.
+
+    This function reads a sample survey CSV file located in the same directory as the script
+    and returns its contents as a Pandas DataFrame. The function assumes that the sample survey
+    CSV file is named "sample_survey.csv".
+
+    Parameters:
+        survey (str): The ID of the survey to be used. This parameter is currently not used in the function.
+
+    Returns:
+        pd.DataFrame: A DataFrame containing the synthetic survey data from the sample CSV file.
+    """
+    script_dir = os.path.dirname(__file__)
+    survey_file = os.path.join(script_dir, "sample_survey.csv")
+    df = pd.read_csv(survey_file)
+    return df