DEV Community: Ngonidzashe Nzenze

Building an AI Anime Suggestion Tool with LangChain and Streamlit

Ngonidzashe Nzenze — Wed, 14 May 2025 14:21:08 +0000

I have been wondering how LLMs can read and analyze files for a while now. So I decided to learn how they did it, and my research led me to tools. I've been looking for a good anime to watch, so I figured, how about I learn how I can create a tool that can recommend some good shows? So I grabbed an anime dataset from Kaggle and went about creating the tool.

It took me a while to figure out the best way to make use of the tool, but hey, I think it was worth it. In this article, we’ll cover:

Creating an agent using Langchain
Tool calling through the agent
Safety considerations when working with tools
Adding memory to the agent
Returning a structured response
Caching in Streamlit

The idea behind the app is pretty simple: we have an Excel file with anime ratings. We make use of pandas, specifically the data frame query method, to filter through the data and return a result. This result is fed to the agent, and it returns a response based on the result. The agent's job, therefore, is to come up with queries that are fed to the tool and to give a response based on the result returned by the tool.

Initial setup

In a virtual environment, install the required packages with:
pip install langchain langchain-core langchain-community langgraph pandas openpyxl streamlit

streamlit: For building the interactive web application.
langchain, langgraph, langchain-core, langchain-community: The core libraries for building the agent, defining tools, and managing the application flow.
langchain-openai: To use OpenAI models like GPT-4o-mini.
pandas, openpyxl: For reading and querying the Excel data (.xlsx file).

I am using python-environ to manage my environmental variables; if you wish to do the same, install it with pip install python-environ

All the code is available on my github

Create two files:

chat.py - for the streamlit interface
llm_utils.py - all our agent code

Inside llm_utils.py, add the following code:

import re, os
from langchain_core.tools import tool
from langchain.chat_models import init_chat_model
from langgraph.prebuilt import create_react_agent
import pandas as pd
from langchain_core.messages import HumanMessage, SystemMessage
from typing import Union, List, Literal, Optional
import streamlit as st
from pydantic import BaseModel, Field

import environ
env = environ.Env()
environ.Env.read_env()
APIKEY = env("OPENAI_API_KEY")

if not os.environ.get("OPENAI_API_KEY"):
    os.environ["OPENAI_API_KEY"] = APIKEY

We import all the necessary packages that we are going to be using in this project. Make sure you have a .env file with your OpenAI key.

Safety Considerations

Before we proceed, we need to talk a bit about safety when working with large language models. You see, the query method is capable of executing Pythonic code, which means if the model were to provide a malicious query, there could be problems. If, for example, the model were to give an input query like
__import__('os').system('rm -rf /')
This query tries to delete everything. The result could be quite catastrophic; therefore, we have to take some safety precautions. Just telling the model not to return any malicious code is not enough. In this demonstration, I am going to create a function that determines whether a query is safe to execute, returning an exception otherwise. This function won't deal with all possibilities; therefore, I suggest you take a look at the best practices when you're implementing your own solution.

In the llm_utils.py file, add the following code:

def is_safe_query(query_string):
    """
    Checks query safety before execution
    Return boolean. True if string is safe and false otherwise
    """
    dangerous_keywords = [
        "import",
        "eval",
        "exec",
        "execfile",
        "globals",
        "locals",
        "open",
        "compile",
        "input",
    ]

    if "__" in query_string or ";" in query_string:
        return False

    # Check for potentially dangerous words
    for keyword in dangerous_keywords:
        pattern = r"\b" + re.escape(keyword) + r"\b"
        if re.search(pattern, query_string, re.IGNORECASE):
            return False

    return True

Creating the tool

Creating a tool in LangChain is quite straightforward; we just need to add the @tool decorator to the function, which makes the function callable by the agent:

@tool
def read_pandas_dataframe(expression: str, limit: Optional[int] = None) -> str:
    """Query a pandas DataFrame using a safe string expression and return filtered results as a JSON string.
    Reads data from 'anime-dataset-2023.xlsx', evaluates a query expression,
    optionally takes the first 'limit' rows using the .head() method,
    and returns a JSON string (records orientation) representing the subset of rows that match the given conditions. Includes a basic security check.
    """
    df = pd.read_excel("anime-dataset-2023.xlsx")

    if not is_safe_query(expression):
        raise ValueError("Invalid query string")
    result = df.query(expression, engine="python", inplace=False)

    if limit is not None and isinstance(limit, int) and limit > 0:
        # Apply the head method to limit the number of rows
        result = result.head(limit)

    if result.empty:
        return "[]"
    elif isinstance(result, pd.Series):
        return result.to_frame().to_json(orient="records")
    else:
        return result.to_json(orient="records")

The tool reads the anime dataset Excel file and executes the query method on the resultant data frame.
Note that the function also takes an optional argument limit. Sometimes the query returns a large number of results, which can be too large an input for the agent's context; therefore, only the first few results should be returned by calling the head method.

When a user asks a question (e.g., "Recommend me an action anime with a score above 8.5"), the agent, guided by the system message, determines it needs data. It formulates a query expression (like 'Genres.str.contains("Action", case=False) and Score >= 8.5') and calls the read_pandas_dataframe tool with that expression as the argument.

The tool executes the query on the data frame and returns the resulting (filtered) data as JSON back to the agent. The agent then uses this data to formulate its final response to the user.

Note that the read_pandas_dataframe function first checks if the query is safe to execute.

Creating the agent

At the heart of the application, we have the agent. I am going to make use of a prebuilt agent offered by langgraph. In llm_utils.py add the following code to create the agent:

def create_llm_agent():
    model = init_chat_model("gpt-4o-mini", model_provider="openai")
    return create_react_agent(
        model=model,
        tools=[read_pandas_dataframe]
    )

In this example, I am using the OpenAI API, but you can use any of the other providers. Make sure to check the official documentation for details. The create_llm_agent function initializes the chat models first, then creates the agent with the create_react_agent function, which takes the model and the tools as an argument.

The system message

The system message serves to instruct the model on its responsibility or what purpose it serves. In our case, it's going to help us get the hottest anime on the planet. Maybe not because the dataset is from 2023😅.

In llm_utils.py add the following code:

def get_system_message():
    message = SystemMessage(
        """
        Role: You are a helpful Anime Recommendation Assistant.

        Tool Access: You can query a large Excel dataset containing anime data using the `pandas` library and `df.query`. Key columns available are:
        - `Name`
        - `Score`
        - `Genres`
        - `Synopsis`
        - `Type` (e.g., "TV", "Movie", "OVA")
        - `Episodes`
        - `Image URL`

        Primary Task: Respond to user requests by querying this dataset and providing relevant anime recommendations or information.

        CRITICAL LIMITATIONS:
        1. Maximum of 20 Results Per Query**: Due to the dataset's size, all queries must return 20 rows or fewer. Larger result sets will cause system failure.
        2. Minimize Tool Usage: Plan queries to be specific and efficient. Avoid broad, vague searches that force you to retry multiple times. Strive to get accurate results in a single call.
        3. Query Safety: Only generate valid, secure `df.query` strings. Invalid syntax or logic will break the tool.

        Query Construction Rules:
        - Be Specific: Use `and` to combine multiple filters and narrow results.
        - Use Ranges: Apply ranges for scores and episode counts to limit query size (e.g., `Score >= 8.0 and Score < 8.5`).
        - Targeted String Matching:
            - Use `==` for exact values (`Name == "Naruto"`).
            - Use `.str.contains(..., case=False)` for flexible matches in `Genres` or partial name searches — but always pair with other filters.

        - Handling Comma-Separated Genres:  
        The `Genres` column contains comma-separated strings like `"Action, Adventure, Fantasy"`.  

        When filtering by genre:
        - To find anime that match *both* genres (AND logic):
            ```

python
            'Genres.str.contains("Action", case=False) and Genres.str.contains("Adventure", case=False)'


            ```

        - To find anime that match either genre (OR logic):
            ```

python
            'Genres.str.contains("Action", case=False) or Genres.str.contains("Adventure", case=False)'


            ```

        - For better accuracy (avoiding partial word matches like "Action-Drama"), consider using word boundaries with regex:
            ```

python
            'Genres.str.contains(r"\\bAction\\b", case=False)'


            ```

        - Quote Escaping: Use single quotes to wrap the whole query, and double quotes inside it (e.g., `'Name == "Bleach"'`).

        If a user request is vague (e.g., “Show me action anime” or “What's popular?”) and will likely return more than 20 results — You are free to ask them for more specific details.

        Good Query Examples (Safe and Specific):
        - Exact Match
        `'Name == "Death Note"'`  
        *(Very specific, 1 result expected)*

        - High-Rated Movie
        `'Type == "Movie" and Score >= 8.7'`  
        *(Highly rated films only)*

        - Top-Scoring Action Shows
        `'Genres.str.contains("Action", case=False) and Score >= 9.0'`  
        *(Very narrow score filter to limit results)*

        - Multiple Genres (AND)
        `'Genres.str.contains("Adventure", case=False) and Genres.str.contains("Fantasy", case=False) and Score >= 8.0'`

        - Multiple Genres (OR)
        `'Genres.str.contains("Romance", case=False) or Genres.str.contains("Drama", case=False)'`

        - Short Action Series

        `'Genres.str.contains("Action", case=False) and Episodes <= 13 and Score >= 7.5'`  

        *(Filters based on episode count and score)*

        - Score Bracket Strategy
        `'Score >= 8.0 and Score < 8.5'`  
        *(Prevents too broad a score range)*

        - Compact Sci-Fi OVAs
        `'Type == "OVA" and Genres.str.contains("Sci-Fi", case=False) and Episodes <= 6 and Score >= 7.5'`

        "Top N Strategy" for Ranked Recommendations (e.g., Top 5 Romance Anime):

        1. Start narrow:  
        `'Genres.str.contains("Romance", case=False) and Score >= 9.0'`
        2. If result count < 5, expand range slightly:  
        `'Genres.str.contains("Romance", case=False) and Score >= 8.5 and Score < 9.0'`
        3. Combine small result sets **locally**, sort by `Score`, return top 5.

        Output Formatting:

        If you are returning a dataframe, make sure to return it as a string that can be converted into a dictionary using json.loads. Use the tool to get information from the file,
        whether it be the file columns or the actual data. When the user asks about a particular anime, return the image and description.
        You can return a modified description to give better explaination.

        Final Guidelines:
        - Always prioritize specific, narrow queries to stay within the 20-row limit.
        - Never make multiple broad queries when one well-structured query will work.
        - Ask users for more input when queries are too general.
        - Only return results in safe JSON format or as **natural conversation summaries**.
        """
    )

    return message

The system message is pretty long, but that's to ensure that we get the results we want. Feel free to modify the prompt as you see fit.

Returning structured data

Langchain models can return structured data, which is useful since sometimes we want to be able to display images of the anime. In this case, I want the model to return a pydantic object:

# llm_utils.py
class ImageCard(BaseModel):
    """Represents an image card with metadata."""

    title: str = Field(description="Title of the image card")
    description: str = Field(description="Description of the image")
    image_url: str = Field(description="URL of the image")


class ResponseClass(BaseModel):
    """Response to a prompt with dynamic data types."""

    response_type: Literal["text", "image", "dataframe"] = Field(
        description="Type of response (text, image, or dataframe)"
    )

    # Union type for response_data with discriminator based on response_type
    response_data: Union[
        str,  # For "text" type
        List[ImageCard],  # For "image" type (list of image cards)
    ] = Field(..., description="Response data matching the response_type")

I have defined two pydantic classes, the ImageCard class, which will hold data for an anime, that is, it's title, description and image url. This will allow us to easily display the anime in a card.

The ResponseClass class has two fields:

response_type: This specifies which response the model is returning between the three option(text, image, and dataframe). This will help us know how to render the response in the frontend
response_data: The actual data the model is returning, which can be a string or a list of ImageCard objects for when the model returns multiple anime in cards.

Getting the model response

Now we define a function that allows us to get the response from the model:

def get_model_response(llm, user_input: str):
    user_message = HumanMessage(user_input)
    messages = [user_message]
    return llm.invoke({"messages": messages})

The chat interface

And now for the chat interface. I'm going to keep it pretty simple with a text input at the bottom of the page and a sidebar with only one button that will allow us to clear the chat:

In chat.py add the following code:

import streamlit as st

from llm_utils import (

    get_model_response,

    ResponseClass,

    create_llm_agent,

    get_system_message,

)

import pandas as pd

import json



# --- Page Configuration (Optional) ---

st.set_page_config(page_title="LLM Chat", layout="wide")

llm = create_llm_agent()

@st.cache_resource
def invoke_system_message():
    system_message = get_system_message()
    llm.invoke({"messages": [system_message]})

invoke_system_message() 

st.title("Anime Bot🤖")
st.caption("This chat can display text, dataframes, and image cards.")


def get_llm_response(user_prompt):
    """
    Calls the LLM with the user prompt and handles any exceptions.
    """
    try:
        response = get_model_response(llm, user_prompt)
        return response
    except Exception as e:
        # Return a basic error response in the expected format
        return {"response_type": "text", "response_data": f"Error: {str(e)}"}



# --- Chat History Management ---
if "messages" not in st.session_state:
    st.session_state.messages = []

# --- Display Previous Messages ---

for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        # Check the type of content to display
        if message["role"] == "user":
            # User messages are always text
            st.write(message["content"])
        else:
            # For assistant, handle different response types
            content = message["content"]
            if content.response_type == "text":
                st.write(content.response_data)
            elif content.response_type == "dataframe":
                # Convert the DataFrameDict to a pandas DataFrame
                if hasattr(content.response_data, "data"):
                    # It's a DataFrameDict object with .data attribute
                    df = pd.DataFrame(json.loads(content.response_data.data))
                elif (
                    isinstance(content.response_data, dict)
                    and "data" in content.response_data
                ):
                    # It's a dict with "data" key
                    df = pd.DataFrame(content.response_data["data"])
                else:
                    # Try to convert directly
                    df = pd.DataFrame(json.loads(content.response_data))
                st.dataframe(df, use_container_width=True)
            elif content.response_type == "image":
                for card in content.response_data:
                    st.image(card.image_url, caption=card.title)
                    st.subheader(card.title)
                    st.write(card.description)
            else:
                # Fallback for unknown types
                st.write(f"Unsupported content type: {content['response_type']}")



