Nutrition AI Implementation Guide

import os
import requests
import base64
import hmac
import hashlib
import json

SPIKE_API_BASE_URL = os.getenv("SPIKE_API_BASE_URL")
SPIKE_APP_HMAC_KEY = os.getenv("SPIKE_APP_HMAC_KEY")

try:
    SPIKE_APP_ID = int(os.getenv("SPIKE_APP_ID"))
except (ValueError, TypeError):
    raise ValueError("SPIKE_APP_ID must be an integer.")

# Implementation depends on your authentication method
# See: /api-docs/authentication for complete details
class SpikeAuth:
    def __init__(self, end_user_id):
        self.user_id = end_user_id
        self.app_id = SPIKE_APP_ID
        self.hmac_key = SPIKE_APP_HMAC_KEY
        self.base_url = SPIKE_API_BASE_URL
        self.hmac_signature = self._generate_hmac_signature()
        self.access_token = None

    def _generate_hmac_signature(self):
        h = hmac.new(self.hmac_key.encode('utf-8'), self.user_id.encode('utf-8'), hashlib.sha256)
        return h.hexdigest()


    def get_bearer_token(self) -> str:
        if self.access_token is not None:
            return self.access_token
        else:
            body = {
                "application_id": self.app_id,
                "application_user_id": self.user_id,
                "signature": self.hmac_signature
            }

            response = requests.post(
                f"{self.base_url}/auth/hmac",
                json=body,
                headers={"Content-Type": "application/json", "Accept": "application/json"},
                timeout=20
            )

            if response.status_code == 200:
                try:
                    print(f"Successfully obtained Bearer token for {self.user_id}")
                    self.access_token = response.json()["access_token"]
                    return self.access_token
                except (KeyError, json.JSONDecodeError) as e:
                    print(f"Failed to parse access token from successful response for {self.user_id}: {e}")
                raise requests.RequestException(f"Token parsing failed for {self.user_id}")
            else:
                raise Exception(f"Authentication failed for {self.user_id} with status {response.status_code}")


class NutritionAPI:
    def __init__(self, auth : SpikeAuth):
        self.spike_auth = auth


    def _create_auth_headers(self, method="GET") -> dict[str, str]:
        """Create authentication headers - see authentication docs for details"""
        try:
            token = self.spike_auth.get_bearer_token()
            req_headers = {
                "Authorization": f"Bearer {token}",
                "Accept": "application/json"
            }
            if method == "POST":
                req_headers["Content-Type"] = "application/json"
            return req_headers
        except requests.RequestException as auth_error:
            print(f"Failed to get bearer token for Nutrition API request header: {auth_error}")
        raise


    def analyze_food_image(self, image_path, **options):
        """Upload a food image for nutritional analysis"""
        # Read and encode image
        try:
            with open(image_path, 'rb') as image_file:
                image_data = base64.b64encode(image_file.read()).decode('utf-8')
        except FileNotFoundError:
            print(f"Image file not found: {image_path}")
            return None

        # Prepare the request body
        body = {
            "body": image_data,
            "analysis_mode": options.get("analysis_mode", "precise"),
            "country_code": options.get("country_code", "us"),
            "language_code": options.get("language_code", "en"),
            "include_ingredients": options.get("include_ingredients", False),
            "include_nutri_score": options.get("include_nutri_score", True),
            "include_dish_description": options.get("include_dish_description", True),
            "ignore_cache": options.get("ignore_cache", False),
            "include_nutrition_fields": options.get("include_nutrition_fields", [
                "energy_kcal", "protein_g", "fat_total_g", "carbohydrate_g",
                "fiber_total_dietary_g", "sodium_mg"
            ]),
            "wait_on_process": options.get("wait_on_process", False)
        }

        # Create request body and headers
        body_json = json.dumps(body)

        headers = self._create_auth_headers(method="POST")

        # Make request
        response = requests.post(
            f"{SPIKE_API_BASE_URL}/nutrition_records",
            headers=headers,
            data=body_json
        )
        if response.status_code == 200:
            return response.json()

        print(f"Response status: {response.status_code}")
        if response.status_code == 400:
            print(f"Bad request. Response: {response.text}")
        elif response.status_code == 401:
            print(f"Unauthorized for Nutrition API. Token expired or invalid. Response: {response.text}")
            self.spike_auth.access_token = None
        elif response.status_code == 404:
            print(f"Resource not found")
        elif response.status_code == 413:
            print(f"Encoded picture exceeds 10MB. Response: {response.text}")
        elif response.status_code == 422:
            print(f"Invalid parameters. Response: {response.text}")
        elif response.status_code == 500:
            print(f"Internal server error - contact Spike support. Response: {response.text}")
        return None

def print_result(res):
    if res is not None:
        print(f"Status: {res['status']}. Record ID: {res['record_id']}")
        if res['status'] == 'failed':
            print(f"Error: {res['failure_reason']}")
        elif res['status'] == 'completed':
            print(f"Dish: {res['dish_name']}")
            if 'dish_name_translated' in res.keys():
                print(f"Dish translated: {res['dish_name_translated']}")
            if 'dish_description' in res.keys():
                print(f"Description: {res['dish_description']}")
            if 'dish_description_translated' in res.keys():
                print(f"Description translated: {res['dish_description_translated']}")
            print(f"Serving size: {res['serving_size']} {res['unit']} ")
            if 'nutri_score' in res.keys():
                print(f"Nutri-Score: {res['nutri_score']}")
            if 'nutritional_fields' in res.keys():
                print("Nutritional fields:")
                for field in res['nutritional_fields']:
                    print(f"- {field}: {res['nutritional_fields'].get(field)}")
            if 'ingredients' in res.keys():
                print("Ingredients:")
                for ingredient in res.get('ingredients'):
                    print(f"- {ingredient['name']}: {ingredient['serving_size']}{ingredient['unit']}")
                    if 'nutritional_fields' in ingredient.keys():
                        for field in ingredient['nutritional_fields']:
                            print(f"  + {field}: {ingredient['nutritional_fields'].get(field)}")

# Example usage
user_id = "nutrition_test_user"
spike_auth = SpikeAuth(user_id)
api = NutritionAPI(spike_auth)

# Upload image asynchronously
print(f"Asynchronous analysis")
result = api.analyze_food_image(
    "/path/to/image.jpg.jpg",
    include_ingredients=True
)
print_result(result)

# For synchronous processing (wait for results)
print(f"\nSynchronous analysis")
result = api.analyze_food_image(
    "/path/to/image.jpg.jpg",
    wait_on_process=True,
    include_ingredients=True,
    language_code="de",
    country_code="de"
)
print_result(result)