get ability estimation from source

app/lib/irt/models/rasch.py

@@ -0,0 +1,23 @@
+import numpy as np
+
+from girth import ability_mle
+
+class Rasch:
+
+  def __init__(self, model_params, kwargs):
+    self.model_params = model_params
+    self.b_param = kwargs['b_param']
+    self.e = 2.71828
+    self.theta = kwargs['theta']
+
+  def result(self):
+    return 0.0
+
+  @classmethod
+  def ability_estimate(self, items) -> float:
+    # we'll likely have to change this to something more robust
+    # when we get into more complex response types
+    responses = np.array([[int(item.response)] for item in items])
+    difficulty = np.array([item.b_param for item in items])
+    discrimination = np.linspace(1, 1, len(difficulty))
+    return ability_mle(responses, difficulty, discrimination)

app/lib/irt/models/three_parameter_logistic.py

@@ -16,3 +16,7 @@ class ThreeParameterLogistic:
         c = self.model_params.c_param
         return c + (1 - c) * (1 / (1 + self.e**(-a *
                                                 (self.theta - self.b_param))))
+
+    @classmethod
+    def ability_estimate(self) -> float:
+        return 0.0

app/models/ability_estimation.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel
+from typing import ClassVar, List
+
+from models.item import Item
+from lib.irt.models.three_parameter_logistic import ThreeParameterLogistic
+from lib.irt.models.rasch import Rasch
+
+class AbilityEstimation(BaseModel):
+  exam_id: int
+  items: List[Item] = []
+  irt_model: str
+  min_theta: float = -3.0
+  max_theta: float = 3.0
+
+  IRT_MODELS: ClassVar[dict] = {
+    'rasch': Rasch,
+    # not supported
+    # '3pl': ThreeParameterLogistic
+  }
+
+  def calculate(self) -> float:
+    model = self.IRT_MODELS[self.irt_model]
+    return model.ability_estimate(self.items)

app/models/item.py

@@ -11,8 +11,9 @@ class Item(BaseModel):
     position: Optional[int] = None
     passage_id: Optional[int] = None
     workflow_state: Optional[str] = None
-    attributes: List[Attribute]
+    attributes: List[Attribute] = None
     b_param: float = 0.00
+    response: Optional[str] = None
 
     def iif(self, solver_run, theta):
         return ItemInformationFunction(solver_run.irt_model).calculate(b_param=self.b_param, theta=theta)

app/services/ability_estimation_service.py

@@ -1,9 +1,12 @@
-import logging
-
 from services.base import Base
+from models.ability_estimation import AbilityEstimation
 
 class AbilityEstimationService(Base):
   ACTION = 'abilityEstimation'
 
   def process(self):
-    logging.info('Ability Estimation Service to be implemented...')
+    attributes = self.service_attributes()
+    ability_estimation = AbilityEstimation.parse_obj(attributes)
+    results = ability_estimation.calculate()
+
+    return None

app/services/base.py

@@ -1,5 +1,25 @@
+import logging, json, re
+
+from helpers import aws_helper, tar_helper
+
 class Base:
 
     def __init__(self, source, ingest_type='message'):
         self.ingest_type = ingest_type
         self.source = source
+
+    def service_attributes(self):
+        logging.info('Retrieving attributes from message...')
+        # get s3 object
+        self.key = aws_helper.get_key_from_message(self.source)
+        s3_object = aws_helper.get_object(
+            self.key, aws_helper.get_bucket_from_message(self.source))
+
+        # convert to tar
+        self.tar = tar_helper.raw_to_tar(s3_object)
+
+        # get attributes file and convert to dict
+        action_snake = re.sub(r'(?<!^)(?=[A-Z])', '_', self.ACTION).lower()
+        attributes = json.loads(
+            tar_helper.extract_file_from_tar(
+                self.tar, f'{action_snake}_attributes.json').read())