DEV Community: Aaron Ploetz

Building a Weather App with a Raspberry Pi, Astra DB, and Langflow

Aaron Ploetz — Fri, 14 Mar 2025 19:11:35 +0000

To celebrate PI Day this year, we thought it would be fun to build something with a Raspberry Pi that uses Astra DB and/or Langflow. Fortunately, I have just the project in-mind: A weather application!

To that end, our goal will be to use the National Weather Service’s (NWS) data API. Essentially, we will call this API to get the most recent weather data, store it in Astra DB, and display it on a simple front-end.

Requirements

To build our project, we’re going to need a few things. First of all, our development environment will use the following:

Java 17
Spring Boot
Maven
Vaadin
An Astra DB account with an active database and Langflow instance.

And of course, we’ll also need a Raspberry Pi. For this project, we used a Cana Kit™ Raspberry Pi 5 Starter Kit PRO Turbine Black (4GB RAM / 128GB Micro SD).

The weather application

The weather application we will use can be found in this GitHub repository: https://github.com/aar0np/weather-app/tree/main

This application originally appeared in Chapter 8 of the book Code with Java 21. The original project was designed to work with DataStax Astra DB, using the CQL protocol. Our fork of it is a bit different, as it can refresh its data view from either the Astra DB Data API or from a Langflow API endpoint.

At its core, the project is a Java Spring Boot application, which has a Vaadin web front end, and also exposes two restful endpoints. The idea behind the endpoints is as much for testing as it is functional. One endpoint pulls the most recent update from the NWS API for a given weather station ID, and stores it in Astra DB. The other endpoint retrieves the most-recent reading from Astra DB for a particular station and year/month combination.

Restful examples

Pulling the latest reading from the NWS for the station KMSP (Minneapolis/St.Paul International Airport), and storing it in Astra DB:

curl -X PUT http://127.0.0.1:8080/weather/astradb/api/latest/station/kmsp

This endpoint returns a response similar to the following:

{"stationId":"https://api.weather.gov/stations/KMSP","monthBucket":202503,"timestamp":"2025-03-07T22:53:00Z","readingIcon":"https://api.weather.gov/icons/land/day/few?size=medium","stationCoordinatesLatitude":-93.22,"stationCoordinatesLongitude":44.88,"temperatureCelsius":5.6,"windDirectionDegrees":310,"windSpeedKMH":20.52,"windGustKMH":0.0,"visibilityM":16090,"precipitationLastHour":0.0,"cloudCover":{"7620":"FEW","1830":"FEW"}}

This endpoint pulls the latest reading for a specific station and year/month combination:

curl -X GET http://127.0.0.1:8080/weather/astradb/api/latest/station/kmsp/month/202503

This endpoint returns a response similar to the following:

{"stationId":"kmsp","monthBucket":202503,"timestamp":"2025-03-07T22:53:00Z","readingIcon":"https://api.weather.gov/icons/land/day/few?size=medium","stationCoordinatesLatitude":-93.22,"stationCoordinatesLongitude":44.88,"temperatureCelsius":5.6,"windDirectionDegrees":310,"windSpeedKMH":0.0,"windGustKMH":0.0,"visibilityM":0,"precipitationLastHour":0.0,"cloudCover":{"7620":"FEW","1830":"FEW"}}

Note: The above restful GET call is also used to populate the web frontend.

The Astra DB Data API

First, create a new Astra DB database. We can also use an existing database, as long as we create the following:

Keyspace named: “weatherapp”
Non-vector collection named: “weather_data”

There are two primary controller methods that handle the above data calls. The first method is named putLatestAstraAPIData and handles the restful PUT call. First, it performs a GET call on the NWS API endpoint for the stationid that was passed-in. It takes the payload, maps it to an Astra DB Data API Document type named weatherDoc. Then, it saves weatherDoc in Astra DB (via the Data API). Finally, it maps the response to a WeatherReading object named currentReading, and returns it. The code for the putLatestAstraAPIData() method is shown below:

@PutMapping("/astradb/api/latest/station/{stationid}")
public ResponseEntity<WeatherReading> putLatestAstraAPIData(
             @PathVariable(value="stationid") String stationId) {

       LatestWeather response = restTemplate.getForObject(
                    "https://api.weather.gov/stations/" + stationId + 
                    "/observations/latest", LatestWeather.class);

       Document weatherDoc = mapLatestWeatherToDocument(response, stationId);

       // save weather reading
       collection.insertOne(weatherDoc);

       // build response
       WeatherReading currentReading =
                    mapLatestWeatherToWeatherReading(response);

      return ResponseEntity.ok(currentReading);
}

