Modules Documentation

This section contains the complete API documentation for all Mycelian modules, automatically generated from the source code.

Overview

Mycelian is a comprehensive streaming toolkit with the following major components:

Alert System: Custom alert processing for Twitch events (follows, subs, bits, etc.)
Chatbot System: Advanced Twitch chatbot with commands, events, and automated responses
Connector Automation: Trigger-action workflows for stream automation
Web Engine: Flask-based server for browser sources and real-time data
Service Integrations: Support for Twitch, Spotify, PlayStation Network, and StreamLabs
Desktop UI: NiceGUI-based interface for configuration and management

Core Modules

Main Application

Configuration Management

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.config_manager.AppConfig(database_type='sql', sql_database_path='mycelian.db', firebase_service_account_path='ServiceAccountKey.json', firebase_database_url='https://your-project-default-rtdb.firebaseio.com/', mongodb_connection_string='mongodb://localhost:27017/', mongodb_database_name='mycelian', connection_timeout=30, retry_attempts=3, config_version='1.0', last_updated='')[source]

Bases: object

Application configuration structure

config_version: str = '1.0'

connection_timeout: int = 30

database_type: str = 'sql'

firebase_database_url: str = 'https://your-project-default-rtdb.firebaseio.com/'

firebase_service_account_path: str = 'ServiceAccountKey.json'

last_updated: str = ''

mongodb_connection_string: str = 'mongodb://localhost:27017/'

mongodb_database_name: str = 'mycelian'

retry_attempts: int = 3

sql_database_path: str = 'mycelian.db'

class modules.config_manager.ConfigManager(config_path='config.json')[source]

Bases: object

Manages application configuration stored in external files

export_config(export_path)[source]

Export configuration to a different file

Return type:: bool

get_config()[source]

Get the current configuration

Return type:: Optional[AppConfig]

get_database_config()[source]

Get database configuration as a dictionary

Return type:: Dict[str, Any]

get_database_type()[source]

Get the configured database type

Return type:: str

import_config(import_path)[source]

Import configuration from a different file

Return type:: bool

initialize()[source]

Initialize the configuration manager

Return type:: bool

set_database_type(database_type)[source]

Set the database type and save configuration

Return type:: bool

update_database_config(**kwargs)[source]

Update database configuration parameters

Return type:: bool

modules.config_manager.get_database_config()[source]

Get database configuration

Return type:: Dict[str, Any]

modules.config_manager.get_database_type()[source]

Get the configured database type

Return type:: str

modules.config_manager.initialize_config()[source]

Initialize the configuration manager

Return type:: bool

modules.config_manager.set_database_type(database_type)[source]

Set the database type

Return type:: bool

Database Management

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.database_manager.DatabaseConfig(database_type='sql', sql_database_path='mycelian.db', firebase_service_account_path='', firebase_database_url='', mongodb_connection_string='', mongodb_database_name='', streamer_name='mycelian', connection_timeout=30, retry_attempts=3)[source]

Bases: object

Configuration settings for database connections

connection_timeout: int = 30

database_type: str = 'sql'

firebase_database_url: str = ''

firebase_service_account_path: str = ''

mongodb_connection_string: str = ''

mongodb_database_name: str = ''

retry_attempts: int = 3

sql_database_path: str = 'mycelian.db'

streamer_name: str = 'mycelian'

class modules.database_manager.DatabaseInterface[source]

Bases: ABC

Abstract base class for database implementations

abstractmethod delete_data(path)[source]

Delete data from the database

Return type:: bool

abstractmethod get_all_paths()[source]

Get all data paths stored in the database

Return type:: List[str]

abstractmethod get_connection_status()[source]

Get the current connection status

Return type:: Dict[str, Any]

abstractmethod get_data(path, request_etag=False)[source]

Get data from the database

Return type:: Dict[str, Any]

abstractmethod async get_multiple_data_async(paths)[source]

Get data from multiple paths asynchronously

Return type:: Dict[str, Dict[str, Any]]

abstractmethod get_snapshot()[source]

Get a complete snapshot of all database data as a nested dictionary

Return type:: Dict[str, Any]

abstractmethod initialize()[source]

Initialize the database connection

Return type:: bool

abstractmethod set_data(path, data)[source]

Set data in the database

Return type:: bool

abstractmethod test_connection()[source]

Test the database connection

Return type:: bool

abstractmethod update_data(path, data)[source]

Update data in the database

Return type:: bool

class modules.database_manager.DatabaseManager[source]

Bases: object

Main database manager that handles all database operations

delete_data(path)[source]

Delete data from the database

Return type:: bool

get_all_paths()[source]

Get all data paths stored in the database

Return type:: List[str]

get_available_databases()[source]

Get list of available database types

Return type:: List[str]

get_config()[source]

Get the current database configuration

Return type:: DatabaseConfig

get_connection_status()[source]

Get the current database connection status

Return type:: Dict[str, Any]

get_data(path, request_etag=False)[source]

Get data from the database

Return type:: Dict[str, Any]

async get_multiple_data_async(paths)[source]

Get data from multiple paths asynchronously

Return type:: Dict[str, Dict[str, Any]]

get_snapshot()[source]

Get a complete snapshot of all database data as a nested dictionary

Return type:: Dict[str, Any]

initialize(config=None)[source]

Initialize the database manager with the specified configuration

Return type:: bool

migrate_data(source_config, target_config)[source]

Migrate data from one database to another

Return type:: bool

set_data(path, data)[source]

Set data in the database

Return type:: bool

test_connection()[source]

Test the database connection

Return type:: bool

update_config(**kwargs)[source]

Update database configuration and reinitialize if necessary

Return type:: bool

update_data(path, data)[source]

Update data in the database

Return type:: bool

class modules.database_manager.FirebaseDatabase(config)[source]

Bases: DatabaseInterface

Firebase Realtime Database implementation

delete_data(path)[source]

Delete data from Firebase database

Return type:: bool

get_all_paths()[source]

Get all data paths from Firebase by traversing the tree

Return type:: List[str]

get_connection_status()[source]

Get Firebase connection status

Return type:: Dict[str, Any]

get_data(path, request_etag=False)[source]

Get data from Firebase database

Return type:: Dict[str, Any]

async get_multiple_data_async(paths)[source]

Get data from multiple Firebase paths asynchronously in batches

Return type:: Dict[str, Dict[str, Any]]

get_snapshot()[source]

Get a complete snapshot of all Firebase data

Return type:: Dict[str, Any]

initialize()[source]

Initialize Firebase connection

Return type:: bool

set_data(path, data)[source]

Set data in Firebase database

Return type:: bool

test_connection()[source]

Test Firebase connection

Return type:: bool

update_data(path, data)[source]

Update data in Firebase database

Return type:: bool

class modules.database_manager.MongoDatabase(config)[source]

Bases: DatabaseInterface

MongoDB database implementation

delete_data(path)[source]

Delete data from MongoDB database

Return type:: bool

get_all_paths()[source]

Get all data paths stored in MongoDB

Return type:: List[str]

get_connection_status()[source]

Get MongoDB connection status

Return type:: Dict[str, Any]

get_data(path, request_etag=False)[source]

Get data from MongoDB database

Return type:: Dict[str, Any]

async get_multiple_data_async(paths)[source]

Get data from multiple MongoDB paths asynchronously

Return type:: Dict[str, Dict[str, Any]]

get_snapshot()[source]

Get a complete snapshot of all MongoDB data as a nested dictionary

Return type:: Dict[str, Any]

initialize()[source]

Initialize MongoDB connection

Return type:: bool

set_data(path, data)[source]

Set data in MongoDB database

Return type:: bool

test_connection()[source]

Test MongoDB connection

Return type:: bool

update_data(path, data)[source]

Update data in MongoDB database

Return type:: bool

class modules.database_manager.SQLDatabase(config)[source]

Bases: DatabaseInterface

SQLite database implementation

delete_data(path)[source]

Delete data from SQLite database

Return type:: bool

get_all_paths()[source]

Get all data paths stored in SQLite database

Return type:: List[str]

get_connection_status()[source]

Get SQLite connection status

Return type:: Dict[str, Any]

get_data(path, request_etag=False)[source]

Get data from SQLite database

Return type:: Dict[str, Any]

async get_multiple_data_async(paths)[source]

Get data from multiple paths asynchronously

Return type:: Dict[str, Dict[str, Any]]

get_snapshot()[source]

Get a complete snapshot of all database data as a nested dictionary

Return type:: Dict[str, Any]

initialize()[source]

Initialize SQLite database and create tables

Return type:: bool

set_data(path, data)[source]

Set data in SQLite database

Return type:: bool

test_connection()[source]

Test SQLite connection

Return type:: bool

update_data(path, data)[source]

Update data in SQLite database

Return type:: bool

modules.database_manager.delete_data(path)[source]

Delete data from the database

Return type:: bool

modules.database_manager.get_all_paths()[source]

Get all data paths stored in the database

Return type:: List[str]

modules.database_manager.get_connection_status()[source]

Get the current database connection status

Return type:: Dict[str, Any]

modules.database_manager.get_data(path, request_etag=False)[source]

Get data from the database

Return type:: Dict[str, Any]

async modules.database_manager.get_multiple_data_async(paths)[source]

Get data from multiple paths asynchronously

Return type:: Dict[str, Dict[str, Any]]

modules.database_manager.get_snapshot()[source]

Get a complete snapshot of all database data as a nested dictionary

Return type:: Dict[str, Any]

modules.database_manager.initialize_database(config=None)[source]

Initialize the database with the specified configuration

Return type:: bool

async modules.database_manager.load_all_initial_data()[source]

Load all startup data in parallel - optimized approach

Return type:: Dict[str, Any]

modules.database_manager.set_data(path, data)[source]

Set data in the database

Return type:: bool

modules.database_manager.test_connection()[source]

Test the database connection

Return type:: bool

modules.database_manager.update_data(path, data)[source]

Update data in the database

Return type:: bool

Statistics Manager

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.statistics_manager.AlertStatistics(bit_alerts_played=0, total_bits=0, resubs_played=0, new_subs_played=0, gift_subs_played=0, total_gift_subs=0, follow_alerts_played=0, raids=0, point_alerts_redeemed=0, total_channel_points_redeemed=0, donations=0, user_stats=<factory>, alert_type_stats=<factory>)[source]

Bases: object

Statistics for alert events

alert_type_stats: Dict[str, IndividualAlertTypeStatistics]

bit_alerts_played: int = 0

donations: int = 0

follow_alerts_played: int = 0

gift_subs_played: int = 0

new_subs_played: int = 0

point_alerts_redeemed: int = 0

raids: int = 0

resubs_played: int = 0

total_bits: int = 0

total_channel_points_redeemed: int = 0

total_gift_subs: int = 0

user_stats: Dict[str, UserAlertStatistics]