# --- Handle User Input ---
if prompt := st.chat_input(
    "Ask the LLM something (e.g., 'show me anime with high ratings', 'recommend me an anime')"

):
    # 1. Add user message to chat history and display it
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.write(prompt)



    # 2. Get LLM response
    with st.spinner("LLM is thinking..."):
        response = get_llm_response(prompt)
        try:
            llm_response = response["structured_response"]
        except KeyError as e:
            llm_response = ResponseClass(
                response_type="text",
                response_data="Failed to return a structured output, query could not be executed successfully.",
            )

    # 3. Add LLM response to chat history and display it
    st.session_state.messages.append({"role": "assistant", "content": llm_response})

    with st.chat_message("assistant"):
        # Check the type of content to display

        if llm_response.response_type == "text":
            st.write(llm_response.response_data)
        elif llm_response.response_type == "dataframe":
            # Convert the DataFrameDict to a pandas DataFrame
            df = pd.DataFrame(json.loads(llm_response.response_data))
            st.dataframe(df, use_container_width=True)
        elif llm_response.response_type == "image":
            for card in llm_response.response_data:
                # Handle both dictionary and ImageCard object formats
                if hasattr(card, "image_url"):
                    # It's an ImageCard object
                    st.image(card.image_url, caption=card.title)
                    st.subheader(card.title)
                    st.write(card.description)
                else:
                    # It's a dictionary
                    st.image(card["image_url"], caption=card["title"])
                    st.subheader(card["title"])
                    st.write(card["description"])
        else:
            # Fallback for unknown types
            st.error(f"Unsupported response type: {llm_response['response_type']}")


# --- Optional: Add a button to clear history ---
if st.sidebar.button("Clear Chat History"):
    st.session_state.messages = []
    st.rerun()  # Rerun the app to reflect the cleared state

In the next section we are going to add memory to the agent, but first, let me explain what's going on with @st.cache_resource.

The @st.cache_resource decorator tells Streamlit to store the result of the decorated function (initialized agent in this case). On subsequent runs of the script (e.g., when the user sends a new message), if the function is called again with the same inputs (which it is, in this case), Streamlit reuses the stored object instead of re-executing the function. Benefits of this include:

Speed: Avoids the time cost of re-creating the agent or memory object.
Cost: Avoids potentially re-running initialization code that might call external APIs (though the agent creation itself might be cheap, the principle applies).
State Preservation: Ensures you're interacting with the same memory object and agent instance across user interactions within a session. We also cache the initial system message invocation (invoke_system_message) to avoid sending it repeatedly.

Adding memory to the agent

For the last section, we are going to add memory to the agent. We are going to be making use of a simple in-memory checkpointer that comes with LangGraph. See its documentation for more detail, including how to use different persistence backends (e.g., SQLite or Postgres).

In llm_utils.py, modify the create_llm_agent and the get_model_response functions as follows:


# llm_utils.py
@st.cache_resource
def create_llm_agent(_memory):
    model = init_chat_model("gpt-4o-mini", model_provider="openai")
    return create_react_agent(
        model=model,
        tools=[read_pandas_dataframe],
        response_format=ResponseClass,
        checkpointer=_memory,
    )



def get_model_response(llm, user_input: str, config):
    user_message = HumanMessage(user_input)
    messages = [user_message]
    return llm.invoke({"messages": messages}, config)

We add an argument to the create_llm_agent function: _memory. We give it an underscore to indicate to @st.cache_resource that this argument does not need to be cached.

We add a config parameter to the get_model_response. This config parameter is used to handle multiple conversations at the same time in an application.
Config looks like this:
config = {"configurable": {"thread_id": "abc123"}}
The thread_id is a unique identifier for each conversation.

And now, make modifications to the chat.py.
Make the following import:

from langgraph.checkpoint.memory import MemorySaver

Then add the following after st.set_page_config:

#chat.py

config = {"configurable": {"thread_id": "abc123"}}  

@st.cache_resource
def get_memory():
    return MemorySaver()

memory = get_memory()

We use @st.cache_resource to avoid recreating the memory object on subsequent runs.

Now add the memory as the parameter when we call create_llm_agent:

llm = create_llm_agent(memory)

Next we modify invoke_system_message to pass the config as an argument:

@st.cache_resource
def invoke_system_message():
    system_message = get_system_message()
    llm.invoke({"messages": [system_message]}, config) # passing in config

Lastly, in the get_llm_response function, we pass the config to get_model_response:

def get_llm_response(user_prompt):
    """
    Calls the LLM with the user prompt and handles any exceptions.
    """

    try:
        response = get_model_response(llm, user_prompt, config) # Passing config
        return response
    except Exception as e:
        # Return a basic error response in the expected format
        return {"response_type": "text", "response_data": f"Error: {str(e)}"}

Try it out

The application is now fully functional and ready to be tried out.
Run the application with streamlit run chat.py

Here are some example prompts that I gave it:
Prompt: Hey, I'm looking to watch a romance anime, with around 12 episodes and with a bit of action. Do you have any suggestions?
Result:

Prompt: Give me more detail about the second title you suggested
Result:

Prompt: Are there any other anime similar to elfen lied?
Result:

Hmm, I think I'll watch Berserk😁.

Conclusion

I hope this article has helped you get a better understanding of how to use tools with agents. Being able to use tools with large language is definitely a game changer. I found it pretty fun to figure out. Happy tooling, folks!

Create a Smart RAG App with LangChain and Streamlit

Ngonidzashe Nzenze — Wed, 06 Nov 2024 13:47:54 +0000

Ever been reading through a PDF document and thought, "Hmm, if only there was a way I could quickly extract the relevant information"? That would save quite a lot of time. It's been a few years since large language models (LLMs) have been introduced, revolutionizing the way we interact with text data.

LLMs are trained on vast amounts of data from the internet and other text sources, making them highly effective at many general-purpose tasks. However, in certain cases, you might want to train or augment an LLM with your own data to make the responses more relevant to your needs. In this article, I’ll show you how to create an app that uses Retrieval-Augmented Generation (RAG) to answer questions specific to particular documents or web pages. But first...

What is RAG?

RAG stands for Retrieval-Augmented Generation. It is a type of natural language processing framework that combines the benefits of retrieval-based and generative models.

These models:

Retrieve relevant info from a database or knowledge graph
Use this information to generate a more accurate response

Pretty useful, especially when you're handling rare topics that the LLM may not know about. I took quite a lot of inspiration from Google's NotebookLM, but this is not going to be anywhere close to what NotebookLM does, just a gentle introduction to help you get a bit of an understanding of what's happening and maybe appreciate RAG.

With that, lets proceed with the...

Setup

For this application, we're going to be making use of the following packages:

langchain - to link us to the open ai large language models
chroma db - vector store we're gonna be using to store our information
streamlit - for the interface because of how easy it is to setup

Before we continue, let me give you a sneak peek at the final product:

You can find the the code for this application on my github.

Install the packages as follows:
pip install streamlit
pip install langchain langchain-community langchain-openai langchain-chroma

We're also going to need BeautifulSoup so install it as follows:
pip install bs4

The application has two databases. The vector database to store the documents and a relational database to store the chats. sqlite3 is simple to setup hence we are going to be using it in this tutorial. Feel free to make use of any other databases.

And with that, we're ready to begin.

Database design

Our relational database is going to have 3 tables:

The chat table - which will store the chat names
The sources table - which will store the different sources we have loaded into the vector store
The messages table - which will store the messages between the human and the AI

Create a file named create_relational_db.py and add the following code:

import sqlite3

# Connect to SQLite database (or create it if it doesn't exist)
conn = sqlite3.connect("doc_sage.sqlite")
cursor = conn.cursor()

# Create 'chat' table
cursor.execute(
    """
    CREATE TABLE IF NOT EXISTS chat (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        title TEXT NOT NULL,
        created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
        updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
    )
"""
)

# Create 'sources' table
cursor.execute(
    """
    CREATE TABLE IF NOT EXISTS sources (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        source_text TEXT,
        type TEXT DEFAULT "document",
        chat_id INTEGER,
        FOREIGN KEY (chat_id) REFERENCES chat(id)
    )
"""
)


# Create 'messages' table
cursor.execute(
    """
    CREATE TABLE IF NOT EXISTS messages (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        chat_id INTEGER NOT NULL,
        sender TEXT NOT NULL,
        content TEXT NOT NULL,
        timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
        FOREIGN KEY(chat_id) REFERENCES chat(id)
    );
"""
)


# Commit the transaction
conn.commit()


# Close the connection
conn.close()

print("Tables created successfully.")

The souces table stores the name of the document or the link address. The type field indicates whether the source is a document (file) or webpage. It can only have two values document or link with document as the default.

The messages table stores the messages between the user and the LLM. It is linked to the chat table by a foreign key on chat_id. The sender is either ai or user, which is a way to keep track of who send which message. The content is the actual content of the message from either the user or the ai.

Create the database and tables by running:
python create_relational_db.py

Now let us move on to the functions that are going to be operating on the database.

Create another file named db.py and add the following code:

import sqlite3

# Connect to SQLite database
def connect_db():
    return sqlite3.connect("doc_sage.sqlite")

# CRUD Operations for 'chat' table
def create_chat(title):
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute("INSERT INTO chat (title) VALUES (?)", (title,))
    chat_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return chat_id


def list_chats():
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM chat ORDER BY created_at DESC")
    chats = cursor.fetchall()
    conn.close()
    return chats


def read_chat(chat_id):
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM chat WHERE id = ?", (chat_id,))
    result = cursor.fetchone()
    conn.close()
    return result


def update_chat(chat_id, new_title):
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute(
        "UPDATE chat SET title = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
        (new_title, chat_id),
    )
    conn.commit()
    conn.close()


def delete_chat(chat_id):
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM chat WHERE id = ?", (chat_id,))
    conn.commit()
    conn.close()


def create_source(name, source_text, chat_id, source_type="document"):
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO sources (name, source_text, chat_id, type) VALUES (?, ?, ?, ?)",
        (name, source_text, chat_id, source_type),
    )
    conn.commit()
    conn.close()


def read_source(source_id):
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM sources WHERE id = ?", (source_id,))
    result = cursor.fetchone()
    conn.close()
    return result

def update_source(source_id, new_name, new_source_text):
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute(
        "UPDATE sources SET name = ?, source_text = ? WHERE id = ?",
        (new_name, new_source_text, source_id),
    )
    conn.commit()
    conn.close()


def list_sources(chat_id, source_type=None):
    conn = connect_db()
    cursor = conn.cursor()
    if source_type:
        cursor.execute(
            "SELECT * FROM sources WHERE chat_id = ? AND type = ?",
            (chat_id, source_type),
        )
    else:
        cursor.execute("SELECT * FROM sources WHERE chat_id = ?", (chat_id,))
    sources = cursor.fetchall()
    conn.close()
    return sources


def delete_source(source_id):
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM sources WHERE id = ?", (source_id,))
    conn.commit()
    conn.close()


# CRUD Operations for 'messages' table
def create_message(chat_id, sender, content):
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO messages (chat_id, sender, content) VALUES (?, ?, ?)",
        (chat_id, sender, content),
    )
    conn.commit()
    conn.close()


def get_messages(chat_id):
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT sender, content FROM messages WHERE chat_id = ? ORDER BY timestamp ASC",
        (chat_id,),
    )
    messages = cursor.fetchall()
    conn.close()
    return messages


def delete_messages(chat_id):
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM messages WHERE chat_id = ?", (chat_id,))
    conn.commit()
    conn.close()

The above code is responsible for all the operations on the sqlite3 database.

RAG Functions

In this section, we are going to create the functions that operate on the vector store, from loading documents to retrieving them and generating the responses. This is where most of the magic happens. Get some coffee because this section is rather long and a little more complicated🙂. But worry not, I will explain as simply as possible.

Create a file named vector_functions.py and add the following lines:

import os
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document
from langchain_text_splitters import CharacterTextSplitter
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_community.document_loaders import (
    TextLoader,
    CSVLoader,
    PyPDFLoader,
    Docx2txtLoader,
    UnstructuredHTMLLoader,
    UnstructuredMarkdownLoader,
)

import environ
env = environ.Env()

# reading .env file
environ.Env.read_env()

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=env("OPENAI_API_KEY"),
)

embeddings = OpenAIEmbeddings(
    api_key=env("OPENAI_API_KEY"),
)

text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)

In this example, I am making use of the python-environ module to handle the api keys so make sure you have it installed with:

pip install python-environ

Create an .env file and add your open ai key e.g:
OPENAI_API_KEY=Your-API-Key

Initialize the large language model with the ChatOpenAI class. In this example we are going to be making use of gpt-4o-mini. You can make use other models Like Anthropic's Claude. Make sure to refer to the documentation for more information.

OpenAIEmbeddings is a class that generates embeddings — a way to convert text into a numeric format so that AI models can process it more easily.

CharacterTextSplitter breaks up long text into smaller chunks to make it easier for the AI model to handle. chunk_size=1000 means each text chunk will contain up to 1000 characters. chunk_overlap=0 means there’s no overlap between chunks; each chunk is independent and contains a unique section of the text. This approach is often useful when working with long articles, documents, or books that need to be processed in parts.

Next we create a function that is going to be responsible for loading different doctypes:

def load_document(file_path: str) -> list[Document]:
    """
    Load a document from a file path.
    Supports .txt, .pdf, .docx, .csv, .html, and .md files.

    Args:
    file_path (str): Path to the document file.

    Returns:
    list[Document]: A list of Document objects.

    Raises:
    ValueError: If the file type is not supported.
    """
    _, file_extension = os.path.splitext(file_path)

    if file_extension == ".txt":
        loader = TextLoader(file_path)
    elif file_extension == ".pdf":
        loader = PyPDFLoader(file_path)
    elif file_extension == ".docx":
        loader = Docx2txtLoader(file_path)
    elif file_extension == ".csv":
        loader = CSVLoader(file_path)
    elif file_extension == ".html":
        loader = UnstructuredHTMLLoader(file_path)
    elif file_extension == ".md":
        loader = UnstructuredMarkdownLoader(file_path)
    else:
        raise ValueError(f"Unsupported file type: {file_extension}")

    return loader.load()

The load_document function returns a list of Document objects, which are later split into texts by a text splitter and then saved in the vector store.

Install the following packages that are required by some of the document loaders:
pip install pypdf unstructured docx2txt Markdown

Next we add the following code to create and load collections in our vector store:

# vector_functions.py

def create_collection(collection_name, documents):
    """
    Create a new Chroma collection from the given documents.

    Args:
    collection_name (str): The name of the collection to create.
    documents (list): A list of documents to add to the collection.

    Returns:
    None

    This function splits the documents into texts, creates a new Chroma collection,
    and persists it to disk.
    """

    # Split the documents into smaller text chunks
    texts = text_splitter.split_documents(documents)
    persist_directory = "./persist"

    # Create a new Chroma collection from the text chunks
    try:
        vectordb = Chroma.from_documents(
            documents=texts,
            embedding=embeddings,
            persist_directory=persist_directory,
            collection_name=collection_name,
        )
    except Exception as e:
        print(f"Error creating collection: {e}")
        return None

    return vectordb


def load_collection(collection_name):
    """
    Load an existing Chroma collection.

    Args:
    collection_name (str): The name of the collection to load.

    Returns:
    Chroma: The loaded Chroma collection.
    This function loads a previously created Chroma collection from disk.
    """
    persist_directory = "./persist"
    
    # Load the Chroma collection from the specified directory
    vectordb = Chroma(
        persist_directory=persist_directory,
        embedding_function=embeddings,
        collection_name=collection_name,
    )

    return vectordb