The other controller method is named getLatestAstraAPIData and it handles the RESTful GET call. This method takes the stationid and the monthBucket that were passed-in, and uses the Data API to find any matching documents. As there might be multiple, the results are sorted in descending order by timestamp, and the top document is processed. This ensures that the latest document is mapped and returned.

The code for the getLatestAstraAPIData() method is shown below:

@GetMapping("/astradb/api/latest/station/{stationid}/month/{month}")
public ResponseEntity<WeatherReading> getLatestAstraAPIData(
             @PathVariable(value="stationid") String stationId,
             @PathVariable(value="month") int monthBucket) {

       Filter filters = Filters.and(eq("station_id",(stationId)),
                           (eq("month_bucket",monthBucket)));
       Sort sort = Sorts.descending("timestamp");
       FindOptions findOpts = new FindOptions().sort(sort);
       FindIterable<Document> weatherDocs = collection.find(filters, findOpts);
       List<Document> weatherDocsList = weatherDocs.all();

       if (weatherDocsList.size() > 0) {
              Document weatherTopDoc = weatherDocsList.get(0);
              WeatherReading currentReading =
                           mapDocumentToWeatherReading(weatherTopDoc);
              return ResponseEntity.ok(currentReading);
       }

       return ResponseEntity.ok(new WeatherReading());
}

The Langflow API

This entire process can also work through Langflow. Open up Langflow, create a new flow, and pick the “Simple Agent” template. This simple agent is all that we need.

A sample flow created by selecting the “Simple Agent” template in Langflow.

The agent is built with the URL “tool,” which allows the agent to call out to external web addresses, including APIs. To expose this agent, we simply need to click on the “API” tab and make note of the Langflow endpoint URL. We will add this URL as an environment variable with our application.

Inside our application, our call to Langflow is handled by a method named askAgent. Simply put, this method calls our Langflow API endpoint, maps the result, and returns it to the UI. The code for the askAgent() method can be seen below:

public WeatherReading askAgent (AgentRequest req) {

       String reqJSON = new Gson().toJson(req);
       HttpEntity<String> requestEntity =
                    new HttpEntity<>(reqJSON, langflowHeader);

       ResponseEntity<LangflowResponse> resp =
                    restTemplate.exchange(LANGFLOW_URL,
                    HttpMethod.POST,
                    requestEntity,
                    LangflowResponse.class);

       LangflowResponse lfResp = resp.getBody();
       LangflowOutput1[] outputs = lfResp.getOutputs();

       return mapLangflowResponseToWeatherReading(outputs);
}

The method inside our Vaadin UI code that calls the askAgent() method, is named refreshLangflow() and is triggered by a button on the UI. It composes a message for our Langflow agent, sends it, and uses the data returned to refresh the UI. The code can be seen below:

private void refreshLangflow() {

      String message = "Please retrieve the latest weather data (including the weather icon url) in a text format using this endpoint: "
+ "https://api.weather.gov/stations/" + stationId.getValue() + 
"/observations/latest";
       latestWeather = controller.askAgent(new AgentRequest(message));

       refreshData(latestWeather);
}

After reviewing the code, we should now be ready to build and configure our hardware.

Raspberry Pi

First of all, we will need to assemble the Pi. Fortunately, Cana Kit has a great setup video that walks through the entire process.

Note: I prefer to use Cana Kits, because they come with everything that you need, such as a Micro HDMI cable, a heat sink with fan, and a Micro SD card.

Once the Pi is assembled and running, it will check for updates and reboot. When we get to the Raspberry Pi OS desktop, we will have a few things to install (using the Terminal application).

Java

For our application to run, we need a Java Virtual Machine (JVM). As we will also need to build our application locally, we’ll need a Java Development Kit (JDK) as well. In our case, our Pi had Java 17 installed, and this is sufficient for our purposes.

Note: The Raspberry Pi OS makes it difficult to install and configure newer versions of Java. Fortunately, our project compiles just fine with Java 17.

Maven

Maven is a build- and dependency-management tool for Java. Our project was built with Maven, so we will need to install it as well:

sudo apt install maven

Git

Our Pi also had Git installed. After creating a new SSH key and adding it to your GitHub account, we should be able to clone the project repository:

git clone git@github.com:aar0np/weather-app.git