class modules.statistics_manager.ChatStatistics(twitch_messages_received=0, total_messages=0, user_stats=<factory>)[source]

Bases: object

Statistics for chat messages

total_messages: int = 0

twitch_messages_received: int = 0

user_stats: Dict[str, UserChatStatistics]

class modules.statistics_manager.ChatbotStatistics(commands_triggered=0, events_triggered=0, total_interactions=0, user_stats=<factory>, user_command_stats=<factory>, user_event_stats=<factory>, command_stats=<factory>, event_stats=<factory>)[source]

Bases: object

Statistics for chatbot system

command_stats: Dict[str, IndividualCommandStatistics]

commands_triggered: int = 0

event_stats: Dict[str, IndividualEventStatistics]

events_triggered: int = 0

total_interactions: int = 0

user_command_stats: Dict[str, UserCommandStatistics]

user_event_stats: Dict[str, UserEventStatistics]

user_stats: Dict[str, UserChatbotStatistics]

class modules.statistics_manager.ConnectorStatistics(connectors_triggered=0, total_triggers=0, user_stats=<factory>, user_connector_stats=<factory>, connector_stats=<factory>)[source]

Bases: object

Statistics for connector system

connector_stats: Dict[str, IndividualConnectorStatistics]

connectors_triggered: int = 0

total_triggers: int = 0

user_connector_stats: Dict[str, UserConnectorStatistics]

user_stats: Dict[str, UserConnectorStatistics]

class modules.statistics_manager.HypeTrainStatistics(level_completions=<factory>, total_completions=0)[source]

Bases: object

Statistics for hype train completions by level

level_completions: Dict[int, int]

total_completions: int = 0

class modules.statistics_manager.IndividualAlertTypeStatistics(alert_type='', alert_name='', trigger_count=0, last_triggered=<factory>, first_triggered=<factory>)[source]

Bases: object

Statistics for individual alert types

alert_name: str = ''

alert_type: str = ''

first_triggered: float

last_triggered: float

trigger_count: int = 0

class modules.statistics_manager.IndividualCommandStatistics(command_name='', usage_count=0, last_used=<factory>, first_used=<factory>)[source]

Bases: object

Statistics for individual commands

command_name: str = ''

first_used: float

last_used: float

usage_count: int = 0

class modules.statistics_manager.IndividualConnectorStatistics(connector_name='', trigger_count=0, last_triggered=<factory>, first_triggered=<factory>)[source]

Bases: object

Statistics for individual connectors

connector_name: str = ''

first_triggered: float

last_triggered: float

trigger_count: int = 0

class modules.statistics_manager.IndividualEventStatistics(event_name='', trigger_count=0, last_triggered=<factory>, first_triggered=<factory>)[source]

Bases: object

Statistics for individual events

event_name: str = ''

first_triggered: float

last_triggered: float

trigger_count: int = 0

class modules.statistics_manager.QuoteStatistics(total_quotes_redeemed=0, individual_quote_usage=<factory>, user_stats=<factory>)[source]

Bases: object

Statistics for quote system

individual_quote_usage: Dict[str, int]

total_quotes_redeemed: int = 0

user_stats: Dict[str, UserQuoteStatistics]

class modules.statistics_manager.SessionStatistics(start_time=<factory>, last_save_time=<factory>, session_duration=0.0)[source]

Bases: object

Statistics for the current session

last_save_time: float

session_duration: float = 0.0

start_time: float

class modules.statistics_manager.StatisticsData(alerts=<factory>, hype_trains=<factory>, connectors=<factory>, chatbot=<factory>, quotes=<factory>, chat=<factory>, templates=<factory>, session=<factory>)[source]

Bases: object

Main statistics data container

alerts: AlertStatistics

chat: ChatStatistics

chatbot: ChatbotStatistics

connectors: ConnectorStatistics

hype_trains: HypeTrainStatistics

quotes: QuoteStatistics

session: SessionStatistics

templates: Dict[str, TemplateStatistics]

class modules.statistics_manager.StatisticsManager[source]

Bases: object

Comprehensive statistics manager for Mycelian

check_data_integrity()[source]

Check data integrity and return list of issues found

Return type:: List[str]

ensure_periodic_saving()[source]: Ensure the periodic saving thread is running

force_save()[source]: Force an immediate save of statistics

get_all_statistics()[source]

Get all statistics as a dictionary

Return type:: Dict[str, Any]

get_all_tracked_usernames()[source]

Get all usernames that have been tracked in the events database.

Return type:: List[str]
Returns:: Sorted list of unique usernames.

get_date_range_highlights(start_time, end_time)[source]

Get highlight data for the export card.

Parameters:

start_time (float) – Start of range (unix timestamp).
end_time (float) – End of range (unix timestamp).

Return type:

Dict[str, Any]

Returns:

Dictionary of highlight metrics (top donors, totals, etc.).

get_date_range_summary(start_time, end_time)[source]

Get aggregate stats across all users for a date range.

Parameters:

start_time (float) – Start of range (unix timestamp).
end_time (float) – End of range (unix timestamp).

Return type:

Dict[str, Any]

Returns:

Dictionary with per-event-type counts/totals and overall metrics.

get_recent_users(stat_type, limit=10)[source]

Get most recent users for a specific statistic type

Parameters:

stat_type (str) – Type of statistic (e.g., ‘alerts’, ‘chat’, ‘connectors’)
limit (int) – Maximum number of users to return

Return type:

List[Dict[str, Any]]

Returns:

List of dictionaries with username and last_seen timestamp, sorted by recency

get_saving_status()[source]: Get the current status of periodic saving

get_statistics_summary()[source]

Get a summary of key statistics

Return type:: Dict[str, Any]

get_template_stats(template_name=None)[source]

Get template statistics

Parameters:: template_name (Optional[str]) – Specific template name, or None for all templates
Return type:: Dict[str, Any]
Returns:: Dictionary of template statistics

get_top_alert_types(limit=10)[source]

Get top alert types by trigger count

Return type:: List[Dict[str, Any]]

get_top_commands(limit=10)[source]

Get top commands by usage count

Return type:: List[Dict[str, Any]]

get_top_connectors(limit=10)[source]

Get top connectors by trigger count

Return type:: List[Dict[str, Any]]

get_top_events(limit=10)[source]

Get top events by trigger count

Return type:: List[Dict[str, Any]]

get_top_users_by_statistic(stat_type, limit=10)[source]

Get top users ranked by a specific statistic

Parameters:

stat_type (str) – Type of statistic (e.g., ‘total_alerts’, ‘bits_donated’, ‘quotes_redeemed’)
limit (int) – Maximum number of users to return

Return type:

List[Dict[str, Any]]

Returns:

List of dictionaries with username and stat value, sorted by value descending

get_top_users_for_command(command_name, limit=10)[source]

Get top users who used a specific command the most

Return type:: List[Dict[str, Any]]

get_top_users_for_connector(connector_name, limit=10)[source]

Get top users who triggered a specific connector the most

Return type:: List[Dict[str, Any]]

get_top_users_for_event(event_name, limit=10)[source]

Get top users who triggered a specific event the most

Return type:: List[Dict[str, Any]]

get_top_users_for_quote(quote_id, limit=10)[source]

Get top users who redeemed a specific quote the most

Return type:: List[Dict[str, Any]]

get_top_users_in_range(start_time, end_time, event_type=None, limit=10)[source]

Get top contributors in a date range.

Parameters:

start_time (float) – Start of range (unix timestamp).
end_time (float) – End of range (unix timestamp).
event_type (Optional[str]) – Filter by event type. None for all types.
limit (int) – Maximum number of users to return.

Return type:

List[Dict[str, Any]]

Returns:

List of dicts with username, event_count, and total_amount.

get_user_activity_summary(username)[source]

Get a summary of user activity across all categories

Return type:: Dict[str, Any]

get_user_events(username, start_time=None, end_time=None, event_type=None, limit=100)[source]

Query timestamped events for a user within a date range.

Parameters:

username (str) – Username to query.
start_time (Optional[float]) – Start of range (unix timestamp). None for no lower bound.
end_time (Optional[float]) – End of range (unix timestamp). None for no upper bound.
event_type (Optional[str]) – Filter by event type. None for all types.
limit (int) – Maximum number of events to return.

Return type:

List[Dict[str, Any]]

Returns:

List of event dictionaries ordered by timestamp descending.

get_user_individual_usage(username)[source]

Get detailed individual item usage for a specific user

Return type:: Dict[str, Any]

get_user_statistics(username)[source]

Get all statistics for a specific user

Return type:: Dict[str, Any]

increment_bit_alerts(bit_amount=0, username=None, alert_name='')[source]: Increment bit alerts played count and total bits amount

increment_channel_points_redeemed(points_amount, username=None)[source]: Increment total channel points redeemed and track per-user statistics

increment_commands_triggered(username=None, command_name='')[source]: Increment chatbot commands triggered count

increment_connector_triggers(count=1, username=None, connector_name='')[source]: Increment total connector triggers

increment_connectors_triggered(username=None, connector_name='')[source]: Increment connectors triggered count

increment_donations(username=None, alert_name='')[source]: Increment donations count

increment_events_triggered(username=None, event_name='')[source]: Increment chatbot events triggered count

increment_follow_alerts(username=None, alert_name='')[source]: Increment follow alerts played count

increment_gift_subs(gift_quantity=1, username=None, alert_name='')[source]: Increment gift subs played count and total gift subs amount

increment_hype_train_completion(level)[source]: Increment hype train completion for specific level (supports unlimited levels)

increment_new_subs(username=None, alert_name='')[source]: Increment new subs played count

increment_point_alerts(username=None, alert_name='')[source]: Increment point alerts redeemed count

increment_quote_redeemed(quote_id, username=None)[source]: Increment quote redeemed count for specific quote

increment_raids(username=None, alert_name='')[source]: Increment raids count

increment_resubs(username=None, alert_name='')[source]: Increment resubs played count

increment_template_counter(template_name, counter_name, increment=1)[source]

Convenience method to increment a template counter

Parameters:

template_name (str) – Name of the template
counter_name (str) – Name of the counter to increment
increment (int) – Amount to increment by (default: 1)

increment_twitch_messages(username=None)[source]: Increment Twitch messages received count

initialize_with_data(all_data)[source]

Initialize the statistics manager with pre-loaded data

Parameters:: all_data (Dict[str, Any]) – Dictionary mapping database paths to their data

repair_data_integrity()[source]

Attempt to repair data integrity issues. Returns True if repairs were made.

Return type:: bool

reset_statistics(category=None)[source]: Reset statistics - optionally for specific category only

reset_template_stats(template_name=None, stat_name=None)[source]

Reset template statistics

Parameters:

template_name (Optional[str]) – Specific template name, or None for all templates
stat_name (Optional[str]) – Specific stat name, or None for all stats in the template

