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.

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 256 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 72 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 173 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 214 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 226 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 239 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 200 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 39 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 119 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 272 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 187 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 82 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 49 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.
    """
    try:
        current_state = wled_requests.get_power_state()
        if current_state is not None:
            wled_requests.change_power_state(not current_state)
            logger.info(f"Toggled WLED power state to: {not current_state}")
        else:
            logger.warning("Unable to toggle WLED power state: Current state is unknown.")
    except Exception as e:
        logger.error(f"Error toggling WLED power state: {e}")
    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 148 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 34 of file app.py.

◆ debug

app.debug

Definition at line 318 of file app.py.

◆ dotenv_path

app.dotenv_path

Definition at line 32 of file app.py.

◆ host

app.host