Functions
Annotated[str, "search page as a rendered template"]	index ()

Annotated[str, "search page as a rendered template"]	toggle_light ()

Annotated[str, "item creation page as a rendered template"]	create_item ()

Annotated[tuple, {"status":str, "status_code":int}]	send_creation ()

Annotated[str, "item page as a rendered template"]	item (item_id)

Annotated[tuple, {"status":str, "status_code":int}]	update_item (item_id)

Annotated[str, "search page as a rendered template"]	delete_item (item_id)

Response	search (term)

Exception	handle_exception (e)

Response	get_env ()

tuple[Response, int]\|Response	get_ollama_models ()

tuple[Response, int]\|Response	get_weight ()

Response	ask_llm_question ()

Annotated[tuple, {"status":str, "status_code":int}]	log_message ()

Variables
	dotenv_path

str	IS_SCALE_ENABLED = "true"

	app = Flask(__name__)

	host

	debug

Detailed Description

ThalamOS Application

This module sets up a Flask web application for managing storage items, controlling WLED devices,
interacting with a WiFi scale, and querying an LLM service. It provides various routes for rendering templates,
handling item creation, deletion, and search, as well as retrieving environment configurations,
scale weight, and logging messages.

Routes:
- /: Renders the search page.
- /toggleLight: Toggles the power state of the WLED device and renders the search template.
- /createItem: Renders the template for creating a new item.
- /sendCreation: Handles the creation of a new item by processing the incoming JSON request data.
- /item/<item_id>: Handles the request to display an item.
- /item/<item_id>/update: Updates an item using the Storage_connector and returns a status message.
- /item/<item_id>/delete: Deletes an item using the Storage_connector and renders the search.html template.
- /search/<term>: Searches for a term in the storage and returns the results in JSON format.
- /config/env: Retrieves the environment configuration.
- /config/ollama/models: Fetches the list of available Ollama models.
- /wifiscale/weight: Retrieves the weight of the scale.
- /llm/ask: Asks a question to the LLM service and returns the response.
- /log: Logs a message with a specified log level.

Error Handling:
- handle_exception: Handles exceptions by passing through HTTP errors.

Setup:
- Initializes the Flask app and sets up CORS.
- Loads environment variables from a .env file.
- Sets up the Storage_connector within the app context.

Usage:
Run the application using the command `python app.py`.

Function Documentation

◆ ask_llm_question()

Response app.ask_llm_question ( )

Ask a question to the Language Learning Model (LLM) service.
This function sends a GET request to the LLM service at the specified endpoint
and returns the response as a JSON object.
Returns:
    Response: A Flask JSON response containing the response from the LLM service.

Definition at line 274 of file app.py.

def ask_llm_question() -> Response:
    """
    Ask a question to the Language Learning Model (LLM) service.
    This function sends a GET request to the LLM service at the specified endpoint
    and returns the response as a JSON object.
    Returns:
        Response: A Flask JSON response containing the response from the LLM service.
    """
    data = request.get_json()
    question = data.get("question")
    logger.info(f"Received question via llm endpoint: {question}")
    response = ollama.ask_question(question)
    return jsonify(response)
 
 
@app.route("/log", methods=["POST"])

◆ create_item()

Annotated[str, "item creation page as a rendered template"] app.create_item ( )

Renders the template for creating a new item.
Returns:
    Response: The rendered HTML template for creating a new item.

Definition at line 90 of file app.py.

def create_item() -> Annotated[str, "item creation page as a rendered template"]:
    """
    Renders the template for creating a new item.
    Returns:
        Response: The rendered HTML template for creating a new item.
    """
    return render_template("createItem.html")
 
 
@app.route("/sendCreation", methods=["POST"])

◆ delete_item()

Annotated[str, "search page as a rendered template"] app.delete_item ( item_id )

Deletes an item using the Storage_connector and renders the search.html template.
Args:
    item_id: The id of the item to be deleted.
Returns:
    A rendered template for the search page.

Definition at line 191 of file app.py.