save_on_close()[source]: Save statistics on application close

set_save_interval(interval_seconds)[source]: Set the periodic save interval

start_periodic_saving()[source]: Start the periodic saving task (deprecated - use start_periodic_saving_safe)

start_periodic_saving_safe()[source]: Start the periodic saving using a background thread

stop_periodic_saving()[source]: Stop the periodic saving thread

submit_template_stat(template_name, stat_name, stat_value, increment=False)[source]

Submit a custom statistic from a template

Parameters:

template_name (str) – Name of the template submitting the stat
stat_name (str) – Name of the statistic
stat_value (Any) – Value of the statistic
increment (bool) – If True, increment existing value by stat_value, otherwise set to stat_value

class modules.statistics_manager.TemplateStatistics(template_name='', custom_stats=<factory>, last_updated=<factory>)[source]

Bases: object

Statistics for individual templates

custom_stats: Dict[str, Any]

last_updated: float

template_name: str = ''

class modules.statistics_manager.UserAlertStatistics(bit_alerts_played=0, resubs_played=0, new_subs_played=0, gift_subs_played=0, follow_alerts_played=0, raids=0, point_alerts_redeemed=0, channel_points_redeemed=0, donations=0, total_alerts=0, first_seen=<factory>, last_seen=<factory>)[source]

Bases: object

Per-user statistics for alert events

bit_alerts_played: int = 0

channel_points_redeemed: int = 0

donations: int = 0

first_seen: float

follow_alerts_played: int = 0

gift_subs_played: int = 0

last_seen: float

new_subs_played: int = 0

point_alerts_redeemed: int = 0

raids: int = 0

resubs_played: int = 0

total_alerts: int = 0

class modules.statistics_manager.UserChatStatistics(twitch_messages_received=0, total_messages=0, first_seen=<factory>, last_seen=<factory>)[source]

Bases: object

Per-user statistics for chat messages

first_seen: float

last_seen: float

total_messages: int = 0

twitch_messages_received: int = 0

class modules.statistics_manager.UserChatbotStatistics(commands_triggered=0, events_triggered=0, total_interactions=0, first_seen=<factory>, last_seen=<factory>)[source]

Bases: object

Per-user statistics for chatbot system

commands_triggered: int = 0

events_triggered: int = 0

first_seen: float

last_seen: float

total_interactions: int = 0

class modules.statistics_manager.UserCommandStatistics(command_usage=<factory>, first_seen=<factory>, last_seen=<factory>)[source]

Bases: object

Per-user statistics for individual commands

command_usage: Dict[str, int]

first_seen: float

last_seen: float

class modules.statistics_manager.UserConnectorStatistics(connectors_triggered=0, total_triggers=0, connector_usage=<factory>, first_seen=<factory>, last_seen=<factory>)[source]

Bases: object

Per-user statistics for connector system

connector_usage: Dict[str, int]

connectors_triggered: int = 0

first_seen: float

last_seen: float

total_triggers: int = 0

class modules.statistics_manager.UserEventStatistics(event_usage=<factory>, first_seen=<factory>, last_seen=<factory>)[source]

Bases: object

Per-user statistics for individual events

event_usage: Dict[str, int]

first_seen: float

last_seen: float

class modules.statistics_manager.UserQuoteStatistics(total_quotes_redeemed=0, individual_quote_usage=<factory>, first_seen=<factory>, last_seen=<factory>)[source]

Bases: object

Per-user statistics for quote system

first_seen: float

individual_quote_usage: Dict[str, int]

last_seen: float

total_quotes_redeemed: int = 0

modules.statistics_manager.get_statistics_manager()[source]

Get the global statistics manager instance

Return type:: StatisticsManager

modules.statistics_manager.initialize_statistics()[source]: Initialize the statistics manager

modules.statistics_manager.initialize_statistics_with_data(all_data)[source]: Initialize the statistics manager with pre-loaded data

modules.statistics_manager.shutdown_statistics()[source]: Shutdown the statistics manager

modules.statistics_manager.start_statistics_saving()[source]: Start periodic saving for the statistics manager (call after event loop is running)

Service Integration Modules

Twitch Integration

Twitch API Reference Documentation

This module contains structured information about all Twitch API endpoints including URLs, request parameters, and response structures.

All endpoints use the base URL: https://api.twitch.tv/helix/

Source: https://dev.twitch.tv/docs/api/reference/

class modules.twitch_api_reference.APIEndpoint(name, endpoint, method, description, authorization, required_params, optional_params, response_fields, rate_limits=None, auto_fill_params=None)[source]

Bases: object

Represents a Twitch API endpoint with its specifications.

authorization: List[str]

auto_fill_params: Dict[str, str] | None = None

description: str

endpoint: str

method: str

name: str

optional_params: Dict[str, str]

rate_limits: str | None = None

required_params: Dict[str, str]

response_fields: List[str]

class modules.twitch_api_reference.TwitchAPIReference[source]

Bases: object

Container for all Twitch API endpoints organized by category.

get_all_endpoints()[source]

Get all endpoints organized by category.

Return type:: Dict[str, Dict[str, APIEndpoint]]

get_base_url()[source]

Get the base URL for all Twitch API endpoints.

Return type:: str

get_endpoint_by_name(category, endpoint_name)[source]

Get a specific endpoint by category and name.

Return type:: Optional[APIEndpoint]

get_endpoints_by_method(method)[source]

Get all endpoints that use a specific HTTP method.

Return type:: List[APIEndpoint]

get_endpoints_requiring_auth(scope=None)[source]

Get all endpoints that require authentication, optionally filtered by scope.

Return type:: List[APIEndpoint]

modules.twitch_api_reference.get_endpoint_details(api_ref, category, endpoint_name)[source]: Get detailed information about a specific endpoint.

modules.twitch_api_reference.print_endpoint_summary(api_ref, category=None)[source]: Print a summary of endpoints, optionally filtered by category.

Spotify Integration

PlayStation Network

StreamLabs

OBS Studio Integration

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

Chatbot System

The Chatbot system provides comprehensive Twitch chat bot functionality including commands, events, quotes, and automated responses.

Chatbot API Integration

Chatbot Core

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.chatbot_core.ChatCommand(**kwargs)[source]

Bases: object

Represents a chat command

can_use(data)[source]

Check if command can be used, returns (can_use, reason)

Return type:: tuple[bool, str]

classmethod from_dict(data)[source]

Create from dictionary

Return type:: ChatCommand

reset_counter()[source]: Reset the command counter

should_trigger(data)[source]

Check if this command should trigger based on conditions

Return type:: bool

to_dict()[source]

Convert to dictionary for serialization

Return type:: Dict[str, Any]

use_command(data)[source]

Use the command and return the processed response

Return type:: str

class modules.chatbot_core.ChatEvent(**kwargs)[source]

Bases: object

Represents an automatic event response

classmethod from_dict(data)[source]

Create from dictionary

Return type:: ChatEvent

get_interval_string()[source]

Get interval as hh:mm:ss format string

Return type:: str

set_interval_from_string(interval_str)[source]

Set interval from hh:mm:ss format string

Return type:: bool

should_trigger(data)[source]

Check if this event should trigger based on conditions and event-specific settings

Return type:: bool

to_dict()[source]

Convert to dictionary for serialization

Return type:: Dict[str, Any]

trigger_event(data)[source]

Trigger the event and return the processed response

Return type:: str

class modules.chatbot_core.CommandType(*values)[source]

Bases: Enum

Types of chat commands

BASIC = 'basic'

COUNTER = 'counter'

RESET = 'reset'

class modules.chatbot_core.ComparisonOperator(*values)[source]

Bases: Enum

Comparison operators for conditions

CONTAINS = 'contains'

ENDS_WITH = 'ends_with'

EQUAL = 'equal'

GREATER_THAN = 'greater_than'

GREATER_THAN_OR_EQUAL = 'greater_than_or_equal'

LESS_THAN = 'less_than'

LESS_THAN_OR_EQUAL = 'less_than_or_equal'

STARTS_WITH = 'starts_with'

class modules.chatbot_core.EventType(*values)[source]

Bases: Enum

Types of automatic events

BITS = 'bits'

CHANNEL_POINT_REDEMPTION = 'channel_point_redemption'

CHAT_MESSAGE = 'chat_message'

DONATION = 'donation'

FOLLOW = 'follow'

GIFT_SUBSCRIPTION = 'gift_subscription'

HYPE_TRAIN_END = 'hype_train_end'

HYPE_TRAIN_PROGRESS = 'hype_train_progress'

HYPE_TRAIN_START = 'hype_train_start'

INTERVAL = 'interval'

RAID = 'raid'

RESUBSCRIPTION = 'resubscription'

SPECIFIC_TIME = 'specific_time'

SUBSCRIPTION = 'subscription'

class modules.chatbot_core.Greeting(**kwargs)[source]

Bases: object

Represents a user greeting in the greetings system

classmethod from_dict(data)[source]

Create from dictionary

Return type:: Greeting

get_display_text()[source]

Get formatted display text for the greeting

Return type:: str

should_greet(current_time, cooldown_hours=24)[source]

Check if user should be greeted based on time cooldown

Return type:: bool

to_dict()[source]

Convert to dictionary for serialization

Return type:: Dict[str, Any]

update_last_greeted(current_time=None)[source]: Update the last greeted timestamp

update_username(new_username)[source]: Update username and track when it was last updated

class modules.chatbot_core.Quote(**kwargs)[source]

Bases: object

Represents a quote in the quote system

format_quote()[source]

Format the quote for display

Return type:: str

classmethod from_dict(data)[source]

Create from dictionary

Return type:: Quote

get_search_text()[source]

Get text for searching (combines text and author)

Return type:: str

to_dict()[source]

Convert to dictionary for serialization

Return type:: Dict[str, Any]

class modules.chatbot_core.TriggerCondition(field, operator, value)[source]

Bases: object

Condition for triggering a command or event

evaluate(data)[source]

Evaluate the condition against provided data

Return type:: bool

class modules.chatbot_core.VariableType(*values)[source]

Bases: Enum

Types of variables that can be used in responses

AMOUNT = 'amount'

COOLDOWN = 'cooldown'

COUNT = 'count'

LEVEL = 'level'

MESSAGE = 'message'

MONTHS = 'months'

TIER = 'tier'

TIMESTAMP = 'timestamp'

USAGE_LEFT = 'usage_left'

USERNAME = 'username'

VIEWER_COUNT = 'viewer_count'

modules.chatbot_core.format_time_with_options(format_options='')[source]

Format current time with various options

Format options syntax: [timezone]:[hour_format]:[ampm_display] - timezone: UTC, EST, PST, etc. (default: local) - hour_format: 12 or 24 (default: 24) - ampm_display: show or hide (default: hide for 24hr, show for 12hr)