def add_documents_to_collection(vectordb, documents):
    """
    Add documents to the vector database collection.

    Args:
        vectordb: The vector database object to add documents to.
        documents: A list of documents to be added to the collection.
    This function splits the documents into smaller chunks, adds them to the
    vector database, and persists the changes.
    """

    # Split the documents into smaller text chunks
    texts = text_splitter.split_documents(documents)

    # Add the text chunks to the vector database
    vectordb.add_documents(texts)

    return vectordb

We have created three function; create_collection, load_collection and add_documents_to_collection.

The create_collection function receives the collection_name and documents as arguments. The documents argument is the list of Document objects loaded using the load_document function. After loading the documents, we split the texts, 1000 characters per split(this of course can be changed). We initialize a persist_directory which is where the collections are going to be stored. Finally, we create a vector store from the given documents.

The load_collection function loads any collection that was created and returns it.

The add_documents_to_collection lets us add new documents to a collection. As we are using the app, we start with an empty collection. When we upload a document, it is saved to the database using this function. The documents are first loaded using the load_document function. This function returns the list of Document objects. The system then loads a collection using the load_collection function and the documents are finally added to the collection using the add_documents_to_collection function.

Next, a retriever is required. This is a function that will return the relevant results based on the search. This retriever queries the vector store and returns the most relevant results:

# vector_functions.py

def load_retriever(collection_name, score_threshold: float = 0.6):
    """
    Create a retriever from a Chroma collection with a similarity score threshold.

    Args:
    collection_name (str): The name of the collection to use.
    score_threshold (float): The minimum similarity score threshold for retrieving documents.
                           Documents with scores below this threshold will be filtered out.
                           Defaults to 0.6.
    Returns:
    Retriever: A retriever object that can be used to query the collection with similarity
              score filtering.
    This function loads a Chroma collection and creates a retriever from it that will only
    return documents meeting the specified similarity score threshold.

    """

    # Load the Chroma collection
    vectordb = load_collection(collection_name)

    # Create a retriever from the collection with specified search parameters
    retriever = vectordb.as_retriever(
        search_type="similarity_score_threshold",
        search_kwargs={"score_threshold": score_threshold},
    )
    return retriever

The load_retriever function accepts two arguments: collection_name and score_threshold.

Some of the methods we can use to perform searches in the vector store are:

Similarity Search: Finds the items that are "closest" to your search queries. This closeness is measured by metrics such as cosine similarity or Euclidean distance. Usually you have to specify that you want k number of items.
Similarity Score Threshold: Returns results with a similarity score that is above a given threshold. This is useful when you want the most relevant matches without unrelated results.
Maximum Margin Ranking: This one focuses on maximizing the gap between relevant and irrelevant items. This approach helps in clearly ordering results, so the most relevant items appear at the top while less relevant ones are pushed lower. We are making use of the Similarity Score Threshold method, however, you can also make use of the other methods to see how well they perform.

And now we add the last function before moving on to the interface:

# vector_functions.py

def generate_answer_from_context(retriever, question: str):
    """
    Ask a question and get an answer based on the provided context.

    Args:
        retriever: A retriever object to fetch relevant context.
        question (str): The question to be answered.

    Returns:
        str: The answer to the question based on the retrieved context.
    """

    # Define the message template for the prompt
    message = """
    Answer this question using the provided context only.
    {question}

    Context:
    {context}
    """

    # Create a chat prompt template from the message
    prompt = ChatPromptTemplate.from_messages([("human", message)])

    # Create a RAG (Retrieval-Augmented Generation) chain
    # This chain retrieves context, passes through the question,
    # formats the prompt, and generates an answer using the language model
    rag_chain = {"context": retriever, "question": RunnablePassthrough()} | prompt | llm

    # Invoke the RAG chain with the question and return the generated content
    return rag_chain.invoke(question).content

The purpose of this function is to return an answer that the LLM generates using data retrieved by retriever. It accepts 2 arguments:

retriever - to find relevant text
question - a string containing the user's query.

We define a message template that structures how we want the question and context to look before we send them to the model. {question} and {context} are placeholders that are going to be filled later on with actual values.

ChatPromptTemplate.from_messages([("human", message)]) creates the a structured prompt using our template. This is useful as it guides the models response helping it generate meaningful output. You can read the documentation for more information.

Next, we create the RAG chain:

rag_chain = {"context": retriever, "question": RunnablePassthrough()} | prompt | llm

Lets breakdown what's happening:

{"context": retriever, "question": RunnablePassthrough()}
- This dictionary stores input components for the pipeline. We set the "context" to retriever, which is going to return the relevant text as context.
- "question" uses RunnablePassthrough(), meaning it passes the question along the chain without modifying it. You can get more information about how it works from the documentation.
| prompt
- This part pipes the output from the previous dictionary into the prompt object which formats these inputs into a structured prompt.
| llm
- The final stage sends the prompt to the LLM which then generates the response.

rag_chain.invoke(question) runs the entire chain, passing in the question and returning the answer based on the given context.

All that's left now is to create the interface and connect it with the different functions...

User Interface

The application has 2 pages:

The chats home page
The chat page

Lets start by creating the chats home page. Create a file named chats.py and add the following code:


import streamlit as st
import os, time, math
import requests
from bs4 import BeautifulSoup
from langchain_core.documents import Document
from db import (
    read_chat,
    create_chat,
    list_chats,
    delete_chat,
    create_message,
    get_messages,
    create_source,
    list_sources,
    delete_source,
)

from vector_functions import (
    load_document,
    create_collection,
    load_retriever,
    generate_answer_from_context,
    add_documents_to_collection,
    load_collection,
)

def chats_home():
    st.markdown(
        "<h1 style='text-align: center;'>DocSage🧙‍♂️</h1>", unsafe_allow_html=True
    )

    with st.container(border=True):
        col1, col2 = st.columns([0.8, 0.2])

        with col1:
            chat_title = st.text_input(
                "Chat Title", placeholder="Enter Chat Title", key="chat_title"
            )

        with col2:
            st.markdown("<br>", unsafe_allow_html=True)  # Add vertical space
            if st.button("Create Chat", type="primary"):
                if chat_title:
                    chat_id = create_chat(chat_title)
                    st.success(f"Created new chat: {chat_title}")
                    st.query_params.from_dict({"chat_id": chat_id})
                    st.rerun()
                else:
                    st.warning("Please enter a chat title")

    with st.container(border=True):
        st.subheader("Previous Chats")

        # get previous chats from db
        previous_chats = list_chats()

        # Pagination settings
        chats_per_page = 5
        total_pages = math.ceil(len(previous_chats) / chats_per_page)

        # Get current page from session state
        if "current_page" not in st.session_state:
            st.session_state.current_page = 1

        # Calculate start and end indices for the current page
        start_idx = (st.session_state.current_page - 1) * chats_per_page
        end_idx = start_idx + chats_per_page

        # Display chats for the current page
        for chat in previous_chats[start_idx:end_idx]:
            chat_id, chat_title = chat[0], chat[1]
            with st.container(border=True):
                col1, col2, col3 = st.columns([0.6, 0.2, 0.2])

                with col1:
                    st.markdown(f"**{chat_title}**")
                with col2:
                    if st.button("📂 Open", key=f"open_{chat_id}"):
                        st.query_params.from_dict({"chat_id": chat_id})
                        st.rerun()

                with col3:
                    if st.button("🗑️ Delete", key=f"delete_{chat_id}"):
                        delete_chat(chat_id)
                        st.success(f"Deleted chat: {chat_title}")
                        st.rerun()


        # Pagination controls
        col1, col2, col3 = st.columns([1, 2, 1])

        with col1:
            if st.button("Previous") and st.session_state.current_page > 1:
                st.session_state.current_page -= 1
                st.rerun()
        with col2:
            st.write(f"Page {st.session_state.current_page} of {total_pages}")
        with col3:
            if st.button("Next") and st.session_state.current_page < total_pages:
                st.session_state.current_page += 1
                st.rerun()



def main():
       chats_home()

if __name__ == "__main__":
    main()

The chat page is going to consist of a text input that allows us to create chats and a list of the chats we have created with pagination:

When a user adds a chat title and clicks the create chat button, the following code is executed:

if st.button("Create Chat", type="primary"):
    if chat_title:
        chat_id = create_chat(chat_title)
        st.success(f"Created new chat: {chat_title}")
        st.query_params.from_dict({"chat_id": chat_id})
        st.rerun()
    else:
        st.warning("Please enter a chat title")

If the user text input is not empty, the create_chat function is called with the chat_title as the argument. This function returns the id of the created chat. A success message is shown and we set the query_params using the from_dict method to the chat_id. This going to allow us to navigate to the chat page for the newly created chat. If the chat_title is empty, a warning is simply displayed.

Now to add the chat page, add the following code to chats.py:


def stream_response(response):
    """
    Stream a response word by word with a delay between each word.
    Args:
        response (str): The text response to stream
    Yields:
        str: Individual words from the response with a space appended
    Note:
        Adds a 50ms delay between each word to create a typing effect
    """

    # Split response into words and stream each one
    for word in response.split():
        # Yield the word with a space and pause briefly
        yield word + " "
        time.sleep(0.05)

The stream_response method is going to simulate the chat GPT model streaming response, writing each word after a 50ms delay rather that spilling out all the text in one go.

Next, add the function for the chat page:


def chat_page(chat_id):
    """
        Renders the main chats page where users can:
        - Create new chats with titles
        - View and manage previous chats
        - Navigate through paginated chat history

        The page displays a header, chat creation form, and list of existing chats
        with options to open each chat.
        """
        chat = read_chat(chat_id)

        if not chat:
            st.error("Chat not found")
            return

        # Retrieve messages from DB
        messages = get_messages(chat_id)

        # Display messages
        if messages:
            for sender, content in messages:
                if sender == "user":
                    with st.chat_message("user"):
                        st.markdown(content)
                elif sender == "ai":
                    with st.chat_message("assistant"):
                        st.markdown(content)
        else:
            st.write("No messages yet. Start the conversation!")

The function first retrieves the chat using the read_chat method. If the chat is not found, it returns an error message. If the chat exists, it uses the get_messages method to retrieve the messages exchanged between the user and the LLM. These messages are added to the page. If no messages are found, a simple text urging the user to start a conversation with the bot is displayed.

Now add the following code to the chat_page function:

def chat_page(chat_id):
    # Rest of the code ...

    # Add a text input for new messages
    prompt = st.chat_input("Type your message here...")
    if prompt:
        # Save user message
        create_message(chat_id, "user", prompt)

        # Display user message
        with st.chat_message("user"):
            st.markdown(prompt)

        # Get AI response
        # Load retriever for the chat context
        collection_name = f"chat_{chat_id}"
        if os.path.exists(f"./persist"):
            retriever = load_retriever(collection_name=collection_name)
        else:
            retriever = None

        # Ask question using the retriever
        response = (
            generate_answer_from_context(retriever, prompt)
            if retriever
            else "I need some context to answer that question."
        )

        # Save AI response
        create_message(chat_id, "ai", response)
    
        # Display AI response
        with st.chat_message("assistant"):
            st.write_stream(stream_response(response))

        st.rerun()

We use the streamlit chat_input to get the user's prompt. The prompt is then saved to the messages table using the create_message function and the message is displayed on the page using st.chat_message.

The next section gets the AI response as follows:

creates the collection name using the chat_id
checks if the persist directory exists which is where the collections are stored
If the persist directory exists, a retriever object is created using load_retriever method.
If the persist directory does not exist, the retriever object is set to None
If the retriever object is not None, the generate_answer_from_context method is used to get a response from the model. Remember the retriever first retrieves relevant texts and feeds them to the LLM. The LLM uses these texts to come up with an appropriate response.

The LLMs response is then saved to the model using the create_message function and written on to the page using st.write_stream and the stream_response function. Finally, the page is reloaded with the rerun method to display the text correctly.

Sidebar

In this section, we look at creating the sidebar as well as saving documents and links to the vector store. Add the following code to the chat_page function:


def chat_page(chat_id):
    # rest of the code ...

    # Sidebar
    with st.sidebar:
        # Button to return to the main chats page
        if st.button("Back to Chats"):
            st.query_params.clear()
            st.rerun()
            
        # Chat name
        st.subheader(f"{chat[1]}")

        # Documents Section
        st.subheader("📑 Documents")

        # Get all "document" type sources
        documents = list_sources(chat_id, source_type="document")

        if documents:
            # list the documents
            for doc in documents:
                doc_id = doc[0]
                doc_name = doc[1]
                col1, col2 = st.columns([0.8, 0.2])
                with col1:
                    st.write(doc_name)
                with col2:
                    if st.button("❌", key=f"delete_doc_{doc_id}"):
                        delete_source(doc_id)
                        st.success(f"Deleted document: {doc_name}")
                        st.rerun()
        else:
            st.write("No documents uploaded.")

        uploaded_file = st.file_uploader("Upload Document", key="file_uploader")

        if uploaded_file:

            # Save document content to database
            with st.spinner("Processing document..."):
                temp_dir = "temp_files"
                os.makedirs(temp_dir, exist_ok=True)
                temp_file_path = os.path.join(temp_dir, uploaded_file.name)
                
                with open(temp_file_path, "wb") as f:
                    f.write(uploaded_file.getbuffer())

                # Load document
                document = load_document(temp_file_path)
                
                # Create or update collection for this chat
                collection_name = f"chat_{chat_id}"

                if not os.path.exists(f"./persist/{collection_name}"):
                    vectordb = create_collection(collection_name, document)
                else:
                    vectordb = load_collection(collection_name)
                    vectordb = add_documents_to_collection(vectordb, document)

                # Save source to database
                create_source(uploaded_file.name, "", chat_id, source_type="document")

                # Remove temp file
                os.remove(temp_file_path)
                del st.session_state["file_uploader"]

                st.rerun()

The above code lists documents that were previously added to the vector store if there are any, and allows the user to delete them also.

When a user uploads a document, a spinner is shown whilst the document is added to the vector database. A temporary file directory is created if it is not created already and the file is saved to that directory. The document is then loaded using the load_document function, which returns the file as a list of Document objects. A collection is created or loaded, depending on whether it had been created before or not. If it had been created before, the documents are added to that collection using the add_documents_to_collection method. The create_source method is used to save the information in the relational database and finally the file is deleted from the temporary directory.

Now let us complete the function by adding link processing. In the chat_page function, add the following code:


def chat_page(chat_id):
    # rest of the code

    with st.sidebar:
        # rest of the code...

        # Links Section
        st.subheader("🔗 Links")

        # Display list of links
        links = list_sources(chat_id, source_type="link")

        if links:
            for link in links:
                link_id = link[0]
                link_url = link[1]
                col1, col2 = st.columns([0.8, 0.2])
                with col1:
                    st.markdown(f"[{link_url}]({link_url})")

                with col2:
                    if st.button("❌    ", key=f"delete_link_{link_id}"):
                        delete_source(link_id)
                        st.success(f"Deleted link: {link_url}")
                        st.rerun()
        else:
            st.write("No links added.")

        # Add new link
        new_link = st.text_input("Add a link", key="new_link")
        if st.button("Add Link", key="add_link_btn"):
            if new_link:
                with st.spinner("Processing link..."):
                    # Fetch content from the link
                    try:
                        headers = {

                            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36"
                        }
                        response = requests.get(new_link, headers=headers)
                        soup = BeautifulSoup(response.text, "html.parser")

                        # Check if the content was successfully retrieved
                        if response.status_code == 200 and soup.text.strip():
                            link_content = soup.get_text(separator="\n")
                        else:
                            st.toast(
                                "Unable to retrieve content from the link. It may be empty or inaccessible.",
                                icon="🚨",
                            )
                            return

                        # Save link content to vector store
                        documents = [
                            Document(
                                page_content=link_content, metadata={"source": new_link}
                            )
                        ]

                        collection_name = f"chat_{chat_id}"
                        
                        if not os.path.exists(f"./persist"):
                            create_collection(collection_name, documents)
                        else:
                            vectordb = load_collection(collection_name)
                            add_documents_to_collection(vectordb, documents)



                        # Save link to database
                        create_source(new_link, "", chat_id, source_type="link")
                        st.success(f"Added link: {new_link}")
                        del st.session_state["add_link_btn"]
                        st.rerun()
                    except Exception as e:
                        st.toast(

                            f"Failed to fetch content from the link: {e}", icon="⚠️"
                        )
            else:
                st.toast("Please enter a link", icon="❗")

The links are listed just like the documents. If a link is pasted into the input and the user clicks the Add Link button, a spinner is shown to show that the request is being processed. Using the requests library, the code attempts to fetch the content of the webpage. The User-Agent header is used to mimic a browser request since most websites will deny access to their content otherwise.

BeautifulSoup is then used to extract the content from the response. The code then checks if the response was successful (status code 200) and if the content is not empty. If not, a toast message is shown indicating that the link may be empty or inaccessible. If the content is valid, it is then saved as a Document object in a list with metadata indicating the source URL and just like before, it is added to the collection.

The full chat_page function looks as follows:

def chat_page(chat_id):

    """
    Display the chat page for a specific chat ID.

    This function handles displaying and managing an individual chat conversation, including:
    - Showing the chat history
    - Allowing users to send new messages
    - Streaming AI responses
    - Managing chat context through a vector store retriever
    Args:
        chat_id (int): The ID of the chat to display
    Returns:
        None
    """
    chat = read_chat(chat_id)
    if not chat:
        st.error("Chat not found")
        return

    # Retrieve messages from DB
    messages = get_messages(chat_id)

    # Display messages
    if messages:
        for sender, content in messages:
            if sender == "user":
                with st.chat_message("user"):
                    st.markdown(content)
            elif sender == "ai":
                with st.chat_message("assistant"):
                    st.markdown(content)
    else:
        st.write("No messages yet. Start the conversation!")

    # Add a text input for new messages
    prompt = st.chat_input("Type your message here...")
    if prompt:
        # Save user message
        create_message(chat_id, "user", prompt)

        # Display user message
        with st.chat_message("user"):
            st.markdown(prompt)

        # Get AI response
        # Load retriever for the chat context
        collection_name = f"chat_{chat_id}"
        if os.path.exists(f"./persist"):
            retriever = load_retriever(collection_name=collection_name)
        else:
            retriever = None

        # Ask question using the retriever
        response = (
            generate_answer_from_context(retriever, prompt)
            if retriever
            else "I need some context to answer that question."
        )

        # Save AI response
        create_message(chat_id, "ai", response)

        # Display AI response
        with st.chat_message("assistant"):
            st.write_stream(stream_response(response))

        st.rerun()

    # Sidebar for context
    with st.sidebar:
        # Button to return to the main chats page
        if st.button("Back to Chats"):
            st.query_params.clear()
            st.rerun()

        st.subheader(f"{chat[1]}")

        # Documents Section
        st.subheader("📑 Documents")

        # Display list of documents
        documents = list_sources(chat_id, source_type="document")
        if documents:
            for doc in documents:
                doc_id = doc[0]
                doc_name = doc[1]
                col1, col2 = st.columns([0.8, 0.2])
                with col1:
                    st.write(doc_name)
                with col2:
                    if st.button("❌", key=f"delete_doc_{doc_id}"):
                        delete_source(doc_id)
                        st.success(f"Deleted document: {doc_name}")
                        st.rerun()
        else:
            st.write("No documents uploaded.")

        uploaded_file = st.file_uploader("Upload Document", key="file_uploader")

        if uploaded_file:
            # Save document content to database
            with st.spinner("Processing document..."):
                temp_dir = "temp_files"
                os.makedirs(temp_dir, exist_ok=True)
                temp_file_path = os.path.join(temp_dir, uploaded_file.name)
                with open(temp_file_path, "wb") as f:
                    f.write(uploaded_file.getbuffer())

                # Load document
                document = load_document(temp_file_path)

                # Create or update collection for this chat
                collection_name = f"chat_{chat_id}"
                if not os.path.exists(f"./persist/{collection_name}"):
                    vectordb = create_collection(collection_name, document)
                else:
                    vectordb = load_collection(collection_name)
                    vectordb = add_documents_to_collection(vectordb, document)

                # Save source to database
                create_source(uploaded_file.name, "", chat_id, source_type="document")

                # Remove temp file
                os.remove(temp_file_path)
                del st.session_state["file_uploader"]
                st.rerun()

        # Links Section
        st.subheader("🔗 Links")

        # Display list of links
        links = list_sources(chat_id, source_type="link")

        if links:
            for link in links:
                link_id = link[0]
                link_url = link[1]
                col1, col2 = st.columns([0.8, 0.2])
                with col1:
                    st.markdown(f"[{link_url}]({link_url})")
                with col2:
                    if st.button("❌    ", key=f"delete_link_{link_id}"):
                        delete_source(link_id)
                        st.success(f"Deleted link: {link_url}")
                        st.rerun()
        else:
            st.write("No links added.")

        # Add new link
        new_link = st.text_input("Add a link", key="new_link")
        if st.button("Add Link", key="add_link_btn"):
            if new_link:
                with st.spinner("Processing link..."):
                    # Fetch content from the link
                    try:
                        headers = {

                            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36"
                        }
                        response = requests.get(new_link, headers=headers)
                        soup = BeautifulSoup(response.text, "html.parser")

                        # Check if the content was successfully retrieved
                        if response.status_code == 200 and soup.text.strip():
                            link_content = soup.get_text(separator="\n")
                        else:
                            st.toast(
                                "Unable to retrieve content from the link. It may be empty or inaccessible.",
                                icon="🚨",
                            )
                            return

                        # Save link content to vector store
                        documents = [
                            Document(
                                page_content=link_content, metadata={"source": new_link}
                            )
                        ]
                        collection_name = f"chat_{chat_id}"

                        if not os.path.exists(f"./persist"):
                            create_collection(collection_name, documents)
                        else:
                            vectordb = load_collection(collection_name)
                            add_documents_to_collection(vectordb, documents)

                        # Save link to database
                        create_source(new_link, "", chat_id, source_type="link")
                        st.success(f"Added link: {new_link}")
                        del st.session_state["add_link_btn"]
                        st.rerun()
                    except Exception as e:
                        st.toast(
                            f"Failed to fetch content from the link: {e}", icon="⚠️"
                        )
            else:
                st.toast("Please enter a link", icon="❗")

Now modify the main function as follows:

def main():
    """
    Main entry point for the chat application.
    Handles routing between the chats list page and individual chat pages:
    - If a chat_id is present in URL parameters, displays that specific chat
    - Otherwise shows the main chats listing page

    The function uses Streamlit query parameters to maintain state between page loads
    and determine which view to display.
    """

    query_params = st.query_params
    if "chat_id" in query_params:
        chat_id = query_params["chat_id"]
        chat_page(chat_id)
    else:
        chats_home()

if __name__ == "__main__":
    main()

What the main function is simply doing is checking if the chat_id is present in the URL parameter. If it is, it calls chat_page and displays the specific chat. If not, it simply shows the main chat listings page.

With that, the application is complete!!!

Run the app with:
streamlit run chats.py

Conclusion

RAG applications opens up possibilities for building really intelligent, context aware systems. Whether you're looking to build a knowledge-based chatbot, a document search tool, or any application requiring contextual responses, RAG is an ideal approach.

I hope you have found this article helpful!

Chat with your CSV: Visualize Your Data with Langchain and Streamlit

Ngonidzashe Nzenze — Wed, 17 May 2023 12:02:25 +0000

Large language models (LLMs) have become increasingly powerful and capable. These models can be used for a variety of tasks, including generating text, translating languages, and answering questions.

Langchain is a Python module that makes it easier to use LLMs. Langchain provides a standard interface for accessing LLMs, and it supports a variety of LLMs, including GPT-3, LLama, and GPT4All.

In this article, I will show how to use Langchain to analyze CSV files. We will use the OpenAI API to access GPT-3, and Streamlit to create a user interface. The user will be able to upload a CSV file and ask questions about the data. The system will then generate answers, and it can also draw tables and graphs.

Getting started

To get started, you will need to install langchain, openai, streamlit and python-environ. You can install them with pip:

pip install langchain openai streamlit python-environ tabulate

Setting up the agent

I have included all the code for this project on my github.

Setting up the agent is fairly straightforward as we're going to be using the create_pandas_dataframe_agent that comes with langchain. For those who might not be familiar, an agent is is a software program that can access and use a large language model (LLM). Agents are responsible for taking user input, processing it, and generating a response. They can also access and process data from other sources, such as databases, APIs and in this case, a csv file.

We are going to use the python-environ module to manage the API key.

Create a .env file and add the keys into it as below:

apikey=your_openai_api_key

Create a file named agent.py and add the following code:

# agent.py
from langchain import OpenAI
from langchain.agents import create_pandas_dataframe_agent
import pandas as pd

# Setting up the api key
import environ

env = environ.Env()
environ.Env.read_env()

API_KEY = env("apikey")


def create_agent(filename: str):
    """
    Create an agent that can access and use a large language model (LLM).

    Args:
        filename: The path to the CSV file that contains the data.

    Returns:
        An agent that can access and use the LLM.
    """

    # Create an OpenAI object.
    llm = OpenAI(openai_api_key=API_KEY)

    # Read the CSV file into a Pandas DataFrame.
    df = pd.read_csv(filename)

    # Create a Pandas DataFrame agent.
    return create_pandas_dataframe_agent(llm, df, verbose=False)

The create_agent function takes a path to a CSV file as input and returns an agent that can access and use a large language model (LLM). The function first creates an OpenAI object and then reads the CSV file into a Pandas DataFrame. Finally, it creates a Pandas DataFrame agent and returns it.

Now add the following function to agent.py:

#agent.py

# ...

def query_agent(agent, query):
    """
    Query an agent and return the response as a string.

    Args:
        agent: The agent to query.
        query: The query to ask the agent.

    Returns:
        The response from the agent as a string.
    """

    prompt = (
        """
            For the following query, if it requires drawing a table, reply as follows:
            {"table": {"columns": ["column1", "column2", ...], "data": [[value1, value2, ...], [value1, value2, ...], ...]}}

            If the query requires creating a bar chart, reply as follows:
            {"bar": {"columns": ["A", "B", "C", ...], "data": [25, 24, 10, ...]}}

            If the query requires creating a line chart, reply as follows:
            {"line": {"columns": ["A", "B", "C", ...], "data": [25, 24, 10, ...]}}

            There can only be two types of chart, "bar" and "line".

            If it is just asking a question that requires neither, reply as follows:
            {"answer": "answer"}
            Example:
            {"answer": "The title with the highest rating is 'Gilead'"}

            If you do not know the answer, reply as follows:
            {"answer": "I do not know."}

            Return all output as a string.

            All strings in "columns" list and data list, should be in double quotes,

            For example: {"columns": ["title", "ratings_count"], "data": [["Gilead", 361], ["Spider's Web", 5164]]}

            Lets think step by step.

            Below is the query.
            Query: 
            """
        + query
    )

    # Run the prompt through the agent.
    response = agent.run(prompt)

    # Convert the response to a string.
    return response.__str__()

The query_agent function is where all the magic happens. This function takes an agent(pandas dataframe agent) and a query as input and returns the response from the agent as a string. The function first creates a prompt for the agent. In this prompt we specify the kind of responses we want. I want the agent to return a string that will later be converted to a dictionary and based on the contents of that dictionary, the program will either render a graph, a table or a simple text response.

Setting up the streamlit interface

Streamlit is an open-source Python library that makes it easy to create web apps for machine learning and data science. Streamlit is designed to be quick and easy to use, and it can be used to create beautiful, interactive apps without any JavaScript or CSS knowledge. For more information, you can check out the documentation.

Streamlit is fairly easy to use. Create a file named interface.py and add the following:

import streamlit as st
import pandas as pd
import json

from agent import query_agent, create_agent


def decode_response(response: str) -> dict:
    """This function converts the string response from the model to a dictionary object.

    Args:
        response (str): response from the model

    Returns:
        dict: dictionary with response data
    """
    return json.loads(response)

The decode_response function is simply going to convert the response from the agent which is a string to a dictionary.

Add the following code to interface.py:

#interface.py

#...

def write_response(response_dict: dict):
    """
    Write a response from an agent to a Streamlit app.

    Args:
        response_dict: The response from the agent.

    Returns:
        None.
    """

    # Check if the response is an answer.
    if "answer" in response_dict:
        st.write(response_dict["answer"])

    # Check if the response is a bar chart.
    if "bar" in response_dict:
        data = response_dict["bar"]
        df = pd.DataFrame(data)
        df.set_index("columns", inplace=True)
        st.bar_chart(df)

    # Check if the response is a line chart.
    if "line" in response_dict:
        data = response_dict["line"]
        df = pd.DataFrame(data)
        df.set_index("columns", inplace=True)
        st.line_chart(df)

    # Check if the response is a table.
    if "table" in response_dict:
        data = response_dict["table"]
        df = pd.DataFrame(data["data"], columns=data["columns"])
        st.table(df)

This function takes a response dictionary as input and writes the response to the Streamlit app. It can be used to write answers, bar charts, line charts, and tables to the app.

It first checks if the response is an 'answer', that is if it is just a normal text response for questions like 'How many rows are in the document?'. If it is, the function writes the answer to the app.

The function then checks if the response is for a bar chart. If it is, the function creates a bar chart from the data in the response and writes the chart to the app.

The function then checks if the response is for a line chart. If it is, the function creates a line chart from the data in the response and writes the chart to the app.

The function then checks if the response is a table. If it is, the function creates a table from the data in the response and writes the table to the app.

Finally we'll create the initial interface. Add the following lines:

#interface.py

#...

st.title("👨‍💻 Chat with your CSV")

st.write("Please upload your CSV file below.")

data = st.file_uploader("Upload a CSV")

query = st.text_area("Insert your query")

if st.button("Submit Query", type="primary"):
    # Create an agent from the CSV file.
    agent = create_agent(data)

    # Query the agent.
    response = query_agent(agent=agent, query=query)

    # Decode the response.
    decoded_response = decode_response(response)

    # Write the response to the Streamlit app.
    write_response(decoded_response)