def delete_item(item_id) -> Annotated[str, "search page as a rendered template"]:
    """
    Deletes an item using the Storage_connector and renders the search.html template.
    Args:
        item_id: The id of the item to be deleted.
    Returns:
        A rendered template for the search page.
    """
 
    Storage_connector.delete_item(item_id)
    return render_template("search.html")
 
 
@app.route("/search/<term>", methods=["GET"])

◆ get_env()

Response app.get_env ( )

Retrieve the environment configuration.
This function uses the config_manager to get the current environment
configuration and returns it as a JSON response.
Returns:
    Response: A Flask JSON response containing the environment configuration.

Definition at line 232 of file app.py.

def get_env() -> Response:
    """
    Retrieve the environment configuration.
    This function uses the config_manager to get the current environment
    configuration and returns it as a JSON response.
    Returns:
        Response: A Flask JSON response containing the environment configuration.
    """
    return jsonify(config_manager.get_env())
 
 
@app.route("/config/ollama/models")

◆ get_ollama_models()

tuple[Response, int] | Response app.get_ollama_models ( )

Fetches the list of available Ollama models.
Returns:
    Annotated[str, "json response"]: A JSON response containing the list of Ollama models.

Definition at line 244 of file app.py.

def get_ollama_models() -> tuple[Response, int] | Response:
    """
    Fetches the list of available Ollama models.
    Returns:
        Annotated[str, "json response"]: A JSON response containing the list of Ollama models.
    """
    values = ollama.get_ollama_models()
    if not values:
        return jsonify({"status": "Ollama service is not enabled"}), 412
    return jsonify(values)
 
 
@app.route("/wifiscale/weight")

◆ get_weight()

tuple[Response, int] | Response app.get_weight ( )

Retrieve the weight of the scale.
This function checks if the scale service is enabled by reading the IS_SCALE_ENABLED environment variable.
If the scale service is not enabled, it returns a JSON response with a status message and HTTP status code 412.
If the scale service is enabled, it uses the wifiscale module to get the weight of the scale and returns it as a JSON response.
Returns:
    Response: A Flask JSON response containing the weight of the scale or a status message.

Definition at line 257 of file app.py.

def get_weight() -> tuple[Response, int] | Response:
    """
    Retrieve the weight of the scale.
    This function checks if the scale service is enabled by reading the IS_SCALE_ENABLED environment variable.
    If the scale service is not enabled, it returns a JSON response with a status message and HTTP status code 412.
    If the scale service is enabled, it uses the wifiscale module to get the weight of the scale and returns it as a JSON response.
    Returns:
        Response: A Flask JSON response containing the weight of the scale or a status message.
    """
    if IS_SCALE_ENABLED == "False":
        return jsonify({"status": "scale service is not enabled"}), 412
 
    weight = wifiscale.get_weight()
    return jsonify({"weight": weight})
 
 
@app.route("/llm/ask", methods=["POST"])

◆ handle_exception()

Exception app.handle_exception ( e )

Handles exceptions by passing through HTTP errors.
Parameters:
e (Exception): The exception to handle.
Returns:
Exception: The same exception that was passed in.

Definition at line 218 of file app.py.

def handle_exception(e) -> Exception:
    """
    Handles exceptions by passing through HTTP errors.
    Parameters:
    e (Exception): The exception to handle.
    Returns:
    Exception: The same exception that was passed in.
    """
 
    # pass through HTTP errors
    return e
 
 
@app.route("/config/env")

◆ index()

Annotated[str, "search page as a rendered template"] app.index ( )

Renders the search page.
Returns:
    Response: The rendered HTML template for the search page.

Definition at line 65 of file app.py.

def index() -> Annotated[str, "search page as a rendered template"]:
    """
    Renders the search page.
    Returns:
        Response: The rendered HTML template for the search page.
    """
    return render_template("search.html")
 
 
@app.route("/toggleLight")

◆ item()

Annotated[str, "item page as a rendered template"] app.item ( item_id )