Examples: - “EST:12:show” -> “02:30 PM EST” - “UTC:24” -> “19:30 UTC” - “PST:12:hide” -> “11:30 PST” - “” -> “19:30” (local time, 24hr)

Return type:: str

modules.chatbot_core.get_statistics_value(stat_name)[source]

Get a statistics value by name from the StatisticsManager

Return type:: str

modules.chatbot_core.get_youtube_value(youtube_var)[source]

Get a YouTube value by variable name

Return type:: str

Chatbot Manager

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.chatbot_manager.ChatbotManager[source]

Bases: object

Manages chat commands and automatic event responses

add_command(command)[source]

Add a new command

Return type:: bool

add_event(event)[source]

Add a new event

Return type:: bool

add_greeting(user_id, username, greeting_text, enabled=True)[source]

Add a new greeting for a user

Parameters:

user_id (str) – Twitch user ID (primary key)
username (str) – Current username
greeting_text (str) – Custom greeting text
enabled (bool) – Whether greeting is enabled

Return type:

tuple[bool, str, str]

Returns:

tuple of (success, error_message, greeting_id)

add_quote(text, author='', added_by='')[source]

Add a new quote

Parameters:

text (str) – The quote text
author (str) – The author of the quote (optional)
added_by (str) – Username of who added the quote

Return type:

tuple[bool, str, int]

Returns:

tuple of (success, error_message, quote_number)

cleanup()[source]: Cleanup resources

delete_quote(quote_id)[source]

Delete a quote by ID

Return type:: bool

delete_quote_by_number(quote_number)[source]

Delete a quote by its number

Return type:: bool

get_all_commands()[source]

Get all commands

Return type:: Dict[str, ChatCommand]

get_all_events()[source]

Get all events

Return type:: Dict[str, ChatEvent]

get_all_greetings()[source]

Get all greetings

Return type:: Dict[str, Greeting]

get_all_quotes()[source]

Get all quotes

Return type:: Dict[str, Quote]

get_command(command_id)[source]

Get a command by ID

Return type:: Optional[ChatCommand]

get_custom_greeting_enabled()[source]

Check if custom greetings are enabled

Return type:: bool

get_default_greeting()[source]

Get the default greeting text

Return type:: str

get_default_greeting_enabled()[source]

Check if default greetings are enabled

Return type:: bool

get_event(event_id)[source]

Get an event by ID

Return type:: Optional[ChatEvent]

get_greeting(greeting_id)[source]

Get a greeting by ID

Return type:: Optional[Greeting]

get_greeting_by_user_id(user_id)[source]

Get a greeting by user ID (fast lookup)

Return type:: Optional[Greeting]

get_greeting_cooldown()[source]

Get the greeting cooldown in hours

Return type:: int

get_greeting_flag_stats()[source]

Get greeting flag file statistics

Return type:: Dict[str, Any]

get_greeting_reset_interval()[source]

Get the greeting reset interval in hours

Return type:: int

get_greetings_enabled()[source]

Check if greetings system is enabled

Return type:: bool

get_greetings_statistics()[source]

Get greetings statistics

Return type:: Dict[str, Any]

get_manager()[source]: Get the manager instance (for compatibility)

get_quote_by_id(quote_id)[source]

Get a quote by its ID

Return type:: Optional[Quote]

get_quote_by_number(quote_number)[source]

Get a quote by its number

Return type:: Optional[Quote]

get_quote_cooldown()[source]

Get the global quote cooldown in seconds

Return type:: int

get_quotes_enabled()[source]

Check if quotes system is enabled

Return type:: bool

get_quotes_statistics()[source]

Get quotes statistics

Return type:: Dict[str, Any]

get_random_quote()[source]

Get a random quote

Return type:: Optional[Quote]

get_statistics()[source]

Get chatbot statistics

Return type:: Dict[str, Any]

process_chat_message(message_data)[source]

Process a chat message for commands

Parameters:: message_data (Dict[str, Any]) – Dictionary containing message information
Return type:: Optional[tuple[str, str]]
Returns:: tuple of (response_message, command_name) if command triggered, None otherwise

process_event(event_type, event_data)[source]

Process an automatic event

Parameters:

event_type (EventType) – Type of event that occurred
event_data (Dict[str, Any]) – Dictionary containing event information

Return type:

Optional[str]

Returns:

Response message if event triggered, None otherwise

process_greeting(user_id, username, current_time=None)[source]

Process a greeting for a user

Parameters:

user_id (str) – Twitch user ID
username (str) – Current username
current_time (float) – Current timestamp (optional)

Return type:

Optional[str]

Returns:

Greeting message if user should be greeted, None otherwise

remove_command(command_id)[source]

Remove a command

Return type:: bool

remove_event(event_id)[source]

Remove an event

Return type:: bool

remove_greeting(greeting_id)[source]

Remove a greeting

Return type:: bool

reset_all_greeting_flags()[source]: Reset all greeting flags (both custom and default)

reset_command_counter(command_id)[source]

Reset a command’s counter

Return type:: bool

search_quotes(search_text, limit=5)[source]

Search quotes by text content

Parameters:

search_text (str) – Text to search for
limit (int) – Maximum number of results to return

Return type:

List[Quote]

Returns:

List of matching quotes ordered by relevance

set_greeting_reset_interval(hours)[source]

Set the greeting reset interval in hours

Return type:: bool

set_quote_cooldown(cooldown_seconds)[source]

Set the global quote cooldown in seconds

Return type:: bool

start_deferred_interval_events()[source]: Start any interval events that were deferred due to chatbot not being ready

async test_command(command_id, test_data)[source]

Test a command with sample data and send message to Twitch chat

Return type:: Dict[str, Any]

async test_event(event_id, test_data)[source]

Test an event with sample data and send message to Twitch chat

Return type:: Dict[str, Any]

toggle_custom_greeting_enabled(enabled)[source]

Toggle whether custom greetings are enabled

Return type:: bool

toggle_default_greeting_enabled(enabled)[source]

Toggle whether default greetings are enabled

Return type:: bool

toggle_greetings_enabled(enabled)[source]

Toggle whether the greetings system is enabled (legacy method)

Return type:: bool

toggle_item(item_id, enabled)[source]

Toggle enabled state of a command or event

Return type:: bool

toggle_quotes_enabled(enabled)[source]

Toggle whether the quote system is enabled

Return type:: bool

update_command(command_id, command)[source]

Update an existing command

Return type:: bool

update_default_greeting(greeting_text)[source]

Update the default greeting text

Return type:: bool

update_event(event_id, event)[source]

Update an existing event

Return type:: bool

update_greeting(greeting_id, greeting_text=None, enabled=None)[source]

Update an existing greeting

Return type:: bool

update_greeting_cooldown(hours)[source]

Update the greeting cooldown in hours

Return type:: bool

update_greeting_username(user_id, new_username)[source]

Update username for a greeting (when username changes)

Return type:: bool

class modules.chatbot_manager.GreetingFlagManager(file_path=None)[source]

Bases: object

Manages greeting flags in a local file for simple boolean-based greeting system

force_save()[source]: Force an immediate save of flags

get_stats()[source]

Get statistics about the flags file

Return type:: Dict[str, Any]

mark_user_greeted(user_id)[source]: Mark a user as having been greeted

perform_reset_if_needed(reset_interval_hours, current_time=None)[source]: Check and perform reset if needed

reset_all_flags()[source]: Reset all greeting flags to False and update reset timestamp

should_greet_user(user_id)[source]

Check if a user should be greeted (returns True if not yet greeted)

Return type:: bool

should_reset_flags(reset_interval_hours, current_time=None)[source]

Check if greeting flags should be reset based on time interval

Return type:: bool

modules.chatbot_manager.get_manager()[source]

Get the global chatbot manager instance

Return type:: ChatbotManager

Alert System

Alert Processing

Alert Utilities

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.alertutils.AlertObj(duration=0.0, alert_name='', display_name='', alert_type='', deleted=False, alert_id='', played=False, stackable=False, timestamp=0.0, skip_alert=False, is_replay=False, is_test=False, username='', anonymous=False, message=None, emotes=None, title=None, tier=0, gift_qty=0, resub_month=0, months_prepaid=0, amt_cheered=0, twitch_reward_id=None, point_cost=0, enable_alert=False, raider_count=0, donation_amount=0.0, currency='USD', hype_train_level=0, hype_train_in_progress=False, fade_in=0, fade_out=0, volume=0, audio_only=False, single_audio_dir=None, single_audio_name=None, gif_dir=None, gif_name=None, randomized=False, randomized_dir=None, randomized_chance=0, randomized_extra=False, randomized_extra_chance=0, randomized_extra_dir=None)[source]

Bases: object

alert_id: str = ''

alert_name: str = ''

alert_type: str = ''

amt_cheered: int = 0

anonymous: bool = False

audio_only: bool = False

currency: str = 'USD'

deleted: bool = False

display_name: str = ''

donation_amount: float = 0.0

duration: float = 0.0

emotes: str = None

enable_alert: bool = False

fade_in: int = 0

fade_out: int = 0

gif_dir: str = None

gif_name: str = None

gift_qty: int = 0

hype_train_in_progress: bool = False

hype_train_level: int = 0

is_replay: bool = False

is_test: bool = False

message: str = None

months_prepaid: int = 0

played: bool = False

point_cost: int = 0

raider_count: int = 0

randomized: bool = False

randomized_chance: int = 0

randomized_dir: str = None

randomized_extra: bool = False

randomized_extra_chance: int = 0

randomized_extra_dir: str = None

resub_month: int = 0

single_audio_dir: str = None

single_audio_name: str = None

skip_alert: bool = False

stackable: bool = False

tier: int = 0

timestamp: float = 0.0

title: str = None

twitch_reward_id: str = None

username: str = ''

volume: int = 0

class modules.alertutils.AlertSettings(global_delay=3, default_bit_alert='bit1', default_sub_alert='sub1', default_follow_alert='follow1', default_point_alert='point1', default_raids_alert='raid1', default_giftsub_alert='giftsub1', default_donation_alert='donation1', FALLBACK_ALERT_ID='subs_fallback')[source]

Bases: object

FALLBACK_ALERT_ID: str = 'subs_fallback'

alert_types = ['sub', 'bit', 'follow', 'point', 'raid', 'giftsub', 'donation', 'hype_train_start', 'hype_train_progress', 'hype_train_end']

default_bit_alert: str = 'bit1'

default_donation_alert: str = 'donation1'

default_follow_alert: str = 'follow1'

default_giftsub_alert: str = 'giftsub1'

default_point_alert: str = 'point1'

default_raids_alert: str = 'raid1'

default_sub_alert: str = 'sub1'

global_delay: int = 3

class modules.alertutils.AlertStateManager[source]

Bases: object