This code creates a Streamlit app that allows users to chat with their CSV files. The app first asks the user to upload a CSV file. The app then asks the user to enter a query. If the user clicks the "Submit Query" button, the app will query the agent and write the response to the app.

The app uses the following functions:

create_agent(): This function creates an agent from a CSV file.
query_agent(): This function queries an agent and returns the response.
decode_response(): This function decodes a response from an agent.
write_response(): This function writes a response to a Streamlit app.

Lets try it out!

Now in the console, start the application with streamlit run interface.py. This should open up a window in your browser that looks as follows:

For this tutorial, I'll be using data on books that can be found on kaggle. Upload your csv and let the prompting begin!

First query: Which book has the highest rating count?

Apparently Master of the game has the highest rating count, guess I should read it.

Second query: Tabulate the first 5 books. Include the title and the rating count columns only.

Note: I limited the columns to the title and rating columns so that we don't exceed the API token limitation.

Third query: Create a bar graph on the first 5 books

The above query will generate a bar graph. I specified the columns I want it to use to make easier for the model to understand my query. Pretty neat.

Forth query: Create a line graph of the first 5 books

In conclusion, Langchain and streamlit are powerful tools that can be used to make it easy for members to ask the LLMs about their data. The application allows them to get visualizations. This can be a valuable resource for members who want to learn more about their data or who need help making sense of it.

If you have any questions, feel free to reach out!

Speak Your Queries: How Langchain Lets You Chat with Your Database

Ngonidzashe Nzenze — Mon, 01 May 2023 12:47:20 +0000

Imagine effortlessly conversing with your database as if it were a close friend, asking questions and receiving instant, accurate responses. Welcome to the world of Langchain, a groundbreaking package that harnesses the power of LLMs(Large Language Models) to redefine the way you interact with your databases. In this tutorial, we'll explore how to seamlessly connect to a PostgreSQL database and start chatting with it using Langchain. Say goodbye to complex queries and embrace the future of database management – let's dive into the realm of conversational AI and revolutionize your data-driven tasks today!

What is Langchain?

Langchain is a user-friendly tool that helps connect advanced language software, called large language models (LLMs), with other sources of information, like databases. LLMs are amazing at understanding and generating human-like text, but to create truly powerful apps, we need to combine their abilities with other resources. Langchain makes this possible by allowing developers to easily link these language models with external data. This means that everyday people can benefit from smarter, more helpful applications that understand their needs and provide accurate information.

Getting started

I have included all the code for this project on my github.
Lets start by installing all the required packages first. Install langchain, openai, psycopg2 and python-environ with pip:

pip install langchain
pip install openai
pip install psycopg2
pip install python-environ

We are going to be using a postgresql database in this tutorial hence we're installing psycopg2. I am already assuming that you have postgresql installed on your device and have an open ai account. We are going to use the python-environ module to manage the API keys and database password.

Create a .env file and add the keys into it as below:

apikey=your_openai_api_key
dbpass=your_database_password

Setting up the database

Create a database named tasks in pgAdmin. Our tasks are going to consist the task name, the completion status which is either 1 or 0, the due date, completion date, and the task priority ranging from 0 to 3.

Create a file named db.py and add the following code to it:

import psycopg2

import environ
env = environ.Env()
environ.Env.read_env()

# Establish a connection to the PostgreSQL database
conn = psycopg2.connect(
    host='localhost',
    port=5433,
    user='postgres',
    password=env('dbpass'),
    database='tasks'
)

# Create a cursor object to execute SQL commands
cursor = conn.cursor()

# Create the tasks table if it doesn't exist
cursor.execute('''CREATE TABLE IF NOT EXISTS tasks
             (id SERIAL PRIMARY KEY,
             task TEXT NOT NULL,
             completed INTEGER,
             due_date DATE,
             completion_date DATE,
             priority INTEGER)''')

# Insert sample tasks into the tasks table
cursor.execute("INSERT INTO tasks (task, completed, due_date, completion_date, priority) VALUES (%s, %s, %s, %s, %s)",
               ('Finish homework', 1, '2023-05-01', None, 1))
cursor.execute("INSERT INTO tasks (task, completed, due_date, completion_date, priority) VALUES (%s, %s, %s, %s, %s)",
               ('Buy groceries', 0, '2023-05-03', None, 2))
cursor.execute("INSERT INTO tasks (task, completed, due_date, completion_date, priority) VALUES (%s, %s, %s, %s, %s)",
               ('Pay bills', 0, '2023-05-05', None, 3))
cursor.execute("INSERT INTO tasks (task, completed, due_date, completion_date, priority) VALUES (%s, %s, %s, %s, %s)",
               ('Clean house', 0, '2023-05-08', None, 4))
cursor.execute("INSERT INTO tasks (task, completed, due_date, completion_date, priority) VALUES (%s, %s, %s, %s, %s)",
               ('Exercise', 1, '2023-05-10', None, 5))

# Commit the changes and close the connection
conn.commit()
conn.close()

This file is going to create a table named tasks and insert demo tasks into it. Run the script with python db.py.

Setting up the chat

Now we are going to create the main console application. First create a file named main.py and import the relevant modules:

#main.py

from langchain import SQLDatabase, SQLDatabaseChain
from langchain.chat_models import ChatOpenAI

# Setting up the api key
import environ
env = environ.Env()
environ.Env.read_env()

API_KEY = env('apikey')

Under the hood, LangChain works with SQLAlchemy to connect to various types of databases. This means it can work with many popular databases, like MS SQL, MySQL, MariaDB, PostgreSQL, Oracle SQL, and SQLite. To learn more about connecting LangChain to your specific database, you can check the SQLAlchemy documentation for helpful information and requirements.

With postgres and psycopg2, the connection string is like this: postgresql+psycopg2://username:password@localhost/mydatabase.

Add the following to main.py:

# main.py
# ...

# Setup database
db = SQLDatabase.from_uri(
    f"postgresql+psycopg2://postgres:{env('dbpass')}@localhost:5433/tasks",
)

The code above sets up our connection to the postgresql database. Feel free to change it to your own configuration.

Add the following code:

# main.py

# ...

# setup llm
llm = ChatOpenAI(temperature=0, openai_api_key=API_KEY, model_name='gpt-3.5-turbo')

# Create query instruction
QUERY = """
Given an input question, first create a syntactically correct postgresql query to run, then look at the results of the query and return the answer.
Use the following format:

Question: "Question here"
SQLQuery: "SQL Query to run"
SQLResult: "Result of the SQLQuery"
Answer: "Final answer here"

{question}
"""

# Setup the database chain
db_chain = SQLDatabaseChain(llm=llm, database=db, verbose=True)

First we define our language model with:

llm = ChatOpenAI(temperature=0, openai_api_key=API_KEY)

We are going to be using the open ai language model.

Next, we create a basic query string, which will tell the model instructions on how we want it to present the data.

QUERY = """
Given an input question, first create a syntactically correct postgresql query to run, then look at the results of the query and return the answer.
Use the following format:

Question: "Question here"
SQLQuery: "SQL Query to run"
SQLResult: "Result of the SQLQuery"
Answer: "Final answer here"

{question}
"""

Finally we, set up the database chain with:

db_chain = SQLDatabaseChain(llm=llm, database=db, verbose=True)

We want to be able to prompt our database multiple times therefore I'm going to setup a while loop:

# main.py

# ...

def get_prompt():
    print("Type 'exit' to quit")

    while True:
        prompt = input("Enter a prompt: ")

        if prompt.lower() == 'exit':
            print('Exiting...')
            break
        else:
            try:
                question = QUERY.format(question=prompt)
                print(db_chain.run(question))
            except Exception as e:
                print(e)


get_prompt()

The get_prompt function will keep on asking for input and giving us the result over and over. We can only exit the loop if we type 'exit'. That is it for the setup, let's test it out and see how it does.

Now let's try it out!

How many tasks do i have?

I want a list of those tasks.

Wait a minute🤔, when should I have bought groceries?

How many incomplete tasks do I even have?

Oh I had forgot! I need to clean the house!!😨

In conclusion, we've explored the incredible potential of LangChain to revolutionize the way we interact with databases. By harnessing the power of large language models and making it easy to connect with various databases, LangChain enables users to engage in natural, conversational exchanges with their data. The accompanying screenshots throughout this article have demonstrated the simplicity and effectiveness of this innovative approach. As technology continues to evolve, LangChain stands as a testament to the endless possibilities that emerge when we seamlessly blend AI capabilities with practical, real-world applications. So, embrace the future of database management with LangChain and experience the ease and efficiency of conversing with your data.

Create a stopwatch and timer with Python and kivymd Part 1

Ngonidzashe Nzenze — Sat, 28 Jan 2023 13:38:47 +0000

In this article I am going to show you how to develop a stop watch and timer application with the kivymd framework. The first part of this tutorial will take you through creating the stopwatch and the second part through creating the the timer.

Final Product

The final application will look as below:

The UI

Lets get started with the user interface.

Prerequisites
For this project, I'm using:

kivy==2.1.0
kivymd==1.1.1

Make sure both are installed in a virtual environment.

Create two files:

main.py
main.kv

Inside main.py add the following code:

import kivy
from kivymd.app import MDApp
from kivymd.uix.boxlayout import MDBoxLayout
from kivymd.uix.dialog import MDDialog
from kivymd.uix.list import TwoLineIconListItem, IconLeftWidget
from kivy.properties import NumericProperty, StringProperty
from datetime import timedelta
from kivy.clock import Clock
kivy.require('2.1.0')


class MainApp(MDApp):

    def build(self):
        # Setting the theme
        self.theme_cls.theme_style = "Dark"
        self.theme_cls.primary_palette = "Gray"
        self.theme_cls.primary_hue = "200"


if __name__ == '__main__':
    app = MainApp()
    app.run()

And inside main.kv, add the following code:

#:kivy 2.1.0

MDScreenManager:
    MDScreen:
        name: 'main_screen'


        MDBottomNavigation:
            selected_color_background: "orange"
            text_color_active: "lightgrey"

            MDBottomNavigationItem:
                name: 'screen 1'
                text: 'Stop Watch'
                icon: 'timer'

                MDBoxLayout:
                    orientation: 'vertical'


                    MDLabel:
                        text: 'Stop Watch'
                        halign: 'center'
                        theme_text_color: "Custom"
                        text_color: 1, 1, 1, 1
                        font_style: 'H2'


            MDBottomNavigationItem:
                name: 'screen 2'
                text: 'Timer'
                icon: 'timer-sand'


                MDBoxLayout:
                    orientation: 'vertical'

                    MDTopAppBar:
                        right_action_items: [["plus", ]]
                        md_bg_color: app.theme_cls.bg_dark

                    MDLabel:
                        text: 'Timer'
                        halign: 'center'
                        theme_text_color: "Custom"
                        text_color: 1, 1, 1, 1
                        font_style: 'H2'

If you run this application right now, you will get the following:

The stopwatch

First, we are going to build the stopwatch. The stop watch is going to be fairly simple. It will consist of a Label, to show the elapsed time, and three buttons, one to reset the stopwatch, another to start, pause or stop the stopwatch and the last one to record the lap times. Lastly, it will have a list within a ScrollView widget in which we will be adding the recorded lap times.

Start by modifying main.kv as follows :

#...

# Replace the label with the text "Timer" with the following
MDLabel:
    id: stopwatch_lbl
    text: app.stopwatch_time
    halign: 'center'
    theme_text_color: "Custom"
    text_color: 1, 1, 1, 1
    font_style: 'H2'

# Add a scroll view and a list
MDScrollView:
    MDList:
        id: count_list

# Add a float layout
MDFloatLayout:
    pos_hint: {'center_x': .5}
    size_hint_y: .3
    size_hint_x: .5
    spacing: 50

    # Add three icon buttons
    MDIconButton:
        id: reset_btn
        icon: 'undo-variant'
        pos_hint: {'center_x': .2}
        on_press: app.reset_stopwatch()
        disabled: True


    MDIconButton:
        id: play_pause_btn
        icon: 'play'
        pos_hint: {'center_x': .5}
        on_press: app.start_or_stop_stopwatch()


   MDIconButton:
       id: record_lap_btn
       icon: 'timelapse'
       pos_hint: {'center_x': .8}
       on_press: app.time_lap()
       disabled: True

#...

main.py


#...

class MainApp(MDApp):

    stopwatch_time = StringProperty()

    milliseconds = NumericProperty()

    seconds = NumericProperty()

    minutes = NumericProperty()

    def build(self):
        # Setting the theme
        self.theme_cls.theme_style = "Dark"
        self.theme_cls.primary_palette = "Gray"
        self.theme_cls.primary_hue = "200"

    def on_start(self):
        self.stopwatch_time = "00:00:00"

    def start_or_stop_stopwatch(self):
        pass

    def reset_stopwatch(self):
        pass

    def time_lap(self):
        pass

#...

The latest changes have added icon buttons and a dynamic label to the interface. By setting the text of the label to app.stopwatch_time, we can change the text displayed on the interface by simply changing the value of the attribute stopwatch_time. Kivy properties make it easy to pass updates from the python side to the user interface and hence we used the StringProperty object.

In the above code we have also defined seconds, milliseconds and minutes as Numeric Properties which we are going to use to make updates to the stopwatch.

The on_start function is going to set the stopwatch to 00:00:00 when the application starts.

The start_or_stop_stopwatch function will start the stopwatch if it isn't running or stop the stopwatch if it is.

The reset_stopwatch function will reset the stopwatch to 00:00:00.

The time_lap funtion will measure the time between laps.

On running the application, you will get the following:

Now lets get our stopwatch to work.

main.py

# ...

class MainApp(MDApp):

    #...

    # Lap counter
    count = 1

    watch_started = False # Whether watch is running or not

    # store the last recorded lap time
    last_lap_time = {
        'minutes': 0,
        'seconds': 0,
        'milliseconds': 0
    }

    #...

count is going to keep track of the list items that are added when timing a lap which we'll see in action soon.

watch_started is going to keep track of whether or not the stop watch is running.

The dictionary last_lap_time contains data on the last recorded lap time which is updated every time we hit the lap button.

main.py

#...

class MainApp(MDApp):

    #...

    # add this function
    def get_string_time(self, dt):
        """Function to increment milliseconds and convert the time elapsed to string format to which the label is set"""
        self.increment_milliseconds()

        milliseconds = str(self.milliseconds)
        seconds = str(self.seconds)
        minutes = str(self.minutes)

        if len(milliseconds) < 2:
            milliseconds = '0' + milliseconds

        if len(seconds) < 2:
            seconds = '0' + seconds

        if len(minutes) < 2:
            minutes = '0' + minutes

        self.stopwatch_time = minutes + ":" + seconds + ":" + milliseconds

    # Modify start_or_stop_stopwatch to look as follows
    def start_or_stop_stopwatch(self):
        """Function to stop the stopwatch if it is not running otherwise stop it"""
        if self.watch_started:
            self.watch_started = False
            self.root.ids['record_lap_btn'].disabled = True
            self.root.ids['play_pause_btn'].icon = 'play'
            self.root.ids['reset_btn'].disabled = False
            Clock.unschedule(self.get_string_time) # Unschedule the get_string_time function
        else:
            self.watch_started = True
            self.root.ids['play_pause_btn'].icon = 'pause'
            self.root.ids['record_lap_btn'].disabled = False
            Clock.schedule_interval(self.get_string_time, 0.1) # schedule the get_string_time function to run every 10ms

    #...

