better abstraction and error handling/logging

app/lib/irt/models/rasch.py

@@ -15,9 +15,18 @@ class Rasch:
 
   @classmethod
   def ability_estimate(self, items) -> float:
+    # responses are mapped into a matrix, where each row and item
+    # and each column is an exam form result
     # 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])
+    # the difficulty (b param) for each item is in an ordered list
     difficulty = np.array([item.b_param for item in items])
+    # the package currently utilizes a fixed a param (discrimination)
     discrimination = np.linspace(1, 1, len(difficulty))
-    return ability_mle(responses, difficulty, discrimination)
+    # there are many methodologies to calculate ability from a data set of responses
+    # this is what our client currently uses but we should expand this to allow for
+    # switching between methodologies when needed
+    # it also currrently only does a single ability estimation
+    # at some point we can also accommodate batch ability estimates if need be
+    return ability_mle(responses, difficulty, discrimination).tolist()[0]

app/main.py

@@ -43,7 +43,7 @@ class ServiceListener(Consumer):
             logging.error(f'action of type {action} does not exist.')
 
 def main():
-    logging.info('Starting IRT Service: That Was Rash (v1.4.1)...')
+    logging.info('Starting IRT Service: That Was Rasch (v1.5.0)...')
 
     # ToDo: Figure out a much better way of doing this.
     # LocalStack wants 'endpoint_url', while prod doesnt :(

app/models/ability_estimation.py

@@ -19,5 +19,9 @@ class AbilityEstimation(BaseModel):
   }
 
   def calculate(self) -> float:
-    model = self.IRT_MODELS[self.irt_model]
-    return model.ability_estimate(self.items)
+    if self.irt_model in self.IRT_MODELS:
+      model = self.IRT_MODELS[self.irt_model]
+      return model.ability_estimate(self.items)
+    else:
+      logging.error(f'model of type {self.irt_model} does not exist.')
+      return None

app/services/ability_estimation_service.py

@@ -11,12 +11,21 @@ class AbilityEstimationService(Base):
   def process(self):
     attributes = self.service_attributes()
     ability_estimation = AbilityEstimation.parse_obj(attributes)
-    # we need to convert the results from numpy array to a list to dump to json
-    json_data = json.dumps(ability_estimation.calculate().tolist())
+    result = ability_estimation.calculate()
 
-    # temp file name
-    self.file_name = f'{ability_estimation.exam_id}_ability_estimation_result.json'
+    if result is not None:
+      response = json.dumps({
+        'status': 'success',
+        'result': result
+      })
+
+      self.file_name = f'{ability_estimation.exam_id}_ability_estimation_result.json'
+    else:
+      response = json.dumps({
+        'status': 'error',
+        'result': None
+      })
 
     return aws_helper.file_stream_upload(
-            io.BytesIO(bytes(json_data.encode('UTF-8'))), self.file_name,
+            io.BytesIO(bytes(response.encode('UTF-8'))), self.file_name,
             ApplicationConfigs.s3_processed_bucket, self.ACTION)