Manages the state of all alert configurations and provides methods to access and update them.

This class serves as the central source of truth for all alert configurations, handling both the in-memory state and synchronization with Firebase.

delete_alert(alert_type, alert_id)[source]

Delete an alert from the state and sync with Firebase

Parameters:

alert_type (str) – The type of alert (bits, subs, etc.)
alert_id (str) – The ID of the alert

Returns:

True if successful, False otherwise

Return type:

bool

get_alert_by_id(alert_type, alert_id)[source]

Get a specific alert from the state

Parameters:

alert_type (str) – The type of alert (bits, subs, etc.)
alert_id (str) – The ID of the alert

Returns:

The alert data or None if not found

Return type:

dict

get_alerts_by_type(alert_type, include_ranges=True)[source]

Get alerts of a specific type from the state

Parameters:

alert_type (str) – The type of alert (bits, subs, etc.)
include_ranges (bool) – Whether to include range alerts

Returns:

Dictionary with alerts of the specified type

Return type:

dict

get_alerts_for_type(alert_type)[source]

Get all alerts for a specific type from the state

Parameters:: alert_type (str) – The type of alert (bits, subs, etc.)
Returns:: Dictionary of alerts for the specified type
Return type:: dict

get_all_alerts()[source]

Get all alerts from the state, including alert configurations and stored alerts

Returns:: Dictionary with all alerts including ‘alert_storage’ key
Return type:: dict

get_default_alert_id(alert_type)[source]

Get the default alert ID for a specific type

Parameters:: alert_type (str) – The type of alert (bits, subs, etc.)
Returns:: The default alert ID for the specified type
Return type:: str

get_display_name(alert_type, alert_id, alert_data=None)[source]

Generate a consistent display name for an alert based on type and ID

Parameters:

alert_type (str) – The type of alert (bits, subs, etc.)
alert_id (str) – The ID of the alert
alert_data (dict) – The alert data (used for point alerts)

Returns:

Formatted display name for the alert

Return type:

str

get_limited_stored_alerts_from_firebase(max_alerts)[source]

Get only the most recent X alerts directly from Firebase to reduce bandwidth

Parameters:: max_alerts (int) – Maximum number of alerts to fetch from Firebase
Returns:: Dictionary of the most recent stored alerts (limited to max_alerts)
Return type:: dict

get_limited_stored_alerts_paginated(page=1, limit=25, max_total_alerts=250)[source]

Get stored completed alerts with pagination, limited to a maximum total number of alerts

Parameters:

page (int) – Page number (1-based)
limit (int) – Number of alerts per page
max_total_alerts (int) – Maximum total number of alerts to consider

Returns:

Dictionary containing:

alerts: List of alert data sorted by timestamp (newest first)
total_count: Total number of stored alerts (limited to max_total_alerts)
page: Current page number
limit: Items per page
total_pages: Total number of pages
has_next: Whether there’s a next page
has_prev: Whether there’s a previous page

Return type:

dict

get_recent_stored_alerts(limit=25)[source]

Get the most recent stored alerts

Parameters:: limit (int) – Maximum number of alerts to return
Returns:: List of recent alerts sorted by timestamp (newest first)
Return type:: list

get_resub_fallback_enabled()[source]

Check if the resub fallback alert is enabled.

Returns:: True if the resub fallback alert is enabled, False otherwise.
Return type:: bool

get_stored_alert_by_id(alert_id)[source]

Get a specific stored alert by its ID

Parameters:: alert_id (str) – The ID of the stored alert to retrieve
Returns:: The stored alert data or None if not found
Return type:: Optional[dict]

get_stored_alerts()[source]

Get all stored completed alerts

Returns:: Dictionary of stored alerts
Return type:: dict

get_stored_alerts_paginated(page=1, limit=25)[source]

Get stored completed alerts with pagination

Parameters:

page (int) – Page number (1-based)
limit (int) – Number of alerts per page

Returns:

Dictionary containing:

alerts: List of alert data sorted by timestamp (newest first)
total_count: Total number of stored alerts
page: Current page number
limit: Items per page
total_pages: Total number of pages
has_next: Whether there’s a next page
has_prev: Whether there’s a previous page

Return type:

dict

initialize()[source]: Initialize the alert state by loading all alerts from Firebase

async initialize_async()[source]: Initialize the alert state by loading all alerts from Firebase asynchronously

initialize_with_data(all_data)[source]

Initialize the alert state with pre-loaded data

Parameters:: all_data (Dict[str, Any]) – Dictionary mapping database paths to their data

migrate_individual_alerts_to_collections()[source]

Migrate individual alert records to collection format

This method finds all individual alert records in the database and consolidates them into their respective collections.

Returns:: True if successful, False otherwise
Return type:: bool

reload_from_firebase()[source]: Force a reload of all alerts from Firebase

save_alert(alert_type, alert_id, alert_data, update_collection=True)[source]

Save an alert to the state and sync with Firebase

Parameters:

alert_type (str) – The type of alert (bits, subs, etc.)
alert_id (str) – The ID of the alert
alert_data (dict) – The alert data to save
update_collection (bool) – Whether to update the collection after saving (default True)

Returns:

True if successful, False otherwise

Return type:

bool

save_alert_by_id(alert_id)[source]

Save an alert by its ID to Firebase using the current data in state manager

Parameters:: alert_id (str) – The ID of the alert to save (format: “bits100”, “subs1”, etc.)
Returns:: True if successful, False otherwise
Return type:: bool

set_resub_fallback_enabled(enabled)[source]

Set the resub fallback alert enabled state.

Parameters:: enabled (bool) – Whether the resub fallback alert should be enabled.
Returns:: True if the setting was saved successfully.
Return type:: bool

store_completed_alert(alert_id, alert_data)[source]

Store a completed alert in the alert storage

Parameters:

alert_id (str) – The ID of the completed alert
alert_data (dict) – The alert data to store (should be from AlertObj.__dict__)

Returns:

True if successful, False otherwise

Return type:

bool

sync_to_firebase()[source]: Force a sync of all alerts to Firebase

update_alert(alert_type, alert_id, alert_data)[source]

Update an alert in the alert state and sync with Firebase

Parameters:

alert_type (str) – The type of alert (bits, subs, etc.)
alert_id (str) – The ID of the alert
alert_data (dict) – The alert data

update_alert_collections(alert_type)[source]

Update the alert collections for a specific alert type after batch operations

Parameters:: alert_type (str) – The type of alert (bits, subs, etc.)
Returns:: True if successful, False otherwise
Return type:: bool

update_alert_data(alert_id, alert_data)[source]

Update alert data in the state manager and save to Firebase

Parameters:

alert_id (str) – The ID of the alert to update (format: “bits100”, “subs1”, etc.)
alert_data (dict) – The alert data to update

Returns:

True if successful, False otherwise

Return type:

bool

modules.alertutils.fetch_bits_alert(quantity)[source]

Get bits alert data based on quantity. Checks for exact match in BitAlerts, otherwise falls back to range alerts. If no match is found, returns the default bit alert.

Parameters:: quantity (int) – The number of bits
Return type:: Optional[AlertObj]
Returns:: AlertObj

modules.alertutils.fetch_cheer_alert(quantity)[source]

Alias for fetch_bits_alert.

Parameters:: quantity (int) – The number of bits
Return type:: AlertObj
Returns:: AlertObj

modules.alertutils.fetch_donation_alert(quantity)[source]

Get donation alert data based on quantity. Checks for exact match in DonationAlerts, otherwise falls back to range alerts. If no match is found, returns the default donation alert.

Parameters:: quantity (int) – The donation amount
Return type:: Optional[AlertObj]
Returns:: AlertObj

modules.alertutils.fetch_follow_alert()[source]

Get follow alert data. Since there’s only one follow alert, it simply returns the first available follow alert or the default follow alert.

Return type:: Optional[AlertObj]
Returns:: AlertObj

modules.alertutils.fetch_giftsub_alert(quantity)[source]

Get giftsub alert data based on quantity. Checks for exact match in GiftsubAlerts, otherwise falls back to range alerts. If no match is found, returns the default giftsub alert.

Parameters:: quantity (int) – The number of gift subs
Return type:: Optional[AlertObj]
Returns:: AlertObj

modules.alertutils.fetch_point_alert(twitch_reward_id)[source]

Get channel point redemption alert data based on reward ID. Returns only exact matches for the twitch_reward_id.

Parameters:: twitch_reward_id (str) – The Twitch reward ID
Return type:: Optional[AlertObj]
Returns:: AlertObj if exact match found, None otherwise

modules.alertutils.fetch_raid_alert(raider_count)[source]

Get raid alert data based on raider count. Checks for exact match in RaidAlerts, otherwise falls back to range alerts. If no match is found, returns the default raid alert.

Parameters:: raider_count (int) – The number of raiders
Return type:: Optional[AlertObj]
Returns:: AlertObj

modules.alertutils.fetch_resub_alert(months)[source]

Get resubscription alert data based on months.

Checks for an exact month match first. If none is found and the resub fallback alert is enabled, uses the fallback alert. Otherwise falls back to the default sub alert (sub1 / first available).

Parameters:: months (int) – The cumulative subscription month count
Return type:: Optional[AlertObj]
Returns:: AlertObj or None

modules.alertutils.fetch_sub_alert(months)[source]

Get subscription alert data based on months. If no match is found, returns the default sub alert.

Parameters:: months (int) – The subscription month count (fixed at 1 for this alert)
Return type:: Optional[AlertObj]
Returns:: AlertObj

modules.alertutils.initialize_alert_state()[source]: Initialize the global alert state manager

async modules.alertutils.initialize_alert_state_async()[source]: Initialize the global alert state manager asynchronously

modules.alertutils.initialize_alert_state_with_data(all_data)[source]: Initialize the global alert state manager with pre-loaded data

modules.alertutils.load_alerts()[source]

Alert Parsing

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.alerts_parser.AlertData(alerttype, audioname, audiodirectory, duration, volume, randomized, randomizedchance, randomizeddirectory, randomizedextrachance, randomizedextradirectory, twitchrewardid, gifname, gifdirectory, alert_id='', alert_category='', alertname=None, disabled=False)[source]

Bases: object

Data class to hold parsed alert information

alert_category: str = ''

alert_id: str = ''

alertname: str | None = None

alerttype: str

audiodirectory: str

audioname: str

disabled: bool = False

duration: str | int

gifdirectory: str | None

gifname: str | None

randomized: str

randomizedchance: str | None

randomizeddirectory: str | None

randomizedextrachance: str | None

randomizedextradirectory: str | None

twitchrewardid: str | None

volume: str | int

class modules.alerts_parser.AlertsParser(file_path=None)[source]

Bases: object

Parser for alerts.json files used in Twitch alert systems.

This class provides methods to parse and extract specific alert configuration data from JSON files, making it easy to work with alert settings programmatically.

