Utilities¶

Data storage¶

Regular files¶

Get a path for storing plugin data. Parent directories are created automatically:

Downloading a model file

model_path = uni.user_data_path("models/voice.onnx")

import requests
response = requests.get(model_url)
model_path.write_bytes(response.content)

Secure JSON files¶

Encrypted file storing a JSON object in /user/data/plugin.{plugin_name}/:

Storing API credentials

api_credentials = uni.secure_json_file("my_api_credentials")

if api_credentials.exists():
    _api_key = api_credentials.read()["api_key"]

api_credentials.save({
   "api_key": "secret-key",
   "refresh_token": "token-value"
})

Secure JSONL files¶

Encrypted file storing multiple JSON records, optimized for appending:

Append-optimized record storage

from datetime import datetime

notes_file = uni.secure_json_lines_file("my_notes")

# Append a record
notes_file.append({
   "created": datetime.now().isoformat(),
   "message": "Hello, world!",
})

# Read all records
if notes_file.exists():
   for note in notes_file.read():
      print(f"{note['created']}: {note['message']}")

# Overwrite all records
notes_file.save([])

Direct encryption¶

Encrypt and decrypt strings directly:

String encryption

encrypted = uni.encrypt("sensitive data")
decrypted = uni.decrypt(encrypted)

Passphrase protection¶

Require a valid passphrase before sensitive operations:

Passphrase verification

if uni.check_passphrase(user_input):
    perform_backup()

Logging¶

Plugin logging

logger = uni.get_logger(__name__)

logger.info("Plugin loaded successfully")
logger.debug(f"Processing request: {data}")
logger.warning("API rate limit approaching")
logger.exception("Failed to fetch data: {e!s}")

Parsing helpers¶

HTML to Markdown¶

Convert HTML to clean Markdown, returning the page's main article and dropping nav, sidebars, and footers.

Converting HTML to Markdown

html_content = """
<article>
    <h1>Breaking News</h1>
    <p>Important update about <a href="https://example.com">the topic</a>.</p>
    <img src="banner.jpg" alt="Banner">
</article>
"""

# Basic conversion (strips images, preserves links)
markdown = uni.html_to_markdown(html_content)

# Strip links too
markdown = uni.html_to_markdown(html_content, preserve_links=False)

# Truncate long content
markdown = uni.html_to_markdown(html_content, truncate_after=500)

Parameter	Type	Description
`html`	`str`	Raw HTML content
`preserve_links`	`bool`	Keep URLs in markdown format (default: `True`)
`truncate_after`	`int \\| None`	Maximum character length before truncation
`body_width`	`int`	Line wrap width, 0 for no wrapping (default: `0`)
`full_document`	`bool`	Convert the whole document instead of just the article (default: `False`)

Stripping display markup¶

Remove tag-shaped markup the model speaks, like emotive tags (<chuckle>) or bracketed stage directions ([whispering]), before showing text to the user. UNI applies this to subtitles. A message channel can opt in before sending.

Cleaning a line before sending

clean = uni.strip_display_markup(message)

Parameter	Type	Description
`text`	`str`	The model's raw line

JSON parsing¶

Parse JSON from a model reply, tolerating markdown fences and surrounding prose:

Parsing JSON from a model reply

reply = "".join(chunk["message"]["content"] for chunk in uni.llm_chat("tasks", messages))
data = uni.parse_json(reply)

Parameter	Type	Description
`text`	`str`	The model's raw reply

Raises json.JSONDecodeError if no JSON can be recovered.

Datetime parsing¶

Parse natural language datetime strings:

Parsing datetimes

from datetime import datetime

# Relative times
dt = uni.parse_datetime("tomorrow at 3pm")      # -> datetime
dt = uni.parse_datetime("next Tuesday")         # -> datetime
dt = uni.parse_datetime("in 2 hours")           # -> datetime

# Absolute dates
dt = uni.parse_datetime("December 25th")        # -> datetime (next Dec 25)
dt = uni.parse_datetime("2024-03-15 14:30")     # -> datetime

# Past dates (disable prefer_future)
dt = uni.parse_datetime("last Monday", prefer_future=False)

Parameter	Type	Description
`text`	`str`	Natural language datetime
`prefer_future`	`bool`	Interpret ambiguous dates as future (default: `True`)

Raises ValueError if the text cannot be parsed.

Duration parsing¶

Parse duration strings into seconds:

Parsing durations

# Short format
seconds = uni.parse_duration("2h30m")           # -> 9000
seconds = uni.parse_duration("45s")             # -> 45
seconds = uni.parse_duration("1h")              # -> 3600

# Natural language
seconds = uni.parse_duration("2 hours")         # -> 7200
seconds = uni.parse_duration("1 hour 15 minutes")  # -> 4500
seconds = uni.parse_duration("90 minutes")      # -> 5400

Parameter	Type	Description
`text`	`str`	Natural language duration

Returns duration in seconds as int. Raises ValueError if the text cannot be parsed or is negative.

Audio¶

Convert between 16-bit PCM (what TTS engines emit and STT consumes) and float32:

PCM conversion

samples = uni.pcm16_to_float32(pcm_bytes)            # -> float32 in [-1, 1]
pcm_bytes = uni.float32_to_pcm16(samples).tobytes()  # -> 16-bit PCM bytes

Caching¶

Cache function results based on arguments:

Caching expensive calls

@uni.cache(ttl="30m")
def fetch_weather_data():
    return requests.get("https://api.weather.com/...").json()

@uni.cache(ttl="1h", max_size=100)
def get_user_profile(user_id: str):
    return database.query(f"SELECT * FROM users WHERE id = {user_id}")

# Clear cache manually
fetch_weather_data.clear_cache()

Parameter	Description
`ttl`	Time-to-live (optional, defaults to forever)
`max_size`	Maximum cached entries (optional, unlimited)

Scheduling¶

One-time tasks¶

Delayed execution

from datetime import timedelta

job = uni.schedule_after(timedelta(seconds=30), lambda: print("Runs after 30s"))

job.cancel()  # Cancel if needed

Background threads¶

Run functions in managed background threads:

Background processing

def expensive_computation():
    result = process_data()
    save_result(result)

uni.run_task(expensive_computation)

Tasks are tracked and waited for during system unload, ensuring clean shutdown. Always use uni.run_task instead of creating raw threads.

For recurring jobs, see Jobs & events.

Restart prompt¶

Prompt the user to restart UNI after a settings change. The reason is shown in the restart banner:

Prompting a restart

uni.prompt_restart("Wake word settings changed")

See also: Flash messages for transient post-action feedback on settings pages.

System information¶

API version¶

Check API version at runtime (semantic versioning):

Version check

from packaging.version import Version

MIN_API_VERSION = Version("0.1.0")
if uni.API_VERSION < MIN_API_VERSION:
   raise RuntimeError(f"{__name__} requires UNI API version {MIN_API_VERSION}")

CUDA availability¶

CUDA check

if not uni.is_cuda_available():
    raise RuntimeError(f"{__name__} requires CUDA")

Docker detection¶

Check whether UNI is running inside a Docker container:

Docker check

if uni.is_in_docker():
    logger.info("Running in Docker, using companion for stdio servers")

Hugging Face authentication¶

Check if the user has connected their HF account (required for gated models):

HF auth check

if not uni.is_hf_authenticated():
    raise RuntimeError("Please connect UNI to your Hugging Face account.")

For configuration (reading/writing settings), see Getting started. For LLM prompts and embeddings, see Context & prompts.