Handles the request to display an item.
This function performs the following steps:
1. Changes the power state of the WLED device to on.
2. Fetches the item details from the storage using the provided item identifier.
3. Sets the color position on the WLED device based on the fetched item details.
4. If the fetched item contains additional information,
   it parses the data and renders the 'item.jinja2' template
   with the item details and information.
6. If the fetched item does not contain additional information,
   it renders the 'item.jinja2' template with only the item details.
Args:
    item_id (int): The id of the item to display.
Returns:
    The rendered HTML template for the item.

Definition at line 137 of file app.py.

def item(item_id) -> Annotated[str, "item page as a rendered template"]:
    """
    Handles the request to display an item.
    This function performs the following steps:
    1. Changes the power state of the WLED device to on.
    2. Fetches the item details from the storage using the provided item identifier.
    3. Sets the color position on the WLED device based on the fetched item details.
    4. If the fetched item contains additional information,
       it parses the data and renders the 'item.jinja2' template
       with the item details and information.
    6. If the fetched item does not contain additional information,
       it renders the 'item.jinja2' template with only the item details.
    Args:
        item_id (int): The id of the item to display.
    Returns:
        The rendered HTML template for the item.
    """
    wled_requests.change_power_state(True)
    item_sql = Storage_connector.fetch_item(item_id)
    wled_requests.color_pos(item_sql[1])
    logger.info(f"Fetched item details for item_id {item_id}: {item_sql}")
    if item_sql[4]:
        json_info = json.loads(item_sql[4])
        return render_template("item.jinja2", item=item_sql, json=json_info, id=item_id)
 
    return render_template("item.jinja2", item=item_sql, id=item_id)
 
 
@app.route("/item/<item_id>/update", methods=["POST"])

◆ log_message()

Annotated[tuple, {"status": str, "status_code": int}] app.log_message ( )

Logs a message with a specified log level.

The log level and message content are extracted from the JSON payload of the request.
If the log level is not provided, it defaults to 'INFO'.
If the message content is not provided, it defaults to an empty string.

Returns:
    tuple: A dictionary with a status message and an HTTP status code 201.

Request JSON structure:
    {
        "level": "DEBUG" | "INFO" | "WARNING" | "ERROR" | "CRITICAL",
        "message": "Your log message here"
    }

Definition at line 290 of file app.py.

def log_message() -> Annotated[tuple, {"status": str, "status_code": int}]:
    """
    Logs a message with a specified log level.
 
    The log level and message content are extracted from the JSON payload of the request.
    If the log level is not provided, it defaults to 'INFO'.
    If the message content is not provided, it defaults to an empty string.
 
    Returns:
        tuple: A dictionary with a status message and an HTTP status code 201.
 
    Request JSON structure:
        {
            "level": "DEBUG" | "INFO" | "WARNING" | "ERROR" | "CRITICAL",
            "message": "Your log message here"
        }
    """
 
    data: Annotated[str, "content of request"] = request.json
    level: Annotated[str, "log level, default value is INFO"] = data.get(
        "level", "INFO"
    )
    message: Annotated[str, "content of log, default value is empty"] = data.get(
        "message", ""
    )
 
    match level:
        case "DEBUG":
            logger.debug(message)
        case "INFO":
            logger.info(message)
        case "WARNING":
            logger.warning(message)
        case "ERROR":
            logger.error(message)
        case "CRITICAL":
            logger.critical(message)
 
    return {"status": "created"}, 201
 
 

◆ search()

Response app.search ( term )

Search for a term in the storage and return the results in JSON format.
Args:
    term (str): The term to search for in the storage.
Returns:
    Response: A Flask Response object containing the search results in JSON format.

Definition at line 205 of file app.py.

def search(term) -> Response:
    """
    Search for a term in the storage and return the results in JSON format.
    Args:
        term (str): The term to search for in the storage.
    Returns:
        Response: A Flask Response object containing the search results in JSON format.
    """
    data = Storage_connector.search(term)
    return jsonify(data)
 
 
@app.errorhandler(Exception)

◆ send_creation()

Annotated[tuple, {"status": str, "status_code": int}] app.send_creation ( )