The get_string_time function converts the timer from numbers to a string, that is, it gets the values of the milliseconds, seconds and minutes and converts them to a string. For the sake of display, it checks whether the values have two digits, and if they do not, it adds a 0 before the digit to make them two and hence we can always display our minutes, seconds and milliseconds with two digits.

We have modified the start_or_stop_stopwatch and now if the stopwatch was not running, it will change the status of watch_started to True to indicate that it is now running. It will then change the 'play' icon to the 'pause' icon, which will allow us to pause the stopwatch. The record_lap_button is then enabled so that we can record our laps. Finally, we will schedule the get_string_time function with kivy clock to run every 10 milliseconds so that the stopwatch is updated every 10 milliseconds.

If the stopwatch was running, it will stop/pause the stopwatch and then disable the lap button since we can't record laps if the watch is not running. It will then switch the icon to the 'play' icon. The reset button is then enabled as well to allow us to reset our stopwatch. Finally, it will unschedule the get_string_time to stop it from executing every 10 milliseconds.

main.py

class MainApp(MDApp):
    #...

    # Modify reset_stopwatch as follows
    def reset_stopwatch(self):
        """Set the stopwatch to 00:00:00"""
        if self.watch_started:
            Clock.unschedule(self.get_string_time)
        self.stopwatch_time = "00:00:00"
        self.milliseconds = 0
        self.seconds = 0
        self.minutes = 0

        # disable reset and lap buttons
        self.root.ids['reset_btn'].disabled = True
        self.root.ids['record_lap_btn'].disabled = True

        # Reset lap time
        self.last_lap_time['minutes'] = 0
        self.last_lap_time['seconds'] = 0
        self.last_lap_time['milliseconds'] = 0

    # Modify time_lap
    def time_lap(self):
        """Get the time between laps and add a list item to list"""
        lap_time = f"Count {self.count}: " + self.stopwatch_time
        list_item = TwoLineIconListItem(
            IconLeftWidget(
                icon="av-timer"
            ),
            text=lap_time,
            secondary_text=self.calculate_time_difference(),
            theme_text_color="Custom",
            text_color=(1, 1, 1, 1),
            secondary_theme_text_color="Custom",
            secondary_text_color=(1, 1, 1, 1)
        )

        self.root.ids['count_list'].add_widget(list_item, index=-1)

        self.count += 1

    # add the following function
    def increment_milliseconds(self):
        """Increment the milliseconds by 10ms"""
        self.milliseconds += 10

        if self.milliseconds == 100:
            self.increment_seconds()
            self.milliseconds = 0

    # add the following function
    def increment_seconds(self):
        """Increment the seconds by 1 second"""
        self.seconds += 1

        if self.seconds == 60:
            self.increment_minutes()
            self.seconds = 0

    # add the following function
    def increment_minutes(self):
        """Increment the minutes by 1 minute"""
        self.minutes += 1


    # add the following function
    def calculate_time_difference(self):
        """Calculate the time difference between the laps records
        """
        lap_time = timedelta(
            minutes=self.minutes,
            seconds=self.seconds,
            milliseconds=self.milliseconds
        ) - timedelta(
            minutes=self.last_lap_time['minutes'],
            seconds=self.last_lap_time['seconds'],
            milliseconds=self.last_lap_time['milliseconds']
        )

        lap_time = str(lap_time)[2:-3]

        lap_time = lap_time.split(':')

        lap_time = [i.split('.') for i in lap_time]

        minutes = int(lap_time[0][0])
        seconds = int(lap_time[1][0])
        milliseconds = int(lap_time[1][1])

        self.last_lap_time['minutes'] = self.minutes
        self.last_lap_time['seconds'] = self.seconds
        self.last_lap_time['milliseconds'] = self.milliseconds

        return f"{minutes}:{seconds}:{milliseconds}"
    #....

Ok, I know it's a lot of code but I'll do my best to explain it.

The reset_stopwatch method will check if the stopwatch is running. If it is, it will unschedule the get_string_time method. It will then set stopwatch_time to 00:00:00 which will update our label to '00:00:00'. Next it will set the milliseconds, seconds and minutes to 0 and also disable the reset and the lap buttons. Lastly it will set the minutes, seconds and milliseconds keys in the last_lap_time dictionary to zero.

The time_lap method gets the current count and creates a string with the count and current stopwatch time: lap_time = f"Count {self.count}: " + self.stopwatch_time. Next a two line icon list item is created with the lap_time as text. The secondary text is calculated by the calculate_time_difference method, which subtracts the last recorded lap time from the current time. The list item is then added to the the list.

Running the application now, we get our functional stopwatch:

All the code for this app is available in this Github repository.

In the next part of the tutorial, we look at how to create the timer so stay tuned!🙂

How to create a simple To-do list application with Kivymd

Ngonidzashe Nzenze — Tue, 19 Oct 2021 13:11:06 +0000

To-do list applications are a simple way to get started with learning different frameworks. I am going show you how to create one. With that, let's look at what the final application looks like on an android device:

Developing the application

Make sure your have installed kivy and kivymd in a virtual environment.

Create 3 files in the same directory, namely:

main.py - will to contain most of the application code and logic.
main.kv - will contain code to display the interface.
database.py - will contain all the database code.

Inside main.py, add the following code:

#main.py
from kivymd.app import MDApp

class MainApp(MDApp):
    def build(self):
        # Setting theme to my favorite theme
        self.theme_cls.primary_palette = "DeepPurple"

if __name__ == '__main__':
    app = MainApp()
    app.run()

In main.kv add the following code:

#main.kv

MDFloatLayout:
    MDLabel:
        id: task_label
        halign: 'center'
        markup: True
        text: "[u][size=48][b]My Tasks[/b][/size][/u]"
        pos_hint: {'y': .45}

    ScrollView:
        pos_hint: {'center_y': .5, 'center_x': .5}
        size_hint: .9, .8

        MDList:
            id: container

    MDFloatingActionButton:
        icon: 'plus-thick'
        on_release: app.show_task_dialog() #functionality to be added later
        elevation_normal: 12
        pos_hint: {'x': .8, 'y':.05}

If you run the application right now, you will get something like this:

Add Tasks

Next we are going to create a dialog box in which we will be able to add tasks. The dialog box will allow us to enter the task name and completion date:

main.py

#main.py

# add the following imports
from kivymd.uix.dialog import MDDialog
from kivymd.uix.boxlayout import MDBoxLayout
from kivymd.uix.picker import MDDatePicker
from datetime import datetime

class DialogContent(MDBoxLayout):
    """OPENS A DIALOG BOX THAT GETS THE TASK FROM THE USER"""
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # set the date_text label to today's date when useer first opens dialog box
        self.ids.date_text.text = str(datetime.now().strftime('%A %d %B %Y'))


    def show_date_picker(self):
        """Opens the date picker"""
        date_dialog = MDDatePicker()
        date_dialog.bind(on_save=self.on_save)
        date_dialog.open()

    def on_save(self, instance, value, date_range):
        """This functions gets the date from the date picker and converts its it a
        more friendly form then changes the date label on the dialog to that"""

        date = value.strftime('%A %d %B %Y')
        self.ids.date_text.text = str(date)

Now change the MainApp class inside main.py to look like this:

# main.py

#...

class MainApp(MDApp):
    task_list_dialog = None # Here
    def build(self):
        # Setting theme to my favorite theme
        self.theme_cls.primary_palette = "DeepPurple"

    # Add the below functions
    def show_task_dialog(self):
        if not self.task_list_dialog:
            self.task_list_dialog = MDDialog(
                title="Create Task",
                type="custom",
                content_cls=DialogContent(),
            )

        self.task_list_dialog.open()

    def close_dialog(self, *args):
        self.task_list_dialog.dismiss()

    def add_task(self, task, task_date):
        '''Add task to the list of tasks'''

        print(task.text, task_date)
        task.text = '' # set the dialog entry to an empty string(clear the text entry)

Now modify main.kv:

# main.kv

#...

# add the following
<DialogContent>:
    orientation: "vertical"
    spacing: "10dp"
    size_hint: 1, None
    height: "130dp"

    GridLayout:
        rows: 1

        MDTextField:
            id: task_text
            hint_text: "Add Task..."
            pos_hint: {"center_y": .4}
            max_text_length: 50
            on_text_validate: (app.add_task(task_text, date_text.text), app.close_dialog())

        MDIconButton:
            icon: 'calendar'
            on_release: root.show_date_picker()
            padding: '10dp'

    MDLabel:
        spacing: '10dp'
        id: date_text

    BoxLayout:
        orientation: 'horizontal'

        MDRaisedButton:
            text: "SAVE"
            on_release: (app.add_task(task_text, date_text.text), app.close_dialog())
        MDFlatButton:
            text: 'CANCEL'
            on_release: app.close_dialog()

Running our code so far:

Now we want to add list items to the screen. We are going to create a custom list item with a checkbox to the left and a delete icon to the right:

main.py

# main.py

#...

# Add these imports
from kivymd.uix.list import TwoLineAvatarIconListItem, ILeftBodyTouch
from kivymd.uix.selectioncontrol import MDCheckbox

# create the following two classes
class ListItemWithCheckbox(TwoLineAvatarIconListItem):
    '''Custom list item'''

    def __init__(self, pk=None, **kwargs):
        super().__init__(**kwargs)
        # state a pk which we shall use link the list items with the database primary keys
        self.pk = pk


    def mark(self, check, the_list_item):
        '''mark the task as complete or incomplete'''
        if check.active == True:
            # add strikethrough to the text if the checkbox is active
            the_list_item.text = '[s]'+the_list_item.text+'[/s]'
        else:
            # we shall add code to remove the strikethrough later
            pass

    def delete_item(self, the_list_item):
        '''Delete the task'''
        self.parent.remove_widget(the_list_item)



class LeftCheckbox(ILeftBodyTouch, MDCheckbox):
    '''Custom left container'''

Modify the add_task function in the MainApp class:

# main.py

#...

class MainApp(MDApp):
    #...
    def add_task(self, task, task_date):
        '''Add task to the list of tasks'''

        print(task.text, task_date)
        self.root.ids['container'].add_widget(ListItemWithCheckbox(text='[b]'+task.text+'[/b]', secondary_text=task_date))
        task.text = '' # set the dialog entry to an empty string(clear the text entry)

main.kv

# main.kv

# add the following code
<ListItemWithCheckbox>:
    id: the_list_item
    markup: True

    LeftCheckbox:
        id: check
        on_release: 
            root.mark(check, the_list_item)

    IconRightWidget:
        icon: 'trash-can-outline'
        theme_text_color: "Custom"
        text_color: 1, 0, 0, 1
        on_release:
            root.delete_item(the_list_item)

Running the application so far:

Ok, now to work on the code for the database. Inside database.py add the following code:

#database.py

import sqlite3

class Database:
    def __init__(self):
        self.con = sqlite3.connect('todo.db')
        self.cursor = self.con.cursor()
        self.create_task_table() #create the tasks table

    def create_task_table(self):
        """Create tasks table"""
        self.cursor.execute("CREATE TABLE IF NOT EXISTS tasks(id integer PRIMARY KEY AUTOINCREMENT, task varchar(50) NOT NULL, due_date varchar(50), completed BOOLEAN NOT NULL CHECK (completed IN (0, 1)))")
        self.con.commit()

    def create_task(self, task, due_date=None):
        """Create a task"""
        self.cursor.execute("INSERT INTO tasks(task, due_date, completed) VALUES(?, ?, ?)", (task, due_date, 0))
        self.con.commit()

        # GETTING THE LAST ENTERED ITEM SO WE CAN ADD IT TO THE TASK LIST
        created_task = self.cursor.execute("SELECT id, task, due_date FROM tasks WHERE task = ? and completed = 0", (task,)).fetchall()
        return created_task[-1]

    def get_tasks(self):
        """Get all completed and uncomplete tasks"""
        uncomplete_tasks = self.cursor.execute("SELECT id, task, due_date FROM tasks WHERE completed = 0").fetchall()
        completed_tasks = self.cursor.execute("SELECT id, task, due_date FROM tasks WHERE completed = 1").fetchall()
        # return the tasks to be added to the list when the application starts
        return completed_tasks, uncomplete_tasks



    def mark_task_as_complete(self, taskid):
        """Mark tasks as complete"""
        self.cursor.execute("UPDATE tasks SET completed=1 WHERE id=?", (taskid,))
        self.con.commit()

    def mark_task_as_incomplete(self, taskid):
        """Mark task as uncomplete"""
        self.cursor.execute("UPDATE tasks SET completed=0 WHERE id=?", (taskid,))
        self.con.commit()

        # return the task text
        task_text = self.cursor.execute("SELECT task FROM tasks WHERE id=?", (taskid,)).fetchall()
        return task_text[0][0]

    def delete_task(self, taskid):
        """Delete a task"""
        self.cursor.execute("DELETE FROM tasks WHERE id=?", (taskid,))
        self.con.commit()

    def close_db_connection(self):
        self.con.close()

The code above allows us to create, delete and modify tasks in the database.

Now to join this with the application interface:

main.py

#main.py

#...

# add import
from database import Database
# Initialize db instance
db = Database()

# Modify the ListItemWithCheckbox class
class ListItemWithCheckbox(TwoLineAvatarIconListItem):
    #...
    def mark(self, check, the_list_item):
        '''mark the task as complete or incomplete'''
        if check.active == True:
            the_list_item.text = '[s]'+the_list_item.text+'[/s]'
            db.mark_task_as_complete(the_list_item.pk)# here
        else:
            the_list_item.text = str(db.mark_task_as_incomplete(the_list_item.pk))# Here

    def delete_item(self, the_list_item):
        '''Delete the task'''
        self.parent.remove_widget(the_list_item)
        db.delete_task(the_list_item.pk)# Here


# Modify the MainApp class
class MainApp(MDApp):
    #...

    # add this entire function
    def on_start(self):
        """Load the saved tasks and add them to the MDList widget when the application starts"""
        try:
            completed_tasks, uncomplete_tasks = db.get_tasks()

            if uncomplete_tasks != []:
                for task in uncomplete_tasks:
                    add_task = ListItemWithCheckbox(pk=task[0],text=task[1], secondary_text=task[2])
                    self.root.ids.container.add_widget(add_task)

            if completed_tasks != []:
                for task in completed_tasks:
                    add_task = ListItemWithCheckbox(pk=task[0],text='[s]'+task[1]+'[/s]', secondary_text=task[2])
                    add_task.ids.check.active = True
                    self.root.ids.container.add_widget(add_task)
        except Exception as e:
            print(e)
            pass

    # Modify the add_task function
    def add_task(self, task, task_date):
        '''Add task to the list of tasks'''

        # Add task to the db
        created_task = db.create_task(task.text, task_date)# Here

        # return the created task details and create a list item
        self.root.ids['container'].add_widget(ListItemWithCheckbox(pk=created_task[0], text='[b]'+created_task[1]+'[/b]', secondary_text=created_task[2]))# Here
        task.text = ''