export_filtered_data(alerts, format='dict')[source]

Export alert data in different formats

Parameters:

alerts (List[AlertData]) – List of alerts to export
format (str) – Export format (“dict” or “dataclass”)

Return type:

Union[List[Dict], List[AlertData]]

Returns:

Exported data in requested format

get_alert_by_id(alert_id)[source]

Get a specific alert by its ID

Parameters:: alert_id (str) – The alert ID to search for
Returns:: The alert data or None if not found
Return type:: Optional[AlertData]

get_alerts_by_category(category)[source]

Get all alerts from a specific category

Parameters:: category (str) – Category name (e.g., “bit_alerts”, “point_alerts”)
Returns:: Filtered list of alerts
Return type:: List[AlertData]

get_alerts_by_type(alerttype)[source]

Get all alerts of a specific type

Parameters:: alerttype (str) – Type of alert (e.g., “bits”, “points”, “sub”, etc.)
Returns:: Filtered list of alerts
Return type:: List[AlertData]

get_alerts_with_audio()[source]

Get all alerts that have audio files configured

Returns:: Alerts with audio configuration
Return type:: List[AlertData]

get_alerts_with_gifs()[source]

Get all alerts that have GIF files configured

Returns:: Alerts with GIF configuration
Return type:: List[AlertData]

get_randomized_alerts()[source]

Get all alerts that have randomization enabled

Returns:: Alerts with randomization enabled
Return type:: List[AlertData]

get_summary_stats()[source]

Get summary statistics about the parsed alerts

Returns:: Summary statistics
Return type:: Dict[str, Any]

get_twitch_reward_alerts()[source]

Get all alerts associated with Twitch channel point rewards

Returns:: Alerts with Twitch reward IDs
Return type:: List[AlertData]

load_file(file_path=None)[source]

Load and parse the alerts JSON file

Parameters:: file_path (Optional[str]) – Path to the JSON file. If None, uses instance file_path.
Returns:: True if successful, False otherwise
Return type:: bool

parse_alerts()[source]

Parse all alerts from the loaded JSON data

Returns:: List of parsed alert data objects
Return type:: List[AlertData]

modules.alerts_parser.main()[source]: Example usage of the AlertsParser

User Interface

Main UI Window

Web Engine

UI Windows

Connector Automation System

The Connector system provides a powerful automation framework for creating trigger-action workflows that respond to stream events automatically.

Core Connector Framework

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.connector_core.ActionType(*values)[source]

Bases: Enum

Types of actions available in the connector system

ADD_GREETING = 'add_greeting'

API_CALL = 'api_call'

AUDIO_CONTROL = 'audio_control'

EXECUTE_COMMAND = 'execute_command'

KEY_PRESS = 'key_press'

SEND_CHAT_MESSAGE = 'send_chat_message'

SEND_GREETING = 'send_greeting'

TEMPLATE_CONTROL = 'template_control'

TRIGGER_ALERT = 'trigger_alert'

UPDATE_GREETING = 'update_greeting'

WEBSOCKET_EMIT = 'websocket_emit'

WRITE_FILE = 'write_file'

class modules.connector_core.BaseAction(action_id, action_type, name, description='', enabled=True, parameters=<factory>, metadata=<factory>)[source]

Bases: ABC

Base class for all action types

action_id: str

action_type: ActionType

description: str = ''

enabled: bool = True

abstractmethod async execute(trigger_data, event_data)[source]

Execute this action

Return type:: bool

metadata: Dict[str, Any]

name: str

parameters: Dict[str, Any]

validate_parameters()[source]

Validate that required parameters are present and valid

Return type:: bool

class modules.connector_core.BaseTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>)[source]

Bases: ABC

Base class for all trigger types

can_trigger()[source]

Check if trigger is enabled and not on cooldown

Return type:: bool

conditions: List[TriggerCondition]

cooldown_seconds: int = 0

description: str = ''

enabled: bool = True

evaluate_conditions(event_data)[source]

Evaluate all conditions for this trigger

Return type:: bool

last_triggered: float = 0

metadata: Dict[str, Any]

name: str

abstractmethod should_trigger(event_data)[source]

Determine if this trigger should fire based on event data

Return type:: bool

trigger(event_data)[source]

Attempt to trigger this trigger

Return type:: bool

trigger_id: str

trigger_type: TriggerType

class modules.connector_core.ComparisonOperator(*values)[source]

Bases: Enum

Comparison operators for trigger conditions

CONTAINS = 'contains'

ENDS_WITH = 'ends_with'

EQUAL = 'equal'

GREATER_THAN = 'greater_than'

GREATER_THAN_OR_EQUAL = 'greater_than_or_equal'

LESS_THAN = 'less_than'

LESS_THAN_OR_EQUAL = 'less_than_or_equal'

NOT_CONTAINS = 'not_contains'

NOT_EQUAL = 'not_equal'

REGEX_MATCH = 'regex_match'

STARTS_WITH = 'starts_with'

class modules.connector_core.Connector(connector_id, name, description='', enabled=True, trigger=None, actions=<factory>, created_at=<factory>, last_triggered=0, trigger_count=0, metadata=<factory>)[source]

Bases: object

A connector that links a trigger to one or more actions

actions: List[BaseAction]

connector_id: str

created_at: float

description: str = ''

enabled: bool = True

last_triggered: float = 0

metadata: Dict[str, Any]

name: str

async process_event(event_data)[source]

Process an event and potentially trigger actions

Return type:: bool

trigger: BaseTrigger = None

trigger_count: int = 0

class modules.connector_core.EventData[source]

Bases: object

Standardized event data structure for the connector system

static from_donation(amount, username, message='', currency='USD', **kwargs)[source]

Create event data from donation event

Return type:: Dict[str, Any]

static from_twitch_bits(bits_amount, username, message='', **kwargs)[source]

Create event data from Twitch bits event

Return type:: Dict[str, Any]

static from_twitch_chat(username, message, is_command=False, command='', **kwargs)[source]

Create event data from Twitch chat message

Return type:: Dict[str, Any]

static from_twitch_follow(username, **kwargs)[source]

Create event data from Twitch follow event

Return type:: Dict[str, Any]

static from_twitch_raid(username, viewer_count, **kwargs)[source]

Create event data from Twitch raid event

Return type:: Dict[str, Any]

static from_twitch_sub(tier, username, message='', months=1, **kwargs)[source]

Create event data from Twitch subscription event

Return type:: Dict[str, Any]

class modules.connector_core.TriggerCondition(field, operator, value, case_sensitive=True)[source]

Bases: object

A single condition that must be met for a trigger to fire

case_sensitive: bool = True

evaluate(event_data)[source]

Evaluate this condition against event data

Return type:: bool

field: str

operator: ComparisonOperator

value: Any

class modules.connector_core.TriggerType(*values)[source]

Bases: Enum

Types of triggers available in the connector system

DONATION = 'donation'

HOTKEY = 'hotkey'

SCHEDULE = 'schedule'

TIMER = 'timer'

TWITCH_BITS = 'twitch_bits'

TWITCH_CHAT_MESSAGE = 'twitch_chat_message'

TWITCH_FOLLOW = 'twitch_follow'

TWITCH_GIFTSUB = 'twitch_giftsub'

TWITCH_HYPE_TRAIN_END = 'twitch_hype_train_end'

TWITCH_HYPE_TRAIN_START = 'twitch_hype_train_start'

TWITCH_POINTS = 'twitch_points'

TWITCH_RAID = 'twitch_raid'

TWITCH_RESUB = 'twitch_resub'

TWITCH_SUB = 'twitch_sub'

WEBHOOK = 'webhook'

Connector Manager

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.connector_manager.ConnectorManager[source]

Bases: object

Manages connectors and processes events through the trigger-action system

add_connector(connector)[source]

Add a new connector

Return type:: bool

async add_event(event_data)[source]: Add an event to the processing queue

ensure_processing_task()[source]: Ensure the processing task is running if we have an event loop

get_all_connectors()[source]

Get all connectors

Return type:: Dict[str, Connector]

get_connector(connector_id)[source]

Get a connector by ID

Return type:: Optional[Connector]

get_connectors_by_trigger_type(trigger_type)[source]

Get connectors with a specific trigger type

Return type:: List[Connector]

get_statistics()[source]

Get connector manager statistics

Return type:: Dict[str, Any]

load_connectors()[source]: Load connectors from database

async process_event(event_data)[source]

Process an event through all enabled connectors

Return type:: bool

remove_connector(connector_id)[source]

Remove a connector

Return type:: bool

save_connectors()[source]: Save connectors to database

start()[source]: Start the connector manager

start_connector_thread()[source]: Start connector processing in a separate thread with its own event loop

async stop()[source]: Stop the connector manager

async test_connector(connector_id, test_event_data)[source]

Test a connector with provided event data

Return type:: Dict[str, Any]

toggle_connector(connector_id)[source]

Toggle a connector’s enabled state

Return type:: bool

update_connector(connector_id, connector)[source]

Update an existing connector

Return type:: bool

async modules.connector_manager.cleanup()[source]: Cleanup the connector system

modules.connector_manager.get_manager()[source]

Get the global connector manager instance

Return type:: ConnectorManager

modules.connector_manager.initialize()[source]: Initialize the connector system

Trigger System

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.connector_triggers.DonationTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>)[source]

Bases: BaseTrigger

Trigger for donation events from StreamLabs

should_trigger(event_data)[source]

Check if this donation trigger should fire

Return type:: bool

class modules.connector_triggers.HotkeyTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>, key_combination='', modifiers=None, key_code='', is_global=True)[source]

Bases: BaseTrigger

Trigger for hotkey/key press activation

is_global: bool = True

key_code: str = ''

key_combination: str = ''

modifiers: List[str] = None

should_trigger(event_data)[source]

Check if this hotkey trigger should fire

Return type:: bool

class modules.connector_triggers.ScheduleTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>, schedule_pattern='', timezone='UTC')[source]

Bases: BaseTrigger

Trigger for scheduled events (cron-like)

schedule_pattern: str = ''

should_trigger(event_data)[source]

Check if this scheduled trigger should fire

Return type:: bool

timezone: str = 'UTC'

class modules.connector_triggers.TimerTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>, interval_seconds=60)[source]

Bases: BaseTrigger

Trigger for timer-based events

interval_seconds: int = 60

should_trigger(event_data)[source]

Check if this timer trigger should fire

Return type:: bool

class modules.connector_triggers.TwitchBitsTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>)[source]

Bases: BaseTrigger

Trigger for Twitch bits/cheers events

should_trigger(event_data)[source]

Check if this bits trigger should fire

Return type:: bool

class modules.connector_triggers.TwitchChatMessageTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>)[source]

Bases: BaseTrigger

Trigger for Twitch chat message events