This will download the code and create a local directory for our application, where we can build and run it.

Putting it all together

First, we will cd into our project directory and then build it with Maven:

cd weather-app
mvn clean install -Pproduction

Next, we need to define three environment variables:

export ASTRA_DB_API_ENDPOINT=https://not-real-us-east1.apps.astra.datastax.com
export ASTRA_DB_APP_TOKEN=AstraCS:wtqNOTglg:725REAL238dEITHER563486d
export ASTRA_LANGFLOW_URL=https://api.langflow.astra.datastax.com/lf/6f-not-real-9493/api/v1/run/060d2-not-real-caef?stream=false

We can now run our application. Maven will create a JAR file in the weather-app/target directory. If we locate this JAR file, we can run it like this:

java -jar target/weatherapp-0.0.1-SNAPSHOT.jar

A successful run should produce several log messages, the last of which should look similar to this:

2025-03-07T20:02:02.259-06:00  INFO 53787 --- [WeatherApp] [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port 8080 (http) with context path '/'
2025-03-07T20:02:02.278-06:00  INFO 53787 --- [WeatherApp] [           main] c.d.weatherapp.WeatherappApplication     : Started WeatherappApplication in 1.795 seconds (process running for 2.147)
2025-03-07T20:02:04.855-06:00  INFO 53787 --- [WeatherApp] [nio-8080-exec-1] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring DispatcherServlet 'dispatcherServlet'
2025-03-07T20:02:04.855-06:00  INFO 53787 --- [WeatherApp] [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Initializing Servlet 'dispatcherServlet'
2025-03-07T20:02:04.856-06:00  INFO 53787 --- [WeatherApp] [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Completed initialization in 0 ms

From another terminal window/tab, let’s add some data to the DB:

curl -X PUT http://127.0.0.1:8080/weather/astradb/api/latest/station/kmsp

Now, navigate to the local IP on port 8080: http://127.0.0.1:8080/

The Station ID is pre-populated with “kmsp” and the current year/month is auto-generated. Clicking either “Astra DB Refresh” or “Langflow Refresh” should produce something similar to this:

Note: The Astra DB Refresh will be faster than the Langflow Refresh.

Depending on the intended usage, it might be preferable to add a crontab entry for the PUT call to keep the data recent:

crontab -e
...
15 * * * * curl -X PUT http://127.0.0.1:8080/weather/astradb/api/latest/station/kmsp

With that, we should have a working weather application running on a Raspberry Pi! And without having to mount any pesky sensors in the backyard. Want to try this yourself? Get started with Astra DB and Langflow today!

Happy Pi Day!

How to Build a Crystal Image Search App with Vector Search

Aaron Ploetz — Mon, 29 Apr 2024 16:54:49 +0000

There are lots of ways to leverage generative AI (GenAI) in a variety of business use cases at companies of all sizes. In this post, we will explore how a store selling crystals and precious stones can use DataStax’s RAGStack to help their customers to identify and find certain crystals. Specifically, we will walk through creating an application designed to help the customers of Healing House Energy Spa (owned by the author’s wife). This will also demonstrate how small businesses can take advantage of GenAI.

What is RAGStack?

RAGStack is DataStax’s Python library that’s designed to help developers build advanced GenAI applications based on retrieval-augmented generation (RAG) techniques. These applications require developers to configure and access data parsers, large language models (LLMs), and vector databases.

With RAGStack, developers can increase their productivity with GenAI toolsets by interacting with them through a single development stack. DataStax’s integrations with many commonly used libraries and providers enable developers to prototype and build applications faster than ever before. All of this happens on top of DataStax Astra DB, which is DataStax’s powerful, multi-region vector database (as shown in Figure 1).

Figure 1 - A high-level view of the Crystal Search application architecture, showing how it leverages RAGStack.

As Astra DB is a key component of RAGStack, we should spend some time discussing vector databases. These are special kinds of databases capable of storing vector data in native structures. When we build RAG applications, we interact with an LLM by using a “vectorized” version of our data. Essentially, the vectors returned are a numerical representation of the individual elements or “chunks” of our data. We will discuss this process in more detail below.

The Crystal Search application

Here we'll walk through how to build up a simple web application to search an inventory of crystals (and other precious stones). We’ll load our data from a CSV file, and then query it using a Flask-based web application with navigation drop-downs and a search-by-image function.

The crystals themselves have several properties:

Name What the crystal is known as.
Image The filename of the on-disk image of the crystal.
Chakras One or more of the seven centers of spiritual power in the human body that the crystal can help attune.
Birth month People with certain birth months will be more receptive to this crystal.
Zodiac sign People born under certain zodiac signs will be more receptive to this crystal.
Mohs hardness A measure of the crystal’s resistance to scratching.

For our drop-down navigation, we will use a crystal’s recommended chakras, birth month, and zodiac signs. The remaining properties will be added to the collection’s metadata (except for the image itself, which will be used to generate the crystal’s vector embedding).

We will use the CLIP model to generate our vector embeddings. CLIP (Contrastive Language-Image Pre-training) is a sentence transformer model (developed by OpenAI) used to store both images and text in the same vector space. The CLIP model is pre-trained with images and text descriptions, and enables us to return results using an approximate nearest neighbor (ANN) algorithm. Leveraging CLIP in this way allows us to support an “identify this crystal” function, where users will be able to search with a picture from their device.

Requirements

Before building our application, let’s make sure that we properly configure our development environment. We will start by making sure that our Python version is at least on version 3.9. We will also need the following libraries (and versions), as specified in our [requirements.txt](https://github.com/aar0np/crystalSearch/blob/main/requirements.txt) file.

Flask==2.3.2
Flask-WTF==1.2.1
sentence-transformers==2.2.2
ragstack-ai==0.8.0
python-dotenv==1.0.0

pip install -r requirements.txt

Flask directory structure

As we are working with a Flask web application, we will need the following directory structure, with crystalSearch as the “root” of the project:

crystalSearch/
      templates/
      static/
            images/
            input_images/
            web_images/

DataStax Astra DB

First, we need to sign up for a free account with DataStax Astra DB, and create a new vector database. Once we have our Astra DB vector database, we will make note of the token and API endpoint. We will define those as environment variables in the next section.

Environment variables

For our application to run properly, we'll need to set some environment variables:

ASTRA_DB_API_ENDPOINT - Connection endpoint for our Astra DB vector database instance.
ASTRA_DB_APPLICATION_TOKEN - Security token used to authenticate to our Astra DB instance.
FLASK_APP - The name of the application’s primary Python file in a Flask web project.
FLASK_ENV - Indicates to Flask if the application is in development or production mode.

Of course, the easiest way to do that is with an .env file. Our .env file, should look something like this:

ASTRA_DB_API_ENDPOINT=https://notreal-blah-4444-blah-blah-region.apps.astra.datastax.com
ASTRA_DB_APPLICATION_TOKEN=AstraCS:NotReal:ButYourTokenWillLookSomethingLikeThis
FLASK_APP=crystalSearch
FLASK_ENV=development

Setting the FLASK_APP variable to “crystalSearch” is important, as it tells Flask which Python module is the primary entrypoint to the application.

crystalLoader.py

With our database and environment all set up, we can build our Python data loader. Create a new Python file named crystalLoader.py, and set up its imports like this:

import csv
import json

from os import path, environ
from dotenv import load_dotenv
from PIL import Image
from astrapy.db import AstraDB
from sentence_transformers import SentenceTransformer

We will start by bringing in the environment variables from our .env file:

basedir = path.abspath(path.dirname(__file__))
load_dotenv(path.join(basedir, '.env'))

Next, we will pull in the application endpoint and token, instantiate a database connection object, and then create a new collection named “crystal_data”:

# Astra connection
ASTRA_DB_APPLICATION_TOKEN = environ.get("ASTRA_DB_APPLICATION_TOKEN")
ASTRA_DB_API_ENDPOINT= environ.get("ASTRA_DB_API_ENDPOINT")

db = AstraDB(
    token=ASTRA_DB_APPLICATION_TOKEN,
    api_endpoint=ASTRA_DB_API_ENDPOINT,
)

# create "collection"
col = db.create_collection("crystal_data", dimension=512, metric="cosine")

Note that our collection will have a vector capable of supporting 512 dimensions, so that it matches the dimensions of the vector embeddings created with the CLIP model. Astra DB supports the use of ANN searches with a cosine, dot product, or Euclidean algorithm. For our purposes, a cosine-based ANN will be fine.

Next, we will define some constants to help our loader:

model = SentenceTransformer('clip-ViT-B-32')
IMAGE_DIR = "static/images/"
CSV = "gemstones_and_chakras.csv"

These will instantiate the clip-ViT-B-32 model locally, define a location for our images, and data filename, respectively.

Now let’s open the CSV file in a with block and initialize the data reader:

with open(CSV) as csvHandler:
    crystalData = csv.reader(csvHandler)
    # skip header row
    next(crystalData)

Our CSV file has a header row that we will skip at read-time. The next() function (from Python’s CSV library) is an easy way to iterate over it.

With that complete, we can now use a for loop to work through the remaining lines in the file. We will first read the line’s image column. As our application is very image-centric, we do not want to spend time processing a line if it doesn’t have a valid image. We will use an if conditional to make sure that the file referenced by image column is both:

not empty
a valid file that exists

for line in crystalData:

    image = line[1]
    # Only load crystals with images
    if image != "" and path.exists(IMAGE_DIR + image):
        # map columns
        gemstone = line[0]
        alt_name = line[2]
        chakras = line[3]
        phys_attributes = line[4]
        emot_attributes = line[5]
        meta_attributes = line[6]
        origin = line[7]
        description = line[8]
        birth_month = line[9]
        zodiac_sign = line[10]
        mohs_hardness = line[11]

If the image for each line in the CSV file is indeed valid, we will then map the remaining columns to local variables.

Two of our variables, chakras and mohs_hardness, will require some extra processing before being written into Astra DB. Our chakra data comes from the file as a comma-delimited list. Crystals can affect multiple chakras. Therefore, we will need to reconstruct it into an array with each item wrapped in quotation marks, so that it is recognized as valid JSON. To do that, we will simply replace the commas with double-quoted commas:

            # reformat chakras to be more JSON-friendly
            chakras = chakras.replace(', ','","')

This will not make it valid JSON on its own, so we will account for that later when we write the chakra data.

Precious stones all have a rating on the Mohs hardness scale, which indicates its resistance to scratches. While some crystals in our data set have a value of a single integer, several do occupy a range on the scale (with the minimum listed first), indicating a maximum and a minimum Mohs hardness. We will split-out these values, and store them as mohs_min_hardness and mohs_max_hardness, respectively. Do note that sometimes the mohs_hardness column will have a value of “Variable” or “Varies,” so we will account for that possibility as well:

            # split out minimum and maximum mohs hardress
            mh_list = mohs_hardness.split('-')
            mohs_min_hardness = 1.0
            mohs_max_hardness = 9.0
            if mh_list[0][0:4] != 'Vari':
                mohs_min_hardness = mh_list[0]
                mohs_max_hardness = mh_list[0]
                if len(mh_list) > 1:
                    mohs_max_hardness = mh_list[1]

With our data prepared, we can now build each crystal’s text and metadata properties:

            metadata = (f"gemstone: {gemstone}")

            text = (<em>f</em>"gemstone: {gemstone}| alternate name: {alt_name}| physical attributes: {phys_attributes}| emotional attributes: {emot_attributes}| metaphysical attributes: {meta_attributes}| origin: {origin}| maximum mohs hardness: {mohs_max_hardness}| minimum mohs hardness: {mohs_min_hardness}")

Next, we can load the crystal’s image using Pillow (Python’s image processing library) and generate a vector embedding for it with the encode() function from our CLIP model:

            img_emb = model.encode(Image.open(IMAGE_DIR + image))

With all that complete, we are ready to build our local JSON document as a string:

            strJson = (f' {{"_id":"{image}","text":"{text}","chakra":["{chakras}"],"birth_month":"{birth_month}","zodiac_sign":"{zodiac_sign}","$vector":{str(img_emb.tolist())}}}')

Finally, we can convert each crystal’s data to JSON and write it into Astra DB:

            doc = json.loads(strJson)
            col.insert_one(doc)

crystalSearch.py

To demonstrate the visual aspects of Crystal Search, we will stand-up a simple web application using Flask. This interface will have a few simple components, including dropdowns (for navigation) and a way to upload an image for searching.

Note: As web front-end development is not the focus, we’ll skip the implementation details. For those who are interested, the code can be accessed in the project repository listed at the end of this post.

astraConn.py

Now that our data has been loaded, we can build the Crystal Search application. First, we will construct the astraConn module, which will act as an abstraction layer for our interactions with the Astra DB vector database. We will create a new file named astraConn.py and add the following two imports:

import os

from astrapy.db import AstraDB

Next, we will pull-in our ASTRA_DB_APPLICATION_TOKEN and ASTRA_DB_API_ENDPOINT variables from our system environment, and instantiate them locally:

ASTRA_DB_APPLICATION_TOKEN = os.environ.get("ASTRA_DB_APPLICATION_TOKEN")
ASTRA_DB_API_ENDPOINT= os.environ.get("ASTRA_DB_API_ENDPOINT")

This module will have a few different methods that will be called by our application, but we won’t want to rebuild our database connection each time. Therefore, we will create two global variables (db and collection) to keep data pertaining to our database cached:

db = None
collection = None

The first method that we will define will be the init_collection() method. This method will be called by every other method in this module. It will first initiate global scope access for the db and collection variables. Its primary function will be to instantiate the db object if it is null or “None.” This way, an existing connection object can be reused. The code for this method is shown below:

def init_collection(table_name):
    global db
    global collection

    if db is None:
        db = AstraDB(
            token=ASTRA_DB_APPLICATION_TOKEN,
            api_endpoint=ASTRA_DB_API_ENDPOINT,
        )

    collection = db.collection(table_name)

Note that the collection variable will be instantiated on every call. This allows us the flexibility to access different collections in Astra DB with the same database connection information.

For our application, there are three ways that we will perform reads on our data. We will search by vector, query by id, and then query by three additional properties that we are going to build into dropdowns in our web application.

First, we will build the get_by_vector() method. This asynchronous method will accept a collection name, a vector embedding, and a maximum (limit) number of results to be returned (defaulting to 1). After initializing our database and collection, we will invoke the vector_find() method with the vector_embedding, the limit, and the list of fields from the collection that we want to receive. We will then return the results to the calling method.

async def get_by_vector(collection_name, vector_embedding, limit=1):
    init_collection(collection_name)

    results = collection.vector_find(vector_embedding.tolist(), limit=limit, fields={"text","chakra","birth_month","zodiac_sign","$vector"})
    return results

Our get_by_id() method will be similar to the previous one, but will work quite differently under the hood. This method is also meant to be called asynchronously, and accepts a collection name as well as the identifier to be queried. As querying by a unique identifier is deterministic, we can invoke the find_one() method with a filter for the specific id, as shown below:

async def get_by_id(collection_name, id):
    init_collection(collection_name)

    result = collection.find_one(filter={"_id": id})
    return result

This method will return a single JSON document as the result.

Finally, get_by_dropdowns() is an asynchronous method that will return all matching rows based on the values of three properties: chakras, birth month, and zodiac sign. First, we will build an array to hold our conditions. This is necessary because not every dropdown is going to be used each time. That way we can dynamically build our conditions based on the state of the dropdowns at query-time.
async def get_by_dropdowns(collection_name, chakra, birth_month, zodiac_sign):

init_collection(collection_name)

    conditions = []

    if chakra != "--Chakra--":
        condition_chakra = {"chakra": {"$in": [chakra]}}
        conditions.append(condition_chakra)

    if birth_month != "--Birth Month--":
        condition_birth_month = {"birth_month": birth_month}
        conditions.append(condition_birth_month)

    if zodiac_sign != "--Zodiac Sign--":
        condition_zodiac_sign = {"zodiac_sign": zodiac_sign}
        conditions.append(condition_zodiac_sign)

    crystal_filter = ""

    if len(conditions) > 2:
        crystal_filter = {"$and": [{"$and": [conditions[0], conditions[1]]}, conditions[2]]}
    elif len(conditions) > 1:
        crystal_filter = {"$and": [conditions[0], conditions[1]]}
    elif len(conditions) > 0:
        crystal_filter = conditions[0]
    else:
        return 

    results = collection.find(crystal_filter)
    return results

Once the conditions array is built, we can then build crystal_filter to use as our JSON query string. To pass a filter with multiple conditions through Astra DB’s Data API, we need to build a nested conditional statement.

A single condition could be sent as a filter on its own. But two would need to use the $and operator. If we were to hard-code our filter, it would be similar to this example:

crystal_filter = {"$and": [{"birth_month": "October"}, {"zodiac_sign": "Libra"}]}

Of course, this also means that three conditions would require a nested $and (one $and inside of another), like this:

crystal_filter = {"$and": [{"$and": [{"birth_month": "October"}, {"zodiac_sign": "Libra"}]}, {"chakra": {"$in": ["Heart"]}}]}

Note that as each crystal’s chakra property is an array, we need to use the $in operator.

crystalServices.py

Next, we will create a new file named crystalServices.py with the following imports:

import json
import os

from astraConn import get_by_vector
from astraConn import get_by_id
from astraConn import get_by_dropdowns
from sentence_transformers import SentenceTransformer
from PIL import Image

We will also define some local variables for our image directory, the name of our collection in Astra DB, and our CLIP model:

INPUT_IMAGE_DIR = "static/input_images/"
DATA_COLLECTION_NAME = "crystal_data"
model = None

Our service layer will expose two asynchronous methods. The first method that we will build, will be named get_crystals_by_image, and it will accept an image filename as a parameter. It will be primarily responsible for generating a vector embedding from an image, using the embedding to invoke a vector similarity search, and returning the results to the view. This method will need the model global variable, and instantiate it if required:

async def get_crystals_by_image(file_path):
    global model

    if model is None:
        model = SentenceTransformer('clip-ViT-B-32')

Next, we will define our result set variable as an empty dictionary. Then we will load the image, generate an embedding for it, and use it to call the get_by_vector() method from (astraConn.py):

    results = {}        
    img_emb = model.encode(Image.open(INPUT_IMAGE_DIR + file_path))
    crystal_data = await get_by_vector(DATA_COLLECTION_NAME, img_emb, 3)

    if crystal_data is not None:
        for crystal in crystal_data:
            id = crystal['_id']
            results[id] = parse_crystal_data(crystal)

    return results

Finally, we will process and return the vector search results. Note that the parse_crystal_data() method does much of the heavy-lifting of building the result set. We will construct that method toward the end of this module.

We will now move on to the get_crystals_by_facets() method. This method accepts the values taken from three dropdown lists containing data for chakras, birth month, and zodiac sign. Similar to the prior method, we will define an empty dictionary for the results and perform a query on our data, before processing and returning the results:

async def get_crystals_by_facets(chakra, birth_month, zodiac_sign):
 results = {}
 crystal_data = await get_by_dropdowns(DATA_COLLECTION_NAME, chakra, birth_month, zodiac_sign)

 if crystal_data is not None:
  for crystal in crystal_data['data']['documents']:
   id = crystal['_id']
   results[id] = parse_crystal_data(crystal)

 return results

There are also two additional code blocks required to more easily transfer our data back up to the view layer. The first is the parse_crystal_data() method. This method is fairly straightforward in that it takes the raw crystal data as a parameter, and processes each property into a new object of the Crystal class. As the final part of this module, we also need to add the Crystal object class. They will not be shown here, but both of these definitions can be found at the end of the crystalServices.py module.

Demo

Let’s see this in action. We will run the application with Flask. The complete code listed above (including all of the front end components) can be found in this GitHub repository.

To run the application, we will use the following command:

flask run -p 8080

If it starts correctly, Flask should display the application name, address and port that it is bound to:

 * Serving Flask app 'crystalSearch'
 * Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on http://127.0.0.1:8080

Press CTRL+C to quit

If we navigate to that address in a browser, we should see a simple web page with a search interface at the top, and three differently-colored dropdowns in the left navigation. If we select values for the dropdowns and click on the “Find Crystals” button, we should see crystals matching those values returned (Figure 2).

Figure 2 - Results for crystals matching the dropdown values where chakra is “Heart”, birth month is “October,” and zodiac sign is “Libra.”

Of course, we can also search with an image. Perhaps we have a picture of a crystal that we cannot identify. We can click on the “Choose File” button, select our image, and then click “Search” to see what the closest matches are. If our picture is of a black obsidian crystal, we will see results similar to Figure 3.

Figure 3 - Results for crystals matching our image of a black obsidian crystal.

Conclusion

In this article, we have demonstrated another possible use case for an image-based search built with RAGStack and Astra DB. We walked through this very unique use case, how to configure the development environment, load and query data using CLIP, and build an application to leverage image-based vector embeddings. We also showed how to use the Astra DB Data API to implement a simple product faceting approach using dropdowns.

As the world continues to embrace GenAI, we will surely see more and more creative use cases spanning multiple industries. Searching by images using CLIP is one of the ways in which we are pushing the boundaries of conventional data applications. With solutions like RAGStack and Astra DB, DataStax continues to help you build the next generation of applications.

Do you have an idea for a great use of GenAI? Pull down RAGStack and start using Astra DB with a free account today!

The Distributed Data Problem

Aaron Ploetz — Tue, 18 Oct 2022 18:33:20 +0000

Getty Images

Today, online retailers sell millions of products and services to customers all around the world. This was more prevalent in 2020, as COVID-19 restrictions all but eliminated visits to brick-and-mortar stores and in-person transactions. Of course, consumers still needed to purchase food, clothing, and other essentials and, as a result, world-wide digital sales channels rose to the tune of $4.2 trillion, up $900 billion from just a year prior.

Was it enough for those retailers to have robust websites and mobile apps to keep their customers from shopping with competitors? Unfortunately, not. Looking across the ecommerce landscape of 2020, there were clear winners and losers. But what was the deciding factor?

Consider this: 40% of consumers will leave after only three seconds if a web page or mobile app fails to fully load. That’s not a lot of time to make sure that everything renders properly. While there is a lot to be said for properly optimizing images and code, all that work can be for naught if data latency consumes a significant portion of load time.

Co-location helps—but can pose new challenges

A good way to cut down on data latency is to deploy databases in the same geographic regions as the applications they serve. For example, an enterprise headquartered in San Diego may have applications deployed in data centers located in the Western United States (US), the Eastern US, and Western Europe. The customers who live in those regions are directed to the applications and services closest to them.

But what kind of experience will a customer in London have if the app performs quickly, but then slows? Or worse, what if it stalls when it has to make a data call back to San Diego? This is why co-locating an application with its data is so important.

The concept is deceptively simple. Data needs to be distributed to the regions where it’s needed. Actually, accomplishing that, however, can be challenging. One option would be to deploy our data in a traditional, relational database management system (RDBMS) with one database server in each of the three regions.

But then we have to deal with questions that arise from an operational standpoint. How would we keep the database servers in-sync across all regions? How would we scale them to meet fluctuating traffic demands? Questions like these drive what I call the distributed data problem. And, like any complex problem, addressing the issue often requires specific tools.

The promise of NoSQL with horizontal scaling and data center awareness

This is where non-relational, NoSQL (“not only SQL”) databases come into play. NoSQL databases primarily evolved over the last decade as an alternative to single-instance relational database management systems, or RDBMS, which had trouble keeping up with the throughput demands of web-scale internet traffic. They solve scalability problems through a process known as “horizontal scaling” where multiple server instances of the database are linked to each other to form a “cluster.”

Some of the NoSQL database products were also engineered with data center awareness, meaning the database is configured to logically group together certain instances to optimize the distribution of user data and workloads.

For instance, Apache Cassandra, the open source NoSQL database that was introduced by Facebook in 2007, is both horizontally scalable and data center aware. If we were to deploy Cassandra to solve this problem, it would look something like the image below.

DataStax

A hypothetical deployment of Apache Cassandra, with one cluster spanning three regional data centers deployed in the Western US, Eastern US, and Western Europe. Map not to scale.

Our hypothetical ecommerce retailer headquartered in San Diego could perform writes on their “local” data center (DC), which would be the Western US DC. That data would then replicate to the other two data centers, in the Eastern US and Western Europe. The applications deployed regionally could then be configured to get data from their local data center.

In this way, all data interactions initiated by the customer would be limited to their own geographic area. This prevents cross-DC, high-latency operations from being noticeable to any end users.

The other advantage to this type of deployment is that the regional data centers can be scaled independently of each other. Perhaps traffic is increasing in Western Europe, requiring additional resources to be added. In this case, new database instances can be added to just that one DC; the resource levels of those that aren’t needed won’t increase—and won’t add unnecessary cost. Once the traffic subsides, the instances in a DC can also be reduced to lower costs.

Of course, scaling a cluster or a DC up and down can require complex infrastructure operations. Deploying databases while using orchestration platforms like Kubernetes can greatly ease that burden, enabling enterprises to worry less about infrastructure and more about new capabilities (To learn more about this topic, check out “A Case for Databases on Kubernetes from a Former Skeptic). Cloud-native, serverless databases like DataStax Astra DB can take that a step further, essentially rendering the underlying infrastructure invisible to application developers.

Two predictions

I expect that in another 10 years, single-instance databases will become a relic of the past. As that happens, more and more database products will embrace data center awareness.

The trends of horizontal scaling and data center awareness are driving more innovation in database technologies than ever before.

It’s difficult to see when the next big uptick in ecommerce will happen or how widespread the effect will be relative to what we saw in 2020. One thing is certain. Those enterprises that are able to solve the distributed data problem by reducing latency to deliver data more quickly to the apps their customers are using will find themselves well ahead of their competition.

To learn more, visit us here.