And with that, we're done!

Packaging for android

I have included all the code on github, including the spec file I used to generate the apk.

A few changes are required so that we can create an android application. Edit main.py as follows:

#...

from kivymd.uix.pickers import MDDatePicker # Here, instead of kivymd,uix.picker

# add the following just under the imports
if platform == "android":
    from android.permissions import request_permissions, Permission
    request_permissions([Permission.READ_EXTERNAL_STORAGE, Permission.WRITE_EXTERNAL_STORAGE])

The above code will prompt the user to allow the application to access storage.

Main changes in the buildozer spec file are as follows:

requirements = python3, kivy==2.1.0, https://github.com/kivymd/KivyMD/archive/master.zip,sdl2_ttf==2.0.15,pillow,android

And

android.permissions = WRITE_EXTERNAL_STORAGE

That's all for this tutorial. I hope you enjoyed it.

Cover Photo by Glenn Carstens-Peters on Unsplash

Created an anime data scraping application with kivyMD and BS4

Ngonidzashe Nzenze — Fri, 20 Aug 2021 18:27:02 +0000

I'm a big fan of anime and as any anime lover, I always want to be up to date with my favorite anime. I want to know whether or not a new episode has been aired and usually, I'm always checking these anime sites for updates. But what often occurs is that I end up getting glued to the site discovering more and more anime until I realize 'Oh, it's been three hours!'😅

I decided to create an application that just keeps me up to date with the few anime that I want to watch. It was a pretty good way to get up to speed with web scraping and working with some of those kivymd components.

Final Result

The application is pretty simple. First add the url of the anime on https://gogoanime.vc then navigate back to the 'Anime Update' screen and refresh by swiping down. This will begin the process of scraping data from the site and will display a list of the anime, with an image to the right of each list item and data on the anime such as the name, number of episodes and whether it has finished airing or not.

How it's done

The setup

First create a virtual environment. Inside it, install kivy, kivymd, beautifulsoup4, lxml and requests with the following commands:

pip install kivy
pip install kivymd
pip install bs4
pip install requests
pip install lxml

We are going to use the lxml parser with BeautifulSoup4

Now create 3 files in the same directory:

main.py - to contain the main application code
main.kv - to contain the interface code
scrap.py - this file will contain all the code responsible for scraping.

Lets get coding!

Inside main.py, add the following lines

from kivymd.app import MDApp
from kivy.uix.screenmanager import ScreenManager
from kivymd.uix.screen import MDScreen

class ScreenManagement(ScreenManager):
    pass

class MainWindow(MDScreen):
    pass

class AddUrlScreen(MDScreen):
    pass

class MainApp(MDApp):
    def build(self):
        self.theme_cls.primary_palette = "DeepPurple"

if __name__ == "__main__":
    app = MainApp()
    app.run()

Here I'm just creating the skeleton🦴 of the application. It will have two screens, the main screen responsible for displaying the anime information and the url screen where we can add the urls for the anime we want to scrap. I like the color purple and so I've set self.theme_cls.primary_palette to "DeepPurple". You can check out the kivymd documentation for more details.

Next, inside main.kv, add the following code:

ScreenManagement:
    MainWindow:
    AddUrlScreen:

<MainWindow>:
    name: 'mainscreen'

    #toolbar
    MDBoxLayout:
        orientation: 'vertical'

        MDTopAppBar:
            pos_hint: {'top': 1}
            title: 'AnimeUpdate'
            right_action_items: [["link-variant-plus", lambda x: lambda x:app.open_settings_screen()]]
        MDLabel:

Here we are adding the two screens we created to the screen manager. We then define the 'MainWindow' screen. Then we give it a name so that we can reference it later. We then add a tool bar with a widget on the right. The widget will call the app.open_settings_screen() function which we will create later. The 'MDLabel' is just a placeholder for now. If you run the application now you'll get the following:

Next, we add the refresh layout. Modify the <MainWindow> class in main.kv to look as follows:

<MainWindow>:
    name: 'mainscreen'

    #toolbar
    MDBoxLayout:
        orientation: 'vertical'

        MDTopAppBar:
            pos_hint: {'top': 1}
            title: 'AnimeUpdate'
            right_action_items: [["link-variant-plus", lambda x: lambda x:app.open_settings_screen()]]

        # add these lines
        MDScrollViewRefreshLayout:
            id: refresh_layout
            refresh_callback: root.refresh_callback
            root_layout: root

            MDGridLayout:
                id: box
                adaptive_height: True
                cols: 1

In the above code, we are adding the MDScrollViewLayout which is going to allow us to refresh the screen by swiping down from the top. Notice we deleted MDLabel. We are also adding a MDGridLayout which is going to contain the list items.

Now modify the MainWindow class in main.py to look as follows:

class MainWindow(MDScreen):
    def refresh_callback(self, *args):
        print("Refreshing...")

We define a function called refresh_callback which is going to be called when we swipe down. For now it is going to print 'Refreshing...' in the terminal.

Save and run the application with python main.py in the terminal and then swipe down on the screen:

Next we are going to create the url screen. Inside main.kv, add the following code:

<AddUrlScreen>:
    name: 'addurl'
    id: addurl

    # toolbar
    MDBoxLayout:
        orientation: 'vertical'

        MDTopAppBar:
            pos_hint: {'top': 1}
            title: 'Settings'
            left_action_items: [["keyboard-backspace", lambda x: app.return_to_main_window()]]

        MDFloatLayout:
            size_hint: 1, .9
            id: linkscreen

            MDTextField:
                id: linkinput
                size_hint: .8, None
                pos_hint: {'center_x': .5, 'y': .9}
                hint_text: 'Add Url'
                mode: 'rectangle'
                text_validate_unfocus: False
                on_text_validate: root.add_url(linkinput.text)

            ScrollView:
                md_bg_color: app.theme_cls.primary_color
                pos_hint: {'center_x': .5, 'y': .1}
                size_hint: .9, .8

                MDList:
                    id: linklist

AddUrlScreen has a toolbar but this time, the widget is on the left. This widget will return us to the main screen by calling the app.return_to_main_window() function which we are going to define next. The url screen has a text field. We have set text_validate_unfocus to False so that the text field does not lose focus when we press enter. When we press enter, the text inside the text field is going to be fed to the function root.add_url() which we are also going to define next.

Modify main.py as follows:

#...
from kivy.uix.screenmanager import ScreenManager, CardTransition # new import
#...

class AddUrlScreen(MDScreen):
    def add_url(self, text):
        """Add the url to list and save the url in shelve file"""
        self.ids.linkinput.focus = True
        self.ids.linkinput.text = ''
        print(text)

class MainApp(MDApp):
    def build(self):
        self.theme_cls.primary_palette = "DeepPurple"
        self.root.transition= CardTransition() # new line

    #define methods to switch between screens
    def open_settings_screen(self):
        """open setting window"""
        self.root.current = 'addurl'
        self.root.transition.direction = 'down'

    # new method
    def return_to_main_window(self):
        self.root.current = 'mainscreen'
        self.root.transition.direction = 'up'

Here we import the CardTransition which is one of the many screen transition animations. In the AddUrlScreen class, we also add the add_url method which right now only displays the text in the terminal. In the MainApp class, we set the screen transition to CardTransition and add methods to switch between windows.

Now we are going to define a custom list item that gets added to the url screen whenever we press enter in the text field:

In main.kv add:

<CustomListItem>:
    IconLeftWidget:
        icon: "web"

    IconRightWidget:
        icon: "trash-can"
        theme_text_color: "Custom"
        text_color: 1, 0, 0, 1
        on_release: root.delete_item(root.text)

Modify main.py:

from kivymd.uix.list import OneLineAvatarIconListItem, ThreeLineAvatarIconListItem, ImageLeftWidget

# new class
class CustomListItem(OneLineAvatarIconListItem):
    def delete_item(self, text):
        """Delete list item"""
        self.parent.remove_widget(self)

# modify AddUrlScreen
class AddUrlScreen(MDScreen):
    def add_url(self, text):
        """Add the url to list"""
        self.ids.linklist.add_widget(CustomListItem(text=text)) # new line
        self.ids.linkinput.focus = True
        self.ids.linkinput.text = ''
        # removed print line

We have imported three items some of which we are going to use later on. Create the CustomListItem class which inherits from the OneLineAvatarIconListItem class. The CustomListItem has an icon on the left and a delete icon on the right. If you run the application now, you can add and delete items:

Now we are going to set up a way to save the urls we have added so that we don't have to keep adding the urls everytime we want to get info on an anime. To do that, we are going to make use of the python shelve module. Modify main.py as follows:

import shelve, os # import shelve and os

# modify CustomListItem to delete item from shelve file
class CustomListItem(OneLineAvatarIconListItem):
    def delete_item(self, text):
        """Delete list item"""
        with shelve.open('./save_files/mydata')as shelf_file:

            url_list = shelf_file['url_list']
            url_list.remove(str(text))
            shelf_file['url_list'] = url_list

        self.parent.remove_widget(self)

# modify the AddUrlScreen class
class AddUrlScreen(MDScreen):
    def add_url(self, text):
        """Add the url to list and save the url in shelve file"""
        self.ids.linklist.add_widget(CustomListItem(text=text))
        self.ids.linkinput.focus = True
        self.ids.linkinput.text = ''

        # saving to shelve file
        with shelve.open('./save_files/mydata') as shelf_file:
            url_list = shelf_file['url_list']
            url_list.append(str(text))
            shelf_file['url_list'] = url_list

    def on_pre_enter(self):
    '''Load the shelve file with list item from shelve file'''
        try:
            with shelve.open('./save_files/mydata') as shelf_file:
                self.ids.linklist.clear_widgets()
                for item in shelf_file['url_list']:
                    self.ids.linklist.add_widget(CustomListItem(text=item))

        except KeyError:
            with shelve.open('./save_files/mydata') as shelf_file:
                shelf_file['url_list'] = []



class MainApp(MDApp):
    #.....

    # add this function to create two directories at start up
    def on_start(self):
        try:
            os.mkdir('images')
            os.mkdir('save_files')
        except:
            pass

Okay, I know that's a lot of code😅 but let me try my best to explain what's going on here. We modify CustomListItem so that when we delete this widget, we also delete the data we has saved in the shelve file. In AddUrlScreen, we add code to save the data in a shelve file in a list called url_list. So every time we press enter in the text field, the text is saved to the shelve file inside a folder named save_files which is in our current working directory. The on_pre_enter method allows us to do things when we are navigating to AddUrlScreen. It will load the saved urls and add CustomListItem widgets. The on_start function creates two folders when the application is started.

Web scraping with beautifulsoup4

I will not get into the details of beautifulsoup4. You can find more information about it in the documentation. The site we are going to be scrapping is gogoanime. If you visit this site, enter an anime name in the search field and press enter, you will be redirected to the results page. Here if you click on an anime, you will be directed to a page with details of that anime. The url will look like this: https://gogoanime.vc/category/name-of-anime.

This is the type of url that our application is going to be using to get details. Right click on the site and click 'inspect'. This will open a small section that allows you look at the details of certain elements like the html used to create them and their classes and so forth. We are going to be selecting the image, the anime name, the status and the number of episodes. There are many ways to select items using beautifulsoup4, I used the css selectors.

Inside scrap.py add the following code:

import os,sys
import requests
import shelve
from bs4 import BeautifulSoup

def download_webpage(url):
    try:
        # use requests to get the url text
        res = requests.get(url)
        res.raise_for_status()

        # parse the text to BeautifulSoup
        get_soup_text = BeautifulSoup(res.text, features='lxml')

        # download the image
        print("Downloading")
        download_details(get_soup_text)
    except Exception as e:
        print(e)

def download_details(soup_text):
    # get the anime title
    anime_title = soup_text.select('.anime_info_episodes > h2')[0].getText()
    # get the number of episodes
    episodes = soup_text.select('.anime_video_body ul li .active')[0].get('ep_end')
    # get the status
    completed = True if soup_text.find(title = 'Completed Anime') != None else False
    ongoing = True if soup_text.find(title = 'Ongoing Anime') != None else False
    # get the image
    image_elements = soup_text.select('.anime_info_body_bg img')

    if image_elements == []:
        pass
    else:
        # get image source
        image_url = image_elements[0].get('src')
        image = requests.get(image_url)
        image.raise_for_status()

    ## Save details to shelve file
    with shelve.open('./save_files/mydata') as shelf_file:
        file_name = os.path.basename(image_url)

        shelf_file[anime_title] = {
            'episodes': episodes,
            'completed': completed,
            'ongoing': ongoing,
            'image': file_name
            }

    # save the image
    if os.path.exists(f'./images/{file_name}') != True:
        image_file = open(os.path.join('images', file_name), 'wb')
        for chuck in image.iter_content(100000):
            image_file.write(chuck)
        image_file.close()
    else:
        pass

I know, it's a lot of code but I promise we're almost done. This code has two functions, download_webpage and download_details. download_webpage uses requests to download the text from the given url. The text is then parsed to BeautifulSoup which enables us to select the elements. download_details then takes this parsed text and selects the elements(name of anime, number of episodes etc). We use requests again to get the image source and download the image and save it in a folder named images.

And that is that for web scraping. Now to finish of the application.

Finishing Off the application

Modify main.py:

#...
from scrap import download_webpage # new import
from kivy.clock import Clock # new import

# modify MainWindow class
class MainWindow(MDScreen):
    #...

    # add new method
    def get_anime_info(self):
        '''Get the anime info and creates a list item widget and adds it to screen'''
        # open shelve files and get the urls
        with shelve.open('./save_files/mydata') as shelf_file:
            url_list = shelf_file['url_list']

        # download data from each url
        for url in url_list:
            download_webpage(url)

        # after downloading the data and saving it shelve file get the data and display it
        with shelve.open('./save_files/mydata') as shelf_file:
            for key in shelf_file.keys():
                if key != 'url_list':
                    print(key)
                    anime = shelf_file[key]
                    episodes = anime['episodes']
                    completed = anime['completed']
                    image = anime['image']

                    anime_complete = 'completed' if completed else 'ongoing'
                    # create a list item with the data
                    list_item = ThreeLineAvatarIconListItem(text=key, secondary_text=f"[b]Status:[/b] {anime_complete}", tertiary_text=f"[b]Episodes:[/b] {episodes}")
                    #add image to the list item
                    list_item.add_widget(ImageLeftWidget(source=f"./images/{image}"))

                    # finally add the list item to screen
                    self.ids.box.add_widget(list_item)