should_trigger(event_data)[source]

Check if this chat message trigger should fire

Return type:: bool

class modules.connector_triggers.TwitchFollowTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>)[source]

Bases: BaseTrigger

Trigger for Twitch follow events

should_trigger(event_data)[source]

Check if this follow trigger should fire

Return type:: bool

class modules.connector_triggers.TwitchGiftSubTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>)[source]

Bases: BaseTrigger

Trigger for Twitch gift subscription events

should_trigger(event_data)[source]

Check if this gift subscription trigger should fire

Return type:: bool

class modules.connector_triggers.TwitchHypeTrainEndTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>)[source]

Bases: BaseTrigger

Trigger for Twitch hype train end events

should_trigger(event_data)[source]

Check if this hype train end trigger should fire

Return type:: bool

class modules.connector_triggers.TwitchHypeTrainStartTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>)[source]

Bases: BaseTrigger

Trigger for Twitch hype train start events

should_trigger(event_data)[source]

Check if this hype train start trigger should fire

Return type:: bool

class modules.connector_triggers.TwitchPointsTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>)[source]

Bases: BaseTrigger

Trigger for Twitch channel points redemption events

should_trigger(event_data)[source]

Check if this points trigger should fire

Return type:: bool

class modules.connector_triggers.TwitchRaidTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>)[source]

Bases: BaseTrigger

Trigger for Twitch raid events

should_trigger(event_data)[source]

Check if this raid trigger should fire

Return type:: bool

class modules.connector_triggers.TwitchResubTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>)[source]

Bases: BaseTrigger

Trigger for Twitch resubscription events

should_trigger(event_data)[source]

Check if this resubscription trigger should fire

Return type:: bool

class modules.connector_triggers.TwitchSubTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>)[source]

Bases: BaseTrigger

Trigger for Twitch subscription events

should_trigger(event_data)[source]

Check if this subscription trigger should fire

Return type:: bool

class modules.connector_triggers.WebhookTrigger(trigger_id, trigger_type, name, description='', enabled=True, conditions=<factory>, cooldown_seconds=0, last_triggered=0, metadata=<factory>, webhook_url='', secret_key='')[source]

Bases: BaseTrigger

Trigger for webhook events

secret_key: str = ''

should_trigger(event_data)[source]

Check if this webhook trigger should fire

Return type:: bool

webhook_url: str = ''

modules.connector_triggers.create_trigger(trigger_type, trigger_id, name, **kwargs)[source]

Factory function to create trigger instances

Return type:: BaseTrigger

Action System

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.connector_actions.AddGreetingAction(action_id, action_type, name, description='', enabled=True, parameters=<factory>, metadata=<factory>, user_id='', username='', greeting_text='')[source]

Bases: BaseAction

Action to add a new greeting for a user

enabled: bool = True

async execute(trigger_data, event_data)[source]

Execute add greeting action

Return type:: bool

greeting_text: str = ''

user_id: str = ''

username: str = ''

validate_parameters()[source]

Validate add greeting action parameters

Return type:: bool

class modules.connector_actions.ApiCallAction(action_id, action_type, name, description='', enabled=True, parameters=<factory>, metadata=<factory>, url='', method='GET', headers=None, body=None)[source]

Bases: BaseAction

Action to make an HTTP API call

body: Dict[str, Any] = None

async execute(trigger_data, event_data)[source]

Execute API call action

Return type:: bool

headers: Dict[str, str] = None

method: str = 'GET'

url: str = ''

validate_parameters()[source]

Validate API call action parameters

Return type:: bool

class modules.connector_actions.AudioChangeEntry(original_values, current_values, remaining_duration, restoration_task=None, task_id='', last_update=0.0)[source]

Bases: object

Entry for tracking active audio changes in the global registry

current_values: Dict[str, Any]

last_update: float = 0.0

original_values: Dict[str, Any]

remaining_duration: float

reset_duration(new_duration)[source]: Reset duration (for random actions)

restoration_task: Task | None = None

task_id: str = ''

update_remaining_duration(new_duration)[source]: Update remaining duration (add to existing if stacking)

class modules.connector_actions.AudioControlAction(action_id, action_type, name, description='', enabled=True, parameters=<factory>, metadata=<factory>, control_type='device', action_mode='set', volume_level=50.0, volume_step=10.0, target_application='', target_device='', device_name='', duration=0.0, _active_restoration_tasks=None, _audio_change_registry=None)[source]

Bases: BaseAction

Action to control system audio (volume, microphone, etc.)

action_mode: str = 'set'

add_restoration_task(task_id, task)[source]: Add a restoration task to the active tasks

classmethod cancel_all_restoration_tasks()[source]: Cancel all active restoration tasks

classmethod cleanup_audio_change_registry()[source]: Clean up the audio change registry

control_type: str = 'device'

device_name: str = ''

duration: float = 0.0

async execute(trigger_data, event_data)[source]

Execute audio control action

Return type:: bool

get_audio_source_key(target_app, target_device, device_name)[source]

Generate a unique key for the audio source

Return type:: str

async get_or_create_registry_entry(source_key, target_app, target_device, device_name)[source]

Get existing registry entry or create a new one

Return type:: AudioChangeEntry

remove_restoration_task(task_id)[source]: Remove a restoration task from the active tasks

target_application: str = ''

target_device: str = ''

update_registry_entry(source_key, new_values, duration, is_random=False)[source]: Update registry entry with new values and duration

volume_level: float = 50.0

volume_step: float = 10.0

class modules.connector_actions.ExecuteCommandAction(action_id, action_type, name, description='', enabled=True, parameters=<factory>, metadata=<factory>, command='', working_directory='', timeout_seconds=30)[source]

Bases: BaseAction

Action to execute a system command

command: str = ''

async execute(trigger_data, event_data)[source]

Execute command action

Return type:: bool

timeout_seconds: int = 30

validate_parameters()[source]

Validate execute command action parameters

Return type:: bool

working_directory: str = ''

class modules.connector_actions.KeyPressAction(action_id, action_type, name, description='', enabled=True, parameters=<factory>, metadata=<factory>, input_type='key', action_mode='press', key_sequence='', repeat_count=1, repeat_interval=0.1, hold_duration=0.5, macro_sequence='')[source]

Bases: BaseAction

Action to simulate keyboard and mouse input

action_mode: str = 'press'

async execute(trigger_data, event_data)[source]

Execute key press action

Return type:: bool

hold_duration: float = 0.5

input_type: str = 'key'

key_sequence: str = ''

macro_sequence: str = ''

repeat_count: int = 1

repeat_interval: float = 0.1

class modules.connector_actions.SendChatMessageAction(action_id, action_type, name, description='', enabled=True, parameters=<factory>, metadata=<factory>, message='')[source]

Bases: BaseAction

Action to send a message to Twitch chat

async execute(trigger_data, event_data)[source]

Execute send chat message action

Return type:: bool

message: str = ''

validate_parameters()[source]

Validate send chat message action parameters

Return type:: bool

class modules.connector_actions.SendGreetingAction(action_id, action_type, name, description='', enabled=True, parameters=<factory>, metadata=<factory>, user_id='', username='', force_send=True)[source]

Bases: BaseAction

Action to send a greeting to a user (even if already sent)

async execute(trigger_data, event_data)[source]

Execute send greeting action

Return type:: bool

force_send: bool = True

user_id: str = ''

username: str = ''

validate_parameters()[source]

Validate send greeting action parameters

Return type:: bool

class modules.connector_actions.TemplateControlAction(action_id, action_type, name, description='', enabled=True, parameters=<factory>, metadata=<factory>, template_name='', control_action='', control_data=None)[source]

Bases: BaseAction

Action to control template elements via WebSocket

control_action: str = ''

control_data: Dict[str, Any] = None

async execute(trigger_data, event_data)[source]

Execute template control action

Return type:: bool

template_name: str = ''

validate_parameters()[source]

Validate template control action parameters

Return type:: bool

class modules.connector_actions.TriggerAlertAction(action_id, action_type, name, description='', enabled=True, parameters=<factory>, metadata=<factory>, alert_type='', alert_id='', alert_data=None)[source]

Bases: BaseAction

Action to trigger an alert through the alert system

alert_data: Dict[str, Any] = None

alert_id: str = ''

alert_type: str = ''

async execute(trigger_data, event_data)[source]

Execute trigger alert action

Return type:: bool

validate_parameters()[source]

Validate trigger alert action parameters

Return type:: bool

class modules.connector_actions.UpdateGreetingAction(action_id, action_type, name, description='', enabled=None, parameters=<factory>, metadata=<factory>, greeting_id='', greeting_text='')[source]

Bases: BaseAction

Action to update an existing greeting

enabled: bool = None

async execute(trigger_data, event_data)[source]

Execute update greeting action

Return type:: bool

greeting_id: str = ''

greeting_text: str = ''

validate_parameters()[source]

Validate update greeting action parameters

Return type:: bool

class modules.connector_actions.WebSocketEmitAction(action_id, action_type, name, description='', enabled=True, parameters=<factory>, metadata=<factory>, event_name='', event_data=None)[source]

Bases: BaseAction

Action to emit a custom WebSocket event

event_data: Dict[str, Any] = None

event_name: str = ''

async execute(trigger_data, event_data)[source]

Execute WebSocket emit action

Return type:: bool

validate_parameters()[source]

Validate WebSocket emit action parameters

Return type:: bool

class modules.connector_actions.WriteFileAction(action_id, action_type, name, description='', enabled=True, parameters=<factory>, metadata=<factory>, file_path='', content='', append=False)[source]

Bases: BaseAction

Action to write data to a file

append: bool = False

content: str = ''

async execute(trigger_data, event_data)[source]

Execute write file action

Return type:: bool

file_path: str = ''

validate_parameters()[source]

Validate write file action parameters

Return type:: bool

modules.connector_actions.create_action(action_type, action_id, name, **kwargs)[source]

Factory function to create action instances

Return type:: BaseAction

Service Integration

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.connector_integration.ConnectorIntegration[source]

Bases: object

Integration layer between the connector system and existing Mycelian modules

get_available_template_actions()[source]

Get available template actions from template configs

Returns:: Template name -> {action_id: action_config} mapping
Return type:: Dict[str, Dict[str, Any]]

async send_hotkey_event(key_code, modifiers=None, is_global=True)[source]: Send a hotkey event to the connector system

async send_timer_event(timer_id, interval_seconds)[source]: Send a timer event to the connector system

async send_webhook_event(webhook_data)[source]: Send a webhook event to the connector system

setup_streamlabs_integration()[source]: Set up integration with StreamLabs events

setup_twitch_integration()[source]: Set up integration with Twitch events

modules.connector_integration.get_integration()[source]

Get the global connector integration instance

Return type:: ConnectorIntegration

