iridescentX/openui

1from pathlib import Path2import hashlib3import json4import re5from datetime import timedelta6from typing import Optional, List, Tuple7import uuid8import time9 10 11def get_last_user_message_item(messages: List[dict]) -> str:12    for message in reversed(messages):13        if message["role"] == "user":14            return message15    return None16 17 18def get_last_user_message(messages: List[dict]) -> str:19    message = get_last_user_message_item(messages)20 21    if message is not None:22        if isinstance(message["content"], list):23            for item in message["content"]:24                if item["type"] == "text":25                    return item["text"]26        return message["content"]27    return None28 29 30def get_last_assistant_message(messages: List[dict]) -> str:31    for message in reversed(messages):32        if message["role"] == "assistant":33            if isinstance(message["content"], list):34                for item in message["content"]:35                    if item["type"] == "text":36                        return item["text"]37            return message["content"]38    return None39 40 41def get_system_message(messages: List[dict]) -> dict:42    for message in messages:43        if message["role"] == "system":44            return message45    return None46 47 48def remove_system_message(messages: List[dict]) -> List[dict]:49    return [message for message in messages if message["role"] != "system"]50 51 52def pop_system_message(messages: List[dict]) -> Tuple[dict, List[dict]]:53    return get_system_message(messages), remove_system_message(messages)54 55 56def add_or_update_system_message(content: str, messages: List[dict]):57    """58    Adds a new system message at the beginning of the messages list59    or updates the existing system message at the beginning.60 61    :param msg: The message to be added or appended.62    :param messages: The list of message dictionaries.63    :return: The updated list of message dictionaries.64    """65 66    if messages and messages[0].get("role") == "system":67        messages[0]["content"] += f"{content}\n{messages[0]['content']}"68    else:69        # Insert at the beginning70        messages.insert(0, {"role": "system", "content": content})71 72    return messages73 74 75def stream_message_template(model: str, message: str):76    return {77        "id": f"{model}-{str(uuid.uuid4())}",78        "object": "chat.completion.chunk",79        "created": int(time.time()),80        "model": model,81        "choices": [82            {83                "index": 0,84                "delta": {"content": message},85                "logprobs": None,86                "finish_reason": None,87            }88        ],89    }90 91 92def get_gravatar_url(email):93    # Trim leading and trailing whitespace from94    # an email address and force all characters95    # to lower case96    address = str(email).strip().lower()97 98    # Create a SHA256 hash of the final string99    hash_object = hashlib.sha256(address.encode())100    hash_hex = hash_object.hexdigest()101 102    # Grab the actual image URL103    return f"https://www.gravatar.com/avatar/{hash_hex}?d=mp"104 105 106def calculate_sha256(file):107    sha256 = hashlib.sha256()108    # Read the file in chunks to efficiently handle large files109    for chunk in iter(lambda: file.read(8192), b""):110        sha256.update(chunk)111    return sha256.hexdigest()112 113 114def calculate_sha256_string(string):115    # Create a new SHA-256 hash object116    sha256_hash = hashlib.sha256()117    # Update the hash object with the bytes of the input string118    sha256_hash.update(string.encode("utf-8"))119    # Get the hexadecimal representation of the hash120    hashed_string = sha256_hash.hexdigest()121    return hashed_string122 123 124def validate_email_format(email: str) -> bool:125    if email.endswith("@localhost"):126        return True127 128    return bool(re.match(r"[^@]+@[^@]+\.[^@]+", email))129 130 131def sanitize_filename(file_name):132    # Convert to lowercase133    lower_case_file_name = file_name.lower()134 135    # Remove special characters using regular expression136    sanitized_file_name = re.sub(r"[^\w\s]", "", lower_case_file_name)137 138    # Replace spaces with dashes139    final_file_name = re.sub(r"\s+", "-", sanitized_file_name)140 141    return final_file_name142 143 144def extract_folders_after_data_docs(path):145    # Convert the path to a Path object if it's not already146    path = Path(path)147 148    # Extract parts of the path149    parts = path.parts150 151    # Find the index of '/data/docs' in the path152    try:153        index_data_docs = parts.index("data") + 1154        index_docs = parts.index("docs", index_data_docs) + 1155    except ValueError:156        return []157 158    # Exclude the filename and accumulate folder names159    tags = []160 161    folders = parts[index_docs:-1]162    for idx, part in enumerate(folders):163        tags.append("/".join(folders[: idx + 1]))164 165    return tags166 167 168def parse_duration(duration: str) -> Optional[timedelta]:169    if duration == "-1" or duration == "0":170        return None171 172    # Regular expression to find number and unit pairs173    pattern = r"(-?\d+(\.\d+)?)(ms|s|m|h|d|w)"174    matches = re.findall(pattern, duration)175 176    if not matches:177        raise ValueError("Invalid duration string")178 179    total_duration = timedelta()180 181    for number, _, unit in matches:182        number = float(number)183        if unit == "ms":184            total_duration += timedelta(milliseconds=number)185        elif unit == "s":186            total_duration += timedelta(seconds=number)187        elif unit == "m":188            total_duration += timedelta(minutes=number)189        elif unit == "h":190            total_duration += timedelta(hours=number)191        elif unit == "d":192            total_duration += timedelta(days=number)193        elif unit == "w":194            total_duration += timedelta(weeks=number)195 196    return total_duration197 198 199def parse_ollama_modelfile(model_text):200    parameters_meta = {201        "mirostat": int,202        "mirostat_eta": float,203        "mirostat_tau": float,204        "num_ctx": int,205        "repeat_last_n": int,206        "repeat_penalty": float,207        "temperature": float,208        "seed": int,209        "tfs_z": float,210        "num_predict": int,211        "top_k": int,212        "top_p": float,213        "num_keep": int,214        "typical_p": float,215        "presence_penalty": float,216        "frequency_penalty": float,217        "penalize_newline": bool,218        "numa": bool,219        "num_batch": int,220        "num_gpu": int,221        "main_gpu": int,222        "low_vram": bool,223        "f16_kv": bool,224        "vocab_only": bool,225        "use_mmap": bool,226        "use_mlock": bool,227        "num_thread": int,228    }229 230    data = {"base_model_id": None, "params": {}}231 232    # Parse base model233    base_model_match = re.search(234        r"^FROM\s+(\w+)", model_text, re.MULTILINE | re.IGNORECASE235    )236    if base_model_match:237        data["base_model_id"] = base_model_match.group(1)238 239    # Parse template240    template_match = re.search(241        r'TEMPLATE\s+"""(.+?)"""', model_text, re.DOTALL | re.IGNORECASE242    )243    if template_match:244        data["params"] = {"template": template_match.group(1).strip()}245 246    # Parse stops247    stops = re.findall(r'PARAMETER stop "(.*?)"', model_text, re.IGNORECASE)248    if stops:249        data["params"]["stop"] = stops250 251    # Parse other parameters from the provided list252    for param, param_type in parameters_meta.items():253        param_match = re.search(rf"PARAMETER {param} (.+)", model_text, re.IGNORECASE)254        if param_match:255            value = param_match.group(1)256 257            try:258                if param_type == int:259                    value = int(value)260                elif param_type == float:261                    value = float(value)262                elif param_type == bool:263                    value = value.lower() == "true"264            except Exception as e:265                print(e)266                continue267 268            data["params"][param] = value269 270    # Parse adapter271    adapter_match = re.search(r"ADAPTER (.+)", model_text, re.IGNORECASE)272    if adapter_match:273        data["params"]["adapter"] = adapter_match.group(1)274 275    # Parse system description276    system_desc_match = re.search(277        r'SYSTEM\s+"""(.+?)"""', model_text, re.DOTALL | re.IGNORECASE278    )279    system_desc_match_single = re.search(280        r"SYSTEM\s+([^\n]+)", model_text, re.IGNORECASE281    )282 283    if system_desc_match:284        data["params"]["system"] = system_desc_match.group(1).strip()285    elif system_desc_match_single:286        data["params"]["system"] = system_desc_match_single.group(1).strip()287 288    # Parse messages289    messages = []290    message_matches = re.findall(r"MESSAGE (\w+) (.+)", model_text, re.IGNORECASE)291    for role, content in message_matches:292        messages.append({"role": role, "content": content})293 294    if messages:295        data["params"]["messages"] = messages296 297    return data298