def _score_json()

in src/vw-serving/src/vw_serving/serve.py [0:0]

• 33 lines of code
• 5 McCabe index (conditional complexity)

def _score_json(model, observation, response_content_type=CONTENT_TYPE_JSON):
    event_id = uuid.uuid1().int
    dt = datetime.datetime.now()
    timestamp = int(dt.strftime("%s"))
    action_probs = model.predict(observation)
    nchoices = len(action_probs)
    action_probs = (action_probs / action_probs.sum())
    action = np.random.choice(nchoices, p=action_probs) + 1
    action_prob = action_probs[action - 1]
    # add sample_prob for later dataset sampling
    sample_prob = random.uniform(0.0, 1.0)
    if response_content_type in (CONTENT_TYPE_JSON, CONTENT_TYPE_JSONLINES):
        # convert to JSON
        response_payload = json.dumps({"action": action,
                                       "action_prob": action_prob,
                                       "event_id": event_id,
                                       "timestamp": timestamp,
                                       "sample_prob": sample_prob,
                                       "model_id": ScoringService._model_id})
    else:
        # convert to CSV
        response_payload = ",".join([
            str(x) for x in
            [action, action_prob, event_id, timestamp, sample_prob, ScoringService._model_id]
        ])
    blob_to_log = json.dumps({"action": action,
                              "action_prob": action_prob,
                              "event_id": event_id,
                              "observation": observation,
                              "timestamp": timestamp,
                              "model_id": ScoringService._model_id,
                              "sample_prob": sample_prob,
                              "type": "actions"})
    if ScoringService.LOG_INFERENCE_DATA:
        # TODO: Log state, action, eventID
        ScoringService._redis_client.publish(REDIS_PUBLISHER_CHANNEL, blob_to_log)
    return response_payload