Handle the creation of a new item by processing the incoming JSON request data.
The function expects a JSON payload with the following structure:
{
    "info": <json>,
    "type": <str>,
    "name": <str>,
    "position": <str>
}
It extracts the necessary information from the JSON payload and attempts to create a new item
using the Storage_connector.CreateItem method.
If an exception occurs during the creation process,
it prints the exception.
Returns:
    tuple: A dictionary with a status message and an HTTP status code.

Definition at line 100 of file app.py.

def send_creation() -> Annotated[tuple, {"status": str, "status_code": int}]:
    """
    Handle the creation of a new item by processing the incoming JSON request data.
    The function expects a JSON payload with the following structure:
    {
        "info": <json>,
        "type": <str>,
        "name": <str>,
        "position": <str>
    }
    It extracts the necessary information from the JSON payload and attempts to create a new item
    using the Storage_connector.CreateItem method.
    If an exception occurs during the creation process,
    it prints the exception.
    Returns:
        tuple: A dictionary with a status message and an HTTP status code.
    """
 
    data = request.get_json()
    logger.info(f"Received creation request with data: {data}")
    info = json.dumps(data["info"])
    obj_type = data["type"]
    name = data["name"]
    pos = data["position"]
    try:
        Storage_connector.create_item(
            pos=pos, obj_type=obj_type, name=name, json_data=info
        )
    except Exception as e:
        logger.error(
            f"Failed to create item with name: {name}, type: {obj_type}, position: {pos}. Error: {e}"
        )
        return {"status": "error"}, 500
    return {"status": "created"}, 201
 
 
@app.route("/item/<item_id>")

◆ toggle_light()

Annotated[str, "search page as a rendered template"] app.toggle_light ( )

Toggles the power state of the WLED device and renders the search template.
This function changes the power state of the WLED device to the opposite of its current state
by calling the `changePowerState` method of the `wled_requests` object.
After toggling the power state,
it returns the rendered "search.html" template.
Returns:
    str: The rendered "search.html" template.

Definition at line 75 of file app.py.

def toggle_light() -> Annotated[str, "search page as a rendered template"]:
    """
    Toggles the power state of the WLED device and renders the search template.
    This function changes the power state of the WLED device to the opposite of its current state
    by calling the `changePowerState` method of the `wled_requests` object.
    After toggling the power state,
    it returns the rendered "search.html" template.
    Returns:
        str: The rendered "search.html" template.
    """
    wled_requests.change_power_state(not wled_requests.get_power_state())
    return render_template("search.html")
 
 
@app.route("/createItem")

◆ update_item()

Annotated[tuple, {"status": str, "status_code": int}] app.update_item ( item_id )

Updates an item using the Storage_connector and returns a status message.
Args:
    item_id: The id of the item to be updated.
Returns:
    A dictionary with a status message and an HTTP status code.

Definition at line 166 of file app.py.

def update_item(item_id) -> Annotated[tuple, {"status": str, "status_code": int}]:
    """
    Updates an item using the Storage_connector and returns a status message.
    Args:
        item_id: The id of the item to be updated.
    Returns:
        A dictionary with a status message and an HTTP status code.
    """
    data = request.get_json()
    logger.info(f"Received update request for item_id {item_id} with data: {data}")
    info = json.dumps(data.get("info", {}))
    obj_type = data.get("type")
    name = data.get("name")
    pos = data.get("position")
    try:
        Storage_connector.update_item(
            item_id=item_id, pos=pos, obj_type=obj_type, name=name, json_data=info
        )
    except Exception as e:
        logger.error(f"Failed to update item with id: {item_id}. Error: {e}")
        return {"status": "error"}, 500
    return {"status": "updated"}, 200
 
 
@app.route("/item/<item>/delete")

Variable Documentation

◆ app

app.app = Flask(__name__)

Definition at line 60 of file app.py.

◆ debug

app.debug

Definition at line 336 of file app.py.

◆ dotenv_path

app.dotenv_path

Definition at line 58 of file app.py.

◆ host

app.host