modules.connector_integration.initialize_integration()[source]: Initialize the connector integration system

Connector Examples

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

modules.connector_examples.create_chat_command_example()[source]

Create an example connector that responds to a chat command

Return type:: Connector

modules.connector_examples.create_donation_log_example()[source]

Create an example connector that logs donations to a file

Return type:: Connector

modules.connector_examples.create_example_connectors()[source]: Create example connectors to demonstrate the system

modules.connector_examples.create_high_bits_counter_example()[source]

Create an example connector that increments counter for high bit amounts

Return type:: Connector

modules.connector_examples.create_new_follower_wheel_example()[source]

Create an example connector that spins roulette for new followers

Return type:: Connector

modules.connector_examples.create_raid_celebration_example()[source]

Create an example connector for raid celebrations

Return type:: Connector

Hotkey System

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.hotkey_listener.HotkeyListener[source]

Bases: object

Global hotkey listener for connector system

get_registered_hotkeys()[source]

Get all registered hotkeys

Return type:: Dict[str, List[str]]

register_hotkey(connector_id, key_combination, is_global=True)[source]: Register a hotkey for a connector

start()[source]: Start the hotkey listener

stop()[source]: Stop the hotkey listener

unregister_hotkey(connector_id, key_combination=None)[source]: Unregister a hotkey for a connector

modules.hotkey_listener.cleanup()[source]: Cleanup the hotkey listener system

modules.hotkey_listener.get_listener()[source]

Get the global hotkey listener instance

Return type:: HotkeyListener

modules.hotkey_listener.initialize()[source]: Initialize the hotkey listener system

Utility Modules

Data Objects

Path Utilities

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

modules.path_utils.ensure_directory_exists(path)[source]

Ensure that a directory exists, creating it if necessary.

Parameters:: path (str) – Path to the directory
Returns:: True if directory exists or was created successfully
Return type:: bool

modules.path_utils.get_assets_path(relative_path='')[source]

Get absolute path to assets directory/files.

Parameters:: relative_path (str) – Relative path within assets directory
Returns:: Absolute path to the assets file/directory
Return type:: str

modules.path_utils.get_data_path(relative_path)[source]

Get absolute path to data files, works for both development and PyInstaller. Data files are those that should persist and be writable (configs, databases, logs).

Parameters:: relative_path (str) – Relative path to the data file
Returns:: Absolute path to the data file
Return type:: str

modules.path_utils.get_executable_dir()[source]

Get the directory containing the executable or script.

Returns:: Directory path
Return type:: str

modules.path_utils.get_resource_path(relative_path)[source]

Get absolute path to resource, works for both development and PyInstaller.

Parameters:: relative_path (str) – Relative path to the resource file
Returns:: Absolute path to the resource
Return type:: str

modules.path_utils.get_static_path(relative_path='')[source]

Get absolute path to static directory/files.

Parameters:: relative_path (str) – Relative path within static directory
Returns:: Absolute path to the static file/directory
Return type:: str

modules.path_utils.get_template_path(relative_path='')[source]

Get absolute path to template directory/files.

Parameters:: relative_path (str) – Relative path within templates directory
Returns:: Absolute path to the template file/directory
Return type:: str

modules.path_utils.get_working_directory()[source]

Get the appropriate working directory for the application.

Returns:: Working directory path
Return type:: str

modules.path_utils.is_frozen()[source]

Check if the application is running as a frozen executable.

Returns:: True if running as executable, False if running as script
Return type:: bool

Encryption Utilities

API Credentials Manager

Template Configuration

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.template_config_parser.TemplateConfigParser(config_dir='templates/template_configs')[source]

Bases: object

Parser for template configuration files. Handles reading, editing, and saving JSON configuration files.

create_config(config_name, default_config=None)[source]

Create a new configuration file

Parameters:

config_name (str) – Name of the config (without extension)
default_config (Optional[Dict[str, Any]]) – Default configuration structure

Returns:

True if successful, False otherwise

Return type:

bool

delete_config(config_name)[source]

Delete a configuration file

Parameters:: config_name (str) – Name of the config (without extension)
Returns:: True if successful, False otherwise
Return type:: bool

get_all_configs()[source]

Get all configurations

Returns:: Dictionary of config names to their configurations
Return type:: Dict[str, Dict[str, Any]]

get_config_files()[source]

Get a list of all configuration files

Returns:: List of config filenames without extension
Return type:: List[str]

get_config_path(config_name)[source]

Get the path to a configuration file

Parameters:: config_name (str) – Name of the config (without extension)
Returns:: Path to the configuration file
Return type:: str

get_non_hidden_config_files()[source]

Get a list of all non-hidden configuration files

Returns:: List of non-hidden config filenames without extension
Return type:: List[str]

get_streamdeck_configs()[source]

Get all template configurations that have Stream Deck integration

Returns:: Dictionary of template names to their Stream Deck configurations
Return type:: Dict[str, Dict[str, Any]]

is_config_hidden(config_name)[source]

Check if a configuration is marked as hidden

Parameters:: config_name (str) – Name of the config (without extension)
Returns:: True if the config is hidden, False otherwise
Return type:: bool

load_config(config_name, include_dynamic_controls=False, include_streamdeck_options=False)[source]

Load configuration from a file

Parameters:

config_name (str) – Name of the config (without extension)
include_dynamic_controls (bool) – Whether to include dynamic_controls section
include_streamdeck_options (bool) – Whether to include streamdeck_options section

Returns:

Configuration data

Return type:

Dict[str, Any]

save_config(config_name, config)[source]

Save configuration to a file

Parameters:

config_name (str) – Name of the config (without extension)
config (Dict[str, Any]) – Configuration data to save

Returns:

True if successful, False otherwise

Return type:

bool

modules.template_config_parser.resolve_dynamic_control_values_from_elements(config)[source]

Return a deep copy of dynamic_controls with display value taken from the main elements entry identified by persist.target_element_id when present, matching what persist_template_control_change writes. Then, when persist includes runtime_database_path and runtime_state_key, overlay value from that database document (see _overlay_source_control_runtime_state). Otherwise each control keeps its JSON defaults.

Return type:: Dict[str, Any]

Preview Mappings

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

modules.preview_mappings.get_builtin_variable_mappings()[source]

Get the comprehensive mapping of built-in variables to example values.

This mapping is used by the chatbot preview functionality to replace variables in command/event responses with realistic example values.

Returns:: Dictionary mapping variable names to example values
Return type:: Dict[str, str]

modules.preview_mappings.get_category_variables(category)[source]

Get variables for a specific category.

Parameters:: category (str) – The category name (e.g., ‘basic’, ‘time_date’, ‘statistics’)
Returns:: Variables for the specified category
Return type:: Dict[str, str]

modules.preview_mappings.get_variable_categories()[source]

Get variable mappings organized by categories for easier maintenance.

Returns:: Variables organized by category
Return type:: Dict[str, Dict[str, str]]

Migration and Maintenance

Database Migration

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.alert_database_migration.MigrationResult(total_alerts=0, successful_migrations=0, failed_migrations=0, skipped_alerts=0, alert_type_counts=None, errors=None)[source]

Bases: object

Result of an alert migration operation

alert_type_counts: Dict[str, int] = None

errors: List[Dict[str, str]] = None

failed_migrations: int = 0

skipped_alerts: int = 0

successful_migrations: int = 0

total_alerts: int = 0

modules.alert_database_migration.migrate_alerts_from_file(file_path, overwrite_existing=False)[source]

Migrate alerts from an external JSON file.

Parameters:

file_path (str) – Path to the JSON file containing alerts
overwrite_existing (bool) – Whether to overwrite existing alerts with the same ID

Returns:

Detailed results of the migration

Return type:

MigrationResult

modules.alert_database_migration.preview_migration(file_path)[source]

Preview what would be migrated from a file without actually migrating.

Parameters:: file_path (str) – Path to the JSON file to preview
Returns:: (total_count, type_counts) where type_counts is a dict of alert_type: count
Return type:: tuple[int, Dict[str, int]]

MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class modules.alerts_migration.AlertMigrationResult[source]

Bases: object

Results of an alert migration operation

add_error(alert_type, alert_id, error)[source]: Record a migration error

add_skip(alert_type, alert_id, reason)[source]: Record a skipped alert

add_success(alert_type, alert_id, alert_name='')[source]: Record a successful migration

get_summary()[source]

Get a summary of the migration results

Return type:: str

class modules.alerts_migration.AlertsMigrator[source]

Bases: object

Migrates alerts from external JSON files to Mycelian’s alert system

migrate_from_file(file_path, overwrite_existing=False)[source]

Migrate alerts from a JSON file

Parameters:

file_path (str) – Path to the alerts JSON file
overwrite_existing (bool) – Whether to overwrite existing alerts

Returns:

Results of the migration

Return type:

AlertMigrationResult

modules.alerts_migration.migrate_alerts_from_file(file_path, overwrite_existing=False)[source]

Convenience function to migrate alerts from a file

Parameters:

file_path (str) – Path to the alerts JSON file
overwrite_existing (bool) – Whether to overwrite existing alerts

Returns:

Results of the migration

Return type:

AlertMigrationResult

modules.alerts_migration.preview_migration(file_path)[source]

Preview what would be migrated from a file without actually migrating

Parameters:: file_path (str) – Path to the alerts JSON file
Returns:: Total count and breakdown by alert type
Return type:: Tuple[int, dict]

Updater

Build System

Improved PyInstaller build script for Mycelian Balances size optimization with proper NiceGUI and web engine support

class build.BuildProgress(total_steps=5)[source]

Bases: object

Simple progress tracking for the build process

error(message)[source]: Display error message

next_step(message)[source]: Move to next step and display progress

success(message='✓ Complete')[source]: Mark current step as successful

summary()[source]: Display build summary

update(message)[source]: Update current step message

build.build_executable()[source]: Build the executable using PyInstaller

build.clean_build()[source]: Clean previous build artifacts

build.create_build_info(dist_dir)[source]: Create build information file

build.create_runtime_hook()[source]: Create a runtime hook to help with DLL loading and encoding

build.create_spec_file()[source]: Create custom spec file for PyInstaller

build.detect_os()[source]: Detect the current operating system

build.ensure_dependencies()[source]: Ensure all required packages are installed

build.get_data_files()[source]: Get list of data files to include

build.get_excluded_modules()[source]: Get list of modules to exclude from build (conservative exclusions only)

build.get_hidden_imports(current_os)[source]: Get list of hidden imports that PyInstaller might miss, filtered by OS

build.get_os_specific_icon_path(os_name)[source]: Get the appropriate icon path for the current OS

build.get_project_root()[source]: Get the project root directory

build.main()[source]: Main build function

build.post_build_tasks(dist_dir)[source]: Perform post-build tasks

build.update_version_across_files()[source]: Update version and build date across all relevant project files