Import the scraping function from scrap.py and Thread from the threading module which we are going to use later. Next we add a new method to MainWindow. This method is going to be responsible for downloading and displaying the anime information every time we refresh. First, it is going to get the urls that were saved in the shelve file. It will pass these urls to the download_webpage function. This function will download all the anime data and save the information in the shelve file. Afterwards, the information is extracted from the shelve file and finally, list items are created and added to the screen.

Now for the final addition to get everything working.

#main.py

# Modify MainWindow
class MainWindow(MDScreen):
    # modify refresh_callback
    def refresh_callback(self, *args):
        print("Refreshing...")

        # new addition
        def refresh_callback():
            self.ids.box.clear_widgets()

            # call the get_anime_info method
            self.get_anime_info()
            self.ids.refresh_layout.refresh_done()

        Clock.schedule_once(refresh_callback, 1)

We modify main.py adding a refresh_callback function inside the other refresh_callback function. Now when we refresh(swipe down), the function is called and the anime data is downloaded and displayed:

The application is complete! Now you can be up to date with your favorite anime. Now go, watch some one punchman😎.

Disclaimer: this is for pure educational purpose only. Do not use this in any way that may be deemed as attacking gogoanime.

Full code available here along with the buildozer spec file.

How to use Google Colab to package KivyMD applications with buildozer.

Ngonidzashe Nzenze — Sat, 14 Aug 2021 15:12:45 +0000

Let's face it, packaging your kivy application for android can be pretty rough. First you create the app, then you install buildozer only to realize that python-for-android doesn't even run on windows! Now you're looking into installing a virtual Linux distribution. If you already use Linux, well lucky you, but if not, you are probably pounding your keyboard in frustration.

Is there a better way to do it? The answer is yes. In this article I'm going to show you how to package an application for android using Google Colab.

First thing's first, let's code up our application. I assume you already have kivy and kivymd installed in a virtual environment.

Create two files:

main.py
main.kv

Inside main.py, add the following code:

from kivymd.app import MDApp
from kivymd.uix.list import OneLineIconListItem, IconLeftWidget

class MainApp(MDApp):
    def build(self):
        self.theme_cls.primary_palette = "Purple"

    def add_item(self, text):
        new_list_item = OneLineIconListItem(text=text)
        new_list_item.add_widget(
            IconLeftWidget(icon = "language-python")
            )
        self.root.ids.listcontainer.add_widget(new_list_item)
        self.root.ids.listinput.text = ''


if __name__ == "__main__":
    app = MainApp()
    app.run()

Inside main.kv add the following lines:

MDBoxLayout:
    orientation: 'vertical'
    id: box
    MDTopAppBar:
        title: 'SampleApp'
        pos_hint: {'top': 1}

    MDTextField:
        id: listinput
        hint_text: 'Add Item'
        mode: 'rectangle'
        size_hint_x: .9
        pos_hint: {'center_x': .5} 
        text_validate_unfocus: False
        on_text_validate: app.add_item(listinput.text)

    ScrollView:
        MDList:
            id: listcontainer

This is a simple application that gets text from the user and creates a list item and adds it to the list.

When we run our application, it looks like this:

Now for packaging!📦

Open google colab on your web browser and create a new notebook or you can use my notebook.

Now create a cell and run the following lines:

!sudo apt update
!sudo apt install -y git zip unzip openjdk-17-jdk python3-pip autoconf libtool pkg-config zlib1g-dev libncurses5-dev libncursesw5-dev libtinfo5 cmake libffi-dev libssl-dev
!pip3 install --user --upgrade Cython==0.29.33 virtualenv

Luckily for you, the developers of kivymd decided to make the process easier so we can install buildozer dependencies in one fell swoop. Create a cell in your notebook and add the following lines and execute:

# git clone, for working on buildozer
!git clone https://github.com/kivy/buildozer
%cd buildozer
!python setup.py build
!pip install -e .
%cd ..

And now, everything is ready for the packaging process.

Upload your files to colab by clicking on the files icon and then right clicking and clicking upload.(Uuuh what?😫)

After the files have been uploaded, run the following command:

!buildozer init

This will create a file named buildozer.spec. Double click the file to edit it. You can change the application name and the package name. I changed mine as follows:

Next, scroll down to the requirements and change them as follows:

requirements = python3, kivy==2.1.0, kivymd==1.1.1, sdl2_ttf==2.0.15, pillow

Note: If your application requires special permissions such as internet or storage, scroll down to android.permissions:

Uncomment the android.permissions line and add the permissions. For a full list of all the available permissions, check out this link.

Now to package our application. Run the following command:

!buildozer -v android debug

This process can take up to 20 minutes so you might need a little bit of patience. In the mean time you could skip to the bonus section below or watch an episode of my hero academia(I won't judge😉).

After successful execution, the apk file is saved in the bin folder:

Download the apk, transfer it over to your android device and install it!

The application works!👏

Full Code is available HERE.

Bonus section(Not really)

Debugging our application.

Packaging applications this way is faster than having to go through installing all the requirements on your PC but this does have a draw back. How do you debug the application? Imagine successfully installing the application and trying to run it only for it to crash. How do you know what the problem is? You can't run buildozer -v android deploy run logcat which displays the logs so that you can search through them to identify errors. You can't do that on colab but you can if you're on Linux(perhaps your should just switch to linux😁).

This is where adb comes in. Android Debug Bridge (adb) is a versatile command-line tool that lets you communicate with a device. You can get more information about it here but for now, we are going to be making use of adb platform tools. Download them here.

This will download a zipped file. Unzip it then copy the apk that we just made and paste it inside the unzipped folder. The selected item is the apk file in the image below:

Connect your device, make sure usb debugging is enabled. Now open a command prompt terminal and navigate to the folder containing the adb files we extracted:

Run adb install name-of-the-app.apk

This will install the application on your device! You can open the application and try it out. To view the application log, run the following command in the terminal:

adb logcat

Now you can see the application log. If any error occurs while running the application, you'll probably find the details in the log, but be warned, it can be like looking for a needle in a haystack!

If you want to learn more about adb and it's commands check out this link.

And that just about wraps it up. I hope you found this article helpful. If you're facing any challenges don't hesitate to reach out in the comments section.

The cost of leveling up🚀🏋️‍♀️

Ngonidzashe Nzenze — Mon, 09 Aug 2021 10:07:38 +0000

If you've ever played dark souls you'll be familiar with these two things:

Death, a lot of it by the way.
Leveling up.

Dark souls is by no means an easy game, I mean just after starting you're already going up against this huge freaking demon 4 times your size! It's not easy but that's what makes it so intriguing.

But enough of that, I'm here to talk about the leveling up system. In dark souls leveling up consists of two parts, collecting souls, which are basically the currency of the game and leveling up at points called bonfires. Pretty straight forward. Beat up some enemies to gain souls, make your way to the bonfire, bump into a fairly weak looking enemy on your way there and think, 'Hmm, why not just beat him up for more souls too.' Then realize that's a mistake, because that weak looking dude wields the sword like Daniel Wu! And just like that you've been stabbed to death and lost all your souls. What a bummer.

Resting by the bonfire

Back to leveling up. Say you do make it to the bonfire and you level up. Immediately, the cost of your next level up increases, but not all that much. Your get more souls, you level up again and the process continues. At some point, you'll realize, leveling up is a hustle. Picture it this way, if I leveled up my strength, the price of the next level up goes up for any other level up as well, which may be stamina or health. Which means it's pretty hard to level up all the stats at the same time.

By stats I'm referring to the characteristics of the avatar you create in the game. In dark souls you can upgrade your avatar's health, strength, dexterity, resistance etc.

Trying to level up all the stats will make your character overly weak which is why most people focus on creating builds. There are a number of builds, such as the strength build, which focuses primarily on leveling up strength and dexterity build which focuses on, well, dexterity. I'm all about that strength build. Beating the living hell out of every enemy I encounter, I love that feeling, but my keyboard doesn't😂.

You might have gotten to this point of the article and are wondering, 'What the hell is this dude talking about?'

Well let me break it down real quick.There are two points I want to make.

In dark souls, as in real life, the cost of leveling up increases the more you level up.
If you level up too many stats at the same time, while your character may be level 50, it will be overly weak.

The cost of leveling up

I am a huge fan of Seth Godin. In his book, The Dip he talks about the importance of becoming exceptional in a particular field. He says most fields have these periods called dips which are the difference between being mediocre and being exceptional.

Here are some quotes from Seth Godin:

"The dip is the long slog between starting and mastery. A long slog that's actually a shortcut, because it gets you where you want to go faster than any other part."

"The dip is the set of artificial screens set up to keep people like you out."

Anybody can quickly pick up a programming language in a few weeks. Your progress is astounding in the first couple of weeks, you're already dreaming of sending space ships to mars thinking, "Oh Elon ain't gonna see me coming," with a big smirk on your face. Then you bump into Algorithms. That's a dip.

Algorithms take time to understand and implement, they need a lot of practice. This is where the real test begins. The key here is patience and consistency. Rushing through the dip is not a great idea. Dark souls has multiple dips, I don't know how many hours it took me to finally beat King Knight Artorias! But one thing I remember is the sense of accomplishment, I had done it!

Knight Artorias the abyss walker

One result you'll get from going through the dip is a major spike in performance. It is easily the fastest way to advance yourself. You don't get buff by lifting tiny weights. You need something to push your muscles to their limit.

There might come times when you feel burnt out, when you feel like you are trying your hardest but things are not improving. Worry not my friend, you are in the dip. You are merely collecting the souls required for your next level up. If you truly want to improve in your craft, do not lose hope. Push yourself.

Know better than to level up too many different stats at the same time.

You are probably familiar with the term:

Jack of all trades, master of none.

When I started playing dark souls that's really what I wanted to be. I wanted to level up on magic, strength, health, intelligence, all of the stats! This is possible no doubt but at the sacrifice of a huge amount of time! No way I was gonna sacrifice all that time when I have code to write! When I realized this, I quickly switched tactics and went for the strength build.

Now get this, my brother had been playing this game for three months before I had started playing it. He would often complain about how hard it was and all, quit it for a few days and then be back. I managed to clear it before he did. Why? The answer is simple really, he had been leveling his stats equally. If he upgraded strength, next he would upgrade health, then dexterity and so on. His avatar was at a high level but was fairly weak.

That is the price of being a jack of all trades, you become mediocre at all the things that you do.

We are human beings, our lives are not infinite that means we don't have all the time in the world. You only have 24 hours each day and the human body has it's limits, that's just reality. Yes, it would be great if we could be the best at everything no doubt, but that's not possible. While you're busy upgrading every stat someone out there is only upgrading their strength and if you were to play against them in a PvP match, they would beat the living hell out of you and your equally upgraded avatar.

Using pyinstaller to package kivy and kivyMD desktop apps

Ngonidzashe Nzenze — Thu, 05 Aug 2021 22:19:46 +0000

Packaging kivy applications can be quite daunting. The documentation can be pretty confusing at times. This guide will walk you through how to package a kivyMD application for windows using pyinstaller through the use of an simple example.

Let's get right into it!🙂

I assume that you already have kivy and kivyMD installed in a virtual environment.

The first step is to install pyinstaller with pip. Open up the command prompt terminal, activate your virtual environment and type in the following:

pip install pyinstaller

Create two files:

main.py - which is going to contain the main code.
main.kv - which is going to contain the code written in kivy lang

Note: These files should be in the same directory.

Inside main.py, enter the following code:

import os, sys
from kivy.resources import resource_add_path, resource_find
from kivymd.app import MDApp

class MainApp(MDApp):
    def build(self):
        self.theme_cls.primary_palette = "DeepPurple"

if __name__ == '__main__':
    try:
        if hasattr(sys, '_MEIPASS'):
            resource_add_path(os.path.join(sys._MEIPASS))
        app = MainApp()
        app.run()
    except Exception as e:
        print(e)
        input("Press enter.")

The above code is very crucial when it comes to packaging our application. The try-catch statement helps us identify errors in our program and the input("Press enter.") stops the console from closing before we want it to.

Edit main.kv:

MDRectangleFlatButton:
    text: "Hello World"
    pos_hint: {'center_x': .5, 'center_y': .5}

We are just going to display a simple button.

Save the files then open the command prompt terminal, activate your environment and navigate to the folder containing the two files.

Now, before you even start the packaging process, make sure when you run your application you are not getting any console errors.

Type pyinstaller --onefile main.py in command prompt and press enter.

When execution is complete, two folders should have been created in the directory containing main.py; dist and build and a file named main.spec.

Navigate into the dist folder. Inside it, there is an executable file, main.exe. If you double click it, a terminal will open and close very quickly.

To solve this issue, add the following lines to the top of main.spec:

from kivy_deps import sdl2, glew
from kivymd import hooks_path as kivymd_hooks_path

Note: If you are not packaging a kivyMD application, there is no need for the second import or modifying hookspath as shown shortly below.

Add in Analysis the lines:

a = Analysis(
    #...
    datas=[('main.kv', '.')],
    hookspath=[kivymd_hooks_path],
   #...
)

datas is made up of a list of tuples where the first item in the tuple is the file name and the second item is the directory that is going to be used to store the file. If for example, you had a folder named images in the same directory as main.py, you would add them as follows:

datas=[('main.kv', '.'), ('./images/*.png', 'images')],

The above line is basically saying, 'Get all the png images in the images folder and save them to a folder named images in the final application.'

Next, edit EXE:

exe = EXE(
    #...
    *[Tree(p) for p in (sdl2.dep_bins + glew.dep_bins)],
    #....
)

The final main.spec file looks like this:

# -*- mode: python ; coding: utf-8 -*-

block_cipher = None
from kivy_deps import sdl2, glew
from kivymd import hooks_path as kivymd_hooks_path

a = Analysis(['main.py'],
             pathex=['C:\\Users\\path\\to\\file'],
             binaries=[],
             datas=[('main.kv', '.')],
             hiddenimports=[],
             hookspath=[kivymd_hooks_path],
             runtime_hooks=[],
             excludes=[],
             win_no_prefer_redirects=False,
             win_private_assemblies=False,
             cipher=block_cipher,
             noarchive=False)
pyz = PYZ(a.pure, a.zipped_data,
             cipher=block_cipher)
exe = EXE(pyz,
          a.scripts,
          a.binaries,
          a.zipfiles,
          a.datas,
          *[Tree(p) for p in (sdl2.dep_bins + glew.dep_bins)],
          name='main',
          debug=False,
          bootloader_ignore_signals=False,
          strip=False,
          upx=True,
          upx_exclude=[],
          runtime_tmpdir=None,
          console=True )

Now save and run: pyinstaller main.spec in the terminal.

After it has finished executing, try executing main.exe again.

And that's it! You have successfully packaged your application.

Errors you may encounter

You may encounter import errors when running pyinstaller. I encountered errors importing pyenchant and cv2. These can easily be fixed by running the following commands in your terminal:

pip install pyenchant
pip install opencv-python

After the installation is complete, run pyinstaller main.spec in your terminal and try executing the main.exe again.

Final thoughts

I hope this guide can prove useful to you. It is by no means a complete guide covering all the aspects of packaging kivy application for windows but I hope you can learn something from it.

For more information, check out: