API Documentation - edgeof8/tIRC GitHub Wiki
API Overview
tIRC provides a comprehensive API for extending and customizing the IRC client. This reference documents the public interfaces available to developers.
Note: All API methods that perform I/O are asynchronous and should be awaited.
Core API
The Core API provides access to the main application functionality and core components.
IRCClient_Logic
The main application class that orchestrates all components of the IRC client.
Note: All methods are asynchronous and should be awaited.
Properties
class IRCClient_Logic:
"""Main application class for tIRC IRC client.
This class serves as the central coordinator for all IRC client functionality,
including network communication, user interface, and script management.
"""
@property
def is_headless(self) -> bool:
"""Check if the client is running in headless mode (without UI)."""
pass
@property
def config(self) -> AppConfig:
"""Get the application configuration."""
pass
@property
def state_manager(self) -> StateManager:
"""Get the state manager instance."""
pass
@property
def event_manager(self) -> EventManager:
"""Get the event manager instance."""
pass
@property
def script_manager(self) -> ScriptManager:
"""Get the script manager instance."""
pass
Methods
async def run_main_loop(self) -> None:
"""
Main asyncio loop to handle user input and update the UI.
This method initializes the client, starts necessary background tasks,
and runs the main event loop until shutdown is requested.
"""
pass
async def connect(
self,
server: str,
port: int,
use_ssl: bool = True,
nick: Optional[str] = None,
username: Optional[str] = None,
realname: Optional[str] = None
) -> None:
"""
Connect to an IRC server.
Args:
server: Server hostname or IP address
port: Server port number
use_ssl: Whether to use SSL/TLS for the connection
nick: Nickname to use (uses config if None)
username: Username to use (uses nick if None)
realname: Real name to use (uses nick if None)
"""
pass
async def disconnect(self, quit_message: str = "Client quitting") -> None:
"""
Disconnect from the current IRC server.
Args:
quit_message: Quit message to send to the server
"""
pass
async def join_channel(self, channel: str, key: Optional[str] = None) -> None:
"""
Join an IRC channel.
Args:
channel: Channel name (including # or & prefix)
key: Optional channel key/password
"""
pass
async def part_channel(self, channel: str, reason: str = "") -> None:
"""
Leave an IRC channel.
Args:
channel: Channel name to leave
reason: Optional part message
"""
pass
async def send_message(self, target: str, message: str) -> None:
"""
Send a PRIVMSG to a channel or user.
Args:
target: Channel or nickname to send to
message: Message text to send
"""
pass
async def send_notice(self, target: str, message: str) -> None:
"""
Send a NOTICE to a channel or user.
Args:
target: Channel or nickname to send to
message: Notice text to send
"""
pass
async def set_nick(self, new_nick: str) -> bool:
"""
Change the client's nickname.
Args:
new_nick: New nickname to use
Returns:
bool: True if the nick change was initiated, False otherwise
"""
pass
def get_channel_users(self, channel: str) -> List[Dict[str, Any]]:
"""
Get a list of users in a channel.
Args:
channel: Channel name
Returns:
List of user dictionaries with nick, modes, etc.
"""
pass
def get_channel_modes(self, channel: str) -> Dict[str, Any]:
"""
Get the current modes for a channel.
Args:
channel: Channel name
Returns:
Dictionary of channel modes and their parameters
"""
pass
Event API
The Event API provides a powerful system for handling IRC events and client state changes. It allows scripts and components to subscribe to specific events and react accordingly.
EventManager
Central class for managing event dispatching and handling throughout the application.
class EventManager:
"""Manages event dispatching and handling for the IRC client.
This class provides methods to dispatch events and allows components to subscribe to
specific events they're interested in. All event handlers are asynchronous.
"""
def __init__(self, client_logic_ref: "IRCClient_Logic", script_manager_ref: "ScriptManager"):
"""Initialize the EventManager with references to client logic and script manager."""
self.client_logic = client_logic_ref
self.script_manager = script_manager_ref
def _prepare_base_event_data(self, raw_line: str = "") -> Dict[str, Any]:
"""
Prepare a base dictionary with common event data.
Args:
raw_line: Raw IRC line that triggered the event (if applicable)
Returns:
Dictionary containing base event data including timestamp and client info
"""
pass
async def dispatch_event(self, event_name: str, specific_event_data: Dict[str, Any], raw_line: str = "") -> None:
"""
Dispatch an event to all registered handlers.
Args:
event_name: Name of the event to dispatch (e.g., "PRIVMSG", "JOIN")
specific_event_data: Dictionary containing event-specific data
raw_line: Optional raw IRC line that triggered the event
"""
pass
# Client Lifecycle Events
async def dispatch_client_connected(self, server: str, port: int, nick: str, ssl: bool, raw_line: str = "") -> None:
"""Dispatch when the client connects to an IRC server."""
pass
async def dispatch_client_disconnected(self, server: str, port: int, raw_line: str = "") -> None:
"""Dispatch when the client disconnects from an IRC server."""
pass
async def dispatch_client_registered(self, nick: str, server_message: str, raw_line: str = "") -> None:
"""Dispatch when the client successfully registers with the IRC server."""
pass
async def dispatch_client_ready(self, nick: str, raw_line: str = "") -> None:
"""Dispatch when the client is fully initialized and ready."""
pass
# IRC Message Events
async def dispatch_privmsg(
self,
nick: str,
userhost: str,
target: str,
message: str,
is_channel_msg: bool,
tags: Dict[str, Any],
raw_line: str = ""
) -> None:
"""Dispatch when a PRIVMSG (private message) is received."""
pass
async def dispatch_notice(
self,
nick: Optional[str],
userhost: Optional[str],
target: str,
message: str,
is_channel_notice: bool,
tags: Dict[str, Any],
raw_line: str = ""
) -> None:
"""Dispatch when a NOTICE is received."""
pass
# Channel Events
async def dispatch_join(
self,
nick: str,
userhost: Optional[str],
channel: str,
account: Optional[str],
real_name: Optional[str],
is_self: bool,
raw_line: str = ""
) -> None:
"""Dispatch when a user (or self) joins a channel."""
pass
async def dispatch_channel_fully_joined(self, channel_name: str, raw_line: str = "") -> None:
"""Dispatch when the client has fully joined a channel (after receiving user list)."""
pass
async def dispatch_part(
self,
nick: str,
userhost: str,
channel: str,
reason: str,
is_self: bool,
raw_line: str = ""
) -> None:
"""Dispatch when a user (or self) parts a channel."""
pass
# User Events
async def dispatch_quit(
self,
nick: str,
userhost: str,
reason: str,
raw_line: str = ""
) -> None:
"""Dispatch when a user quits the IRC network."""
pass
async def dispatch_nick(
self,
old_nick: str,
new_nick: str,
userhost: str,
is_self: bool,
raw_line: str = ""
) -> None:
"""Dispatch when a user (or self) changes their nickname."""
pass
# Channel Mode Events
async def dispatch_mode(
self,
target_name: str,
setter_nick: Optional[str],
setter_userhost: Optional[str],
mode_string: str,
mode_params: List[str],
parsed_modes: List[Dict[str, Any]],
raw_line: str = ""
) -> None:
"""Dispatch when channel or user modes are changed."""
pass
async def dispatch_channel_mode_applied(
self,
channel: str,
setter_nick: Optional[str],
setter_userhost: Optional[str],
mode_changes: List[Dict[str, Any]],
current_channel_modes: List[str],
raw_line: str = ""
) -> None:
"""Dispatch when channel modes are applied to a specific channel."""
pass
# DCC (Direct Client-to-Client) Events
async def dispatch_dcc_transfer_status_change(
self,
transfer: 'DCCTransfer',
raw_line: str = ""
) -> None:
"""Dispatch when a DCC transfer's status changes."""
pass
Common Event Types
Here are some of the most commonly used events in the tIRC client:
Event Name | Description | Data Structure |
---|---|---|
CLIENT_CONNECTED |
Dispatched when connecting to an IRC server | {"server": str, "port": int, "nick": str, "ssl": bool} |
CLIENT_DISCONNECTED |
Dispatched when disconnecting from an IRC server | {"server": str, "port": int} |
CLIENT_READY |
Dispatched when client is fully initialized | {"nick": str, "client_logic_ref": IRCClient_Logic} |
PRIVMSG |
Received a private message (channel or query) | {"nick": str, "userhost": str, "target": str, "message": str, "is_channel_msg": bool, "tags": dict} |
NOTICE |
Received a notice (channel or user) | {"nick": str, "userhost": str, "target": str, "message": str, "is_channel_notice": bool, "tags": dict} |
JOIN |
A user joined a channel | {"nick": str, "userhost": str, "channel": str, "account": str, "real_name": str, "is_self": bool} |
PART |
A user left a channel | {"nick": str, "userhost": str, "channel": str, "reason": str, "is_self": bool} |
QUIT |
A user disconnected from the network | {"nick": str, "userhost": str, "reason": str} |
NICK |
A user changed their nickname | {"old_nick": str, "new_nick": str, "userhost": str, "is_self": bool} |
MODE |
Channel or user modes were changed | {"target": str, "setter": str, "setter_userhost": str, "mode_string": str, "mode_params": List[str], "parsed_modes": List[dict]} |
DCC_TRANSFER_STATUS_CHANGE |
Status of a DCC transfer changed | {"transfer_id": str, "transfer_type": str, "peer_nick": str, "filename": str, "status": str, ...} |
Example: Subscribing to Events
Here's how you can subscribe to events in your script:
from typing import Dict, Any
import asyncio
class MyScript:
def __init__(self, client):
self.client = client
self.script_name = "MyScript"
async def on_privmsg(self, event_data: Dict[str, Any]) -> None:
"""Handle PRIVMSG events."""
if event_data["is_channel_msg"]:
print(f"Message in {event_data['target']} from {event_data['nick']}: {event_data['message']}")
else:
print(f"Private message from {event_data['nick']}: {event_data['message']}")
async def on_join(self, event_data: Dict[str, Any]) -> None:
"""Handle JOIN events."""
if event_data["is_self"]:
print(f"I joined channel: {event_data['channel']}")
else:
print(f"{event_data['nick']} joined {event_data['channel']}")
def register(self):
"""Register event handlers."""
self.client.event_manager.subscribe("PRIVMSG", self.on_privmsg)
self.client.event_manager.subscribe("JOIN", self.on_join)
# In your script's initialization:
# script = MyScript(client)
# script.register()
State API
The State API provides a robust system for managing application state with persistence, validation, and change notification. It's designed to be thread-safe and supports automatic persistence to disk.
StateManager
Central class for managing application state with persistence and validation.
class StateManager:
"""Manages application state with persistence and validation.
The StateManager provides a thread-safe way to store, retrieve, and observe changes
to application state. It supports automatic persistence to disk and validation of state values.
"""
def __init__(
self,
state_file: str = "state.json",
auto_save: bool = True,
save_interval: int = 60, # seconds
validate_on_change: bool = True,
):
"""Initialize the StateManager.
Args:
state_file: Path to the file where state will be persisted
auto_save: If True, automatically save state to disk on changes
save_interval: How often to auto-save (in seconds) if auto_save is True
validate_on_change: If True, validate state values when they change
"""
pass
# Core State Management
def get(self, key: str, default: Any = None) -> Any:
"""
Get a state value.
Args:
key: The state key to retrieve
default: Default value if key doesn't exist
Returns:
The state value or default if not found
"""
pass
def set(self, key: str, value: Any, metadata: Optional[Dict[str, Any]] = None) -> bool:
"""
Set a state value with optional validation.
Args:
key: The state key to set
value: The value to set
metadata: Optional metadata about the change
Returns:
bool: True if the state was set successfully, False if validation failed
Raises:
ValueError: If validation is enabled and the value is invalid
"""
pass
def delete(self, key: str) -> bool:
"""
Delete a state value.
Args:
key: The state key to delete
Returns:
bool: True if the key existed and was deleted, False otherwise
"""
pass
def clear(self) -> None:
"""Clear all state."""
pass
def get_all(self) -> Dict[str, Any]:
"""
Get a copy of all state values.
Returns:
Dict containing all state key-value pairs
"""
pass
# Connection State Management
def set_connection_state(
self,
state: "ConnectionState",
error: Optional[str] = None,
config_errors: Optional[List[str]] = None,
) -> None:
"""
Update the connection state and optionally set an error message.
Args:
state: The new connection state
error: Optional error message if the state is ERROR
config_errors: List of configuration errors if any
"""
pass
def get_connection_state(self) -> "ConnectionState":
"""
Get the current connection state.
Returns:
The current ConnectionState enum value
"""
pass
def get_connection_info(self) -> Optional["ConnectionInfo"]:
"""
Get the current connection information.
Returns:
ConnectionInfo object or None if not connected
"""
pass
def update_connection_attempt(
self,
success: bool,
error: Optional[str] = None,
config_errors: Optional[List[str]] = None,
) -> None:
"""
Update connection attempt statistics.
Args:
success: Whether the connection attempt was successful
error: Optional error message if the attempt failed
config_errors: List of configuration errors if any
"""
pass
# Validation
def register_validator(self, key: str, validator: "StateValidator") -> None:
"""
Register a validator for a state key.
Args:
key: The state key to validate
validator: A StateValidator instance to use for validation
"""
pass
def unregister_validator(self, key: str) -> None:
"""
Unregister a validator for a state key.
Args:
key: The state key to remove validation for
"""
pass
def validate_all(self) -> bool:
"""
Validate all state values with registered validators.
Returns:
bool: True if all validations passed, False otherwise
"""
pass
# Change Handlers
def register_change_handler(
self,
key: str,
handler: Callable[["StateChange"], Any]
) -> None:
"""
Register a handler for state changes on a specific key.
Args:
key: The state key to monitor
handler: Callback function that takes a StateChange parameter
"""
pass
def register_global_handler(
self,
handler: Callable[["StateChange"], Any]
) -> None:
"""
Register a handler for all state changes.
Args:
handler: Callback function that takes a StateChange parameter
"""
pass
def unregister_change_handler(
self,
key: str,
handler: Callable[["StateChange"], Any]
) -> None:
"""
Unregister a handler for state changes on a specific key.
Args:
key: The state key being monitored
handler: The handler function to remove
"""
pass
def unregister_global_handler(
self,
handler: Callable[["StateChange"], Any]
) -> None:
"""
Unregister a global state change handler.
Args:
handler: The handler function to remove
"""
pass
StateChange
Represents a state change event, dispatched by the StateManager.
@dataclass
class StateChange(Generic[T]):
"""Represents a state change event.
Attributes:
key: The key of the state that changed
old_value: The value before the change (None for new keys)
new_value: The new value (None for deleted keys)
change_type: A StateChangeType enum value
timestamp: When the change occurred
metadata: Additional context about the change
"""
key: str
old_value: Optional[T]
new_value: Optional[T]
change_type: "StateChangeType"
timestamp: datetime = field(default_factory=datetime.now)
metadata: Dict[str, Any] = field(default_factory=dict)
class StateChangeType(Enum):
"""Types of state changes that can occur."""
CREATED = auto() # A new key was created
UPDATED = auto() # An existing key's value was updated
DELETED = auto() # A key was deleted
VALIDATED = auto() # A key's value was validated
INVALIDATED = auto() # A key's value failed validation
ConnectionState
Represents the possible connection states of the IRC client.
class ConnectionState(Enum):
"""Possible connection states for the IRC client."""
DISCONNECTED = auto() # Not connected to any server
CONNECTING = auto() # Attempting to connect to a server
CONNECTED = auto() # TCP connection established, but not registered yet
REGISTERED = auto() # Successfully registered with the IRC server
READY = auto() # Fully connected and ready for use
ERROR = auto() # An error occurred during connection
CONFIG_ERROR = auto() # Configuration error preventing connection
Example: Using StateManager
Here's how you can use the StateManager in your code:
from tirc_core.state_manager import StateManager, StateChange
from enum import Enum
from typing import Dict, Any, Optional
import asyncio
# Define custom state keys
class AppState:
USER_PREFERENCES = "user_prefs"
CONNECTION = "connection"
LAST_UPDATED = "last_updated"
# Initialize state manager
state_manager = StateManager("my_app_state.json", auto_save=True)
# Set some initial state
state_manager.set(AppState.USER_PREFERENCES, {
"theme": "dark",
"notifications": True,
"font_size": 14
})
# Register a handler for specific state changes
def on_theme_changed(change: StateChange) -> None:
if change.key == AppState.USER_PREFERENCES and \
change.old_value and change.new_value and \
change.old_value.get("theme") != change.new_value.get("theme"):
print(f"Theme changed to: {change.new_value['theme']}")
state_manager.register_change_handler(AppState.USER_PREFERENCES, on_theme_changed)
# Update theme - this will trigger our handler
state_manager.set(AppState.USER_PREFERENCES, {
**state_manager.get(AppState.USER_PREFERENCES, {}),
"theme": "light"
})
# Monitor connection state changes
async def on_connection_change(change: StateChange) -> None:
if change.key == "connection_state":
print(f"Connection state changed to: {change.new_value}")
if change.new_value == ConnectionState.ERROR:
error = state_manager.get("last_error")
if error:
print(f"Error: {error}")
state_manager.register_change_handler("connection_state", on_connection_change)
# Simulate a connection state change
state_manager.set_connection_state(ConnectionState.CONNECTING)
# Later...
state_manager.set_connection_state(ConnectionState.REGISTERED)
# Clean up when done
state_manager.shutdown()
State Validation
You can create custom validators to ensure state values meet specific criteria:
from tirc_core.state_manager import StateValidator
from typing import Any, Dict
class UserPrefsValidator(StateValidator[Dict[str, Any]]):
"""Validates user preferences."""
def validate(self, value: Dict[str, Any]) -> bool:
if not isinstance(value, dict):
return False
# Validate theme
if "theme" in value and value["theme"] not in ["light", "dark", "system"]:
return False
# Validate font size
if "font_size" in value and not (8 <= value["font_size"] <= 24):
return False
return True
def get_error_message(self, value: Dict[str, Any]) -> Optional[str]:
if not isinstance(value, dict):
return "Preferences must be a dictionary"
if "theme" in value and value["theme"] not in ["light", "dark", "system"]:
return f"Invalid theme: {value['theme']}. Must be 'light', 'dark', or 'system'"
if "font_size" in value and not (8 <= value["font_size"] <= 24):
return f"Font size {value['font_size']} is out of range (8-24)"
return None
# Register the validator
state_manager.register_validator(AppState.USER_PREFERENCES, UserPrefsValidator())
# This will fail validation
state_manager.set(AppState.USER_PREFERENCES, {"theme": "ocean", "font_size": 6})
# Error: Invalid theme: ocean. Must be 'light', 'dark', or 'system'
Example: Subscribing to Events
Here's how you can subscribe to events in your script:
from typing import Dict, Any
import asyncio
class MyScript:
def __init__(self, client):
self.client = client
self.script_name = "MyScript"
async def on_privmsg(self, event_data: Dict[str, Any]) -> None:
"""Handle PRIVMSG events."""
if event_data["is_channel_msg"]:
print(f"Message in {event_data['target']} from {event_data['nick']}: {event_data['message']}")
else:
print(f"Private message from {event_data['nick']}: {event_data['message']}")
async def on_join(self, event_data: Dict[str, Any]) -> None:
"""Handle JOIN events."""
if event_data["is_self"]:
print(f"I joined channel: {event_data['channel']}")
else:
print(f"{event_data['nick']} joined {event_data['channel']}")
def register(self):
"""Register event handlers."""
self.client.event_manager.subscribe("PRIVMSG", self.on_privmsg)
self.client.event_manager.subscribe("JOIN", self.on_join)
# In your script's initialization:
# script = MyScript(client)
# script.register()
Network API
The Network API provides the core networking functionality for the tIRC IRC client, handling connections, message sending/receiving, and network state management.
NetworkHandler
Manages network connections, message transmission, and connection state for the IRC client.
class NetworkHandler:
"""Handles all network operations for the IRC client including connection management,
message sending/receiving, and network state tracking.
"""
def __init__(self, client_logic_ref: "IRCClient_Logic"):
"""Initialize the NetworkHandler with a reference to the client logic.
Args:
client_logic_ref: Reference to the main IRC client logic instance
"""
self.client_logic_ref = client_logic_ref
self.connected = False
self.buffer = b""
# ... other initialization ...
async def start(self) -> bool:
"""Start the network handler and its asyncio task.
Returns:
bool: True if the network handler started successfully, False otherwise
"""
pass
async def stop(self) -> bool:
"""Stop the network handler and wait for task completion.
Returns:
bool: True if stopped successfully, False otherwise
"""
pass
async def disconnect_gracefully(self, quit_message: Optional[str] = None) -> None:
"""Disconnect from the server gracefully.
Args:
quit_message: Optional quit message to send to the server
"""
pass
async def update_connection_params(
self,
server: str,
port: int,
use_ssl: bool,
channels_to_join: Optional[List[str]] = None
) -> None:
"""Update connection parameters and reconnect if needed.
Args:
server: Server hostname or IP address
port: Server port number
use_ssl: Whether to use SSL/TLS for the connection
channels_to_join: Optional list of channels to join after connecting
"""
pass
async def send_raw(self, data: str) -> None:
"""Send raw data to the IRC server.
Args:
data: Raw IRC protocol data to send
Raises:
ConnectionError: If not connected or connection is closing
"""
pass
async def _process_received_data(self, data: bytes) -> None:
"""Process received data from the network socket.
Args:
data: Raw bytes received from the socket
"""
pass
Key Methods
Method | Description |
---|---|
start() |
Initialize and start the network handler's background task |
stop() |
Stop the network handler and clean up resources |
disconnect_gracefully() |
Disconnect from the server with an optional quit message |
update_connection_params() |
Update server connection parameters and reconnect |
send_raw() |
Send raw IRC protocol data to the server |
send_cap_ls() |
Send CAP LS command to the server |
send_cap_req() |
Send CAP REQ command with specified capabilities |
send_authenticate() |
Send AUTHENTICATE command for SASL authentication |
Usage Example
# Example of using the Network API
# Initialize and start the network handler
network_handler = NetworkHandler(client_logic_ref)
await network_handler.start()
# Update connection parameters and connect
await network_handler.update_connection_params(
server="irc.example.com",
port=6697,
use_ssl=True,
channels_to_join=["#test", "#python"]
)
# Send raw IRC commands
await network_handler.send_raw("PRIVMSG #test :Hello, world!")
# Handle disconnection
try:
# ... do work ...
finally:
# Clean up
await network_handler.disconnect_gracefully("Goodbye!")
await network_handler.stop()
Network Connection States
The network handler maintains the following connection states:
- Disconnected: Not connected to any server
- Connecting: Attempting to establish a connection
- Connected: Successfully connected and authenticated
- Disconnecting: Gracefully closing the connection State changes are automatically handled and can be monitored through the State API.
UI API
The UI API provides comprehensive control over the user interface, including message display, window management, and user interaction.
UIManager
Central class for managing all UI components and user interactions.
class UIManager:
"""Manages all UI components including message display, input handling, and window layouts.
The UIManager handles the curses-based interface, including:
- Message display in channels/queries
- User input handling
- Window layout and rendering
- Status updates
- Context switching
"""
def __init__(self, stdscr: 'curses.window', client_ref: 'IRCClient_Logic'):
"""Initialize the UI manager.
Args:
stdscr: The main curses window
client_ref: Reference to the main client logic
"""
self.stdscr = stdscr
self.client = client_ref
self.colors = {}
self.split_mode_active = False
self.active_split_pane = "top"
self.top_pane_context_name = ""
self.bottom_pane_context_name = ""
self.msg_win_width = 80
self.msg_win_height = 24
async def add_message_to_context(
self,
text: str,
color_attr: int,
prefix_time: bool,
context_name: str
) -> None:
"""Add a message to the specified context's message history.
Args:
text: The message text to display
color_attr: Color attribute for the message
prefix_time: Whether to prefix the message with a timestamp
context_name: Name of the context (channel, query, etc.)
"""
pass
async def refresh_all_windows(self) -> None:
"""Refresh all windows in the UI."""
pass
async def scroll_messages(self, direction: str, lines: int = 1) -> None:
"""Scroll the message view in the active window.
Args:
direction: Scroll direction ('up', 'down', 'page_up', 'page_down', 'home', 'end')
lines: Number of lines to scroll (for 'up'/'down' directions)
"""
pass
async def scroll_user_list(self, direction: str, lines_arg: Optional[int] = None) -> None:
"""Scroll the user list in the sidebar.
Args:
direction: Scroll direction ('up', 'down', 'page_up', 'page_down', 'top', 'bottom')
lines_arg: Optional number of lines to scroll
"""
pass
async def switch_context(self, context_name: str) -> None:
"""Switch to a different context (channel, query, etc.).
Args:
context_name: Name of the context to switch to
"""
pass
def get_input_char(self) -> int:
"""Get a single character of input from the user.
Returns:
int: The character code, or curses.ERR if no input is available
"""
pass
def shutdown(self) -> None:
"""Clean up curses resources and shut down the UI."""
pass
Key Methods
Method | Description |
---|---|
add_message_to_context() |
Add a message to a specific context's message history |
refresh_all_windows() |
Refresh all UI windows and redraw content |
scroll_messages() |
Scroll the message view in the active window |
scroll_user_list() |
Scroll the user list in the sidebar |
switch_context() |
Switch between different chat contexts |
get_input_char() |
Get user input character by character |
shutdown() |
Clean up resources when shutting down |
UI Components
Message Window Displays messages for the current context with features like:
- Scrolling through message history
- Split view for multiple contexts
- Color-coded message types
- Timestamps and message formatting
Sidebar Shows available channels and users with:
- Scrollable user lists
- User status indicators
- Channel/user search
- Context switching
Usage Example
# Example of using the UI API
# Initialize the UI manager
ui = UIManager(stdscr, client_ref)
# Add a message to a channel
await ui.add_message_to_context(
text="Hello, world!",
color_attr=ui.colors["default"],
prefix_time=True,
context_name="#python"
)
# Switch to a different context
await ui.switch_context("#python")
# Scroll the message view
ui.scroll_messages("page_up")
# Handle user input
while True:
char = ui.get_input_char()
if char != curses.ERR:
# Process input
pass
# Clean up when done
ui.shutdown()
UI Architecture
The UI is built using the following components:
- Message Panel: Displays chat messages with formatting and colors
- Sidebar: Shows channel list and user information
- Status Bar: Displays connection status and notifications
- Input Line: Handles user input and commands The UI is designed to be responsive and adapt to different terminal sizes.
Command System API
The Command System API provides a flexible way to extend tIRC's functionality through custom commands. Commands can be added by creating Python modules in the appropriate command category directory.
Command Structure
Commands are defined in Python modules within the tirc_core/commands/
directory. Each command module should define a COMMAND_DEFINITIONS
list containing one or more command definitions.
Example Command Definition
# Example: help_command.py
COMMAND_DEFINITIONS = [
{
"name": "help",
"handler": "handle_help_command",
"help": {
"usage": "/help [command_name|category|script <script_name>]",
"description": "Displays general help, help for a specific command, category, or script.",
"aliases": ["h"],
},
}
]
async def handle_help_command(client: "IRCClient_Logic", args_str: str):
"""Handle the help command.
Args:
client: Reference to the IRC client instance
args_str: Arguments passed to the command
"""
# Command implementation here
pass
Command Definition Fields
Field | Type | Description | Required |
---|---|---|---|
name |
str | Primary name of the command (without the leading '/') | Yes |
handler |
str | Name of the function that handles this command | Yes |
help |
dict | Help information for the command (see below) | No |
Help Dictionary Fields
Field | Type | Description | Required |
---|---|---|---|
usage |
str | Example usage of the command | Yes |
description |
str | Detailed description of what the command does | Yes |
aliases |
List[str] | List of alternative command names | No |
Command Categories
Core Commands Essential commands for basic client operation.
/help
- Show help information/quit
- Disconnect and exit the client/connect
- Connect to an IRC server/disconnect
- Disconnect from the current server
Channel Commands Commands for managing IRC channels.
/join
- Join a channel/part
- Leave a channel/kick
- Remove a user from a channel/ban
- Ban a user from a channel/topic
- View or set the channel topic
User Commands Commands for interacting with other users.
/msg
- Send a private message/me
- Send an action message/notice
- Send a notice/whois
- Get user information
DCC Commands Commands for file transfers and chat.
/dcc send
- Send a file/dcc get
- Accept a file transfer/dcc chat
- Start a DCC chat session/dcc list
- List active DCC transfers
Creating Custom Commands
To create a new command:
- Create a new Python file in the appropriate category directory
- Define your command using the
COMMAND_DEFINITIONS
list - Implement the handler function with the signature:
async def handle_command(client: "IRCClient_Logic", args_str: str)
- Restart tIRC or use
/rehash
to load the new command
Command Handler Tips
- Commands are executed asynchronously - use
async/await
for any I/O operations- Use
client.add_message()
to send output to the user- Access the current context with
client.context_manager.active_context
- Check
client.state_manager.state
for connection state
Scripting API
Extend tIRC's functionality with custom Python scripts that can add commands, handle events, and interact with the IRC client's state.
Note: Scripts must be placed in the
scripts/
directory and must include acreate_script(api)
function that returns an instance of your script class.
Script Structure
Here's the basic structure of a tIRC script:
from tirc_core.scripting.script_base import ScriptBase
from typing import Dict, Any
import json
import os
class MyScript(ScriptBase):
"""My custom script documentation."""
def __init__(self, api):
super().__init__(api)
self.my_data = []
def load(self):
"""Called when the script is loaded."""
# Register commands and event handlers
self.api.register_command(
name="mycommand",
handler=self.handle_my_command,
help={
"usage": "/mycommand [args]",
"description": "Description of what this command does",
"aliases": ["mc"]
}
)
# Subscribe to events
self.api.subscribe_to_event("on_privmsg", self.handle_private_message)
# Load data
data_file = os.path.join(self.get_script_data_dir(), "data.json")
if os.path.exists(data_file):
with open(data_file, 'r') as f:
self.my_data = json.load(f)
def unload(self):
"""Called when the script is unloaded."""
# Save data and clean up
data_file = os.path.join(self.get_script_data_dir(), "data.json")
with open(data_file, 'w') as f:
json.dump(self.my_data, f)
async def handle_my_command(self, args_str: str) -> None:
"""Handle the /mycommand command."""
await self.api.send_message(
target=self.api.get_current_context_name() or "Status",
message=f"You ran mycommand with args: {args_str}"
)
async def handle_private_message(self, event_data: Dict[str, Any]) -> None:
"""Handle private messages."""
if "hello" in event_data["message"].lower():
await self.api.send_message(
target=event_data["nick"],
message=f"Hello, {event_data['nick']}!"
)
# Required function to create an instance of the script
def create_script(api):
return MyScript(api)
Script Metadata
Create a script_metadata.json
file in your script's directory to provide information about your script:
{
"name": "My Script",
"version": "1.0.0",
"description": "A useful script that does something cool",
"author": "Your Name",
"dependencies": ["requests"],
"min_tirc_version": "1.0.0",
"created_at": "2023-01-01T00:00:00",
"updated_at": "2023-01-01T00:00:00",
"tags": ["utility", "fun"],
"is_enabled": true
}
Core Components
ScriptBase Class Base class that all scripts must inherit from. Provides common functionality and lifecycle methods.
load()
- Called when the script is loadedunload()
- Called when the script is unloadedget_script_data_dir()
- Get directory for script dataget_script_dir()
- Get script's directory
ScriptAPIHandler
Provides access to tIRC's functionality. Available as self.api
in scripts.
register_command()
- Add new commandssubscribe_to_event()
- Listen for eventssend_message()
- Send messagesget_client_state()
- Access client state
API Reference
Method | Description |
---|---|
load() |
Called when the script is loaded. Register commands and event handlers here. |
unload() |
Called when the script is unloaded. Clean up resources here. |
get_script_data_dir() |
Returns the path to this script's data directory. |
Command Registration
Register new commands that users can execute in tIRC:
def load(self):
self.api.register_command(
name="greet",
handler=self.handle_greet,
help={
"usage": "/greet [name]",
"description": "Sends a greeting to the specified user or channel",
"aliases": ["hello", "hi"]
}
)
async def handle_greet(self, args_str: str) -> None:
args = args_str.split()
target = args[0] if args else self.api.get_current_context_name()
await self.api.send_message(
target=target,
message=f"Hello, {args[0] if args else 'everyone'}!"
)
Event Handling
Subscribe to and handle IRC events:
def load(self):
# Subscribe to message events
self.api.subscribe_to_event("on_privmsg", self.handle_message)
self.api.subscribe_to_event("on_join", self.handle_join)
async def handle_message(self, event_data: dict) -> None:
"""Handle incoming messages."""
if "hello" in event_data["message"].lower():
await self.api.send_message(
target=event_data["nick"],
message=f"Hello, {event_data['nick']}! How can I help you?"
)
async def handle_join(self, event_data: dict) -> None:
"""Handle when a user joins a channel."""
if event_data["nick"] != self.api.get_client_nick():
await self.api.send_message(
target=event_data["channel"],
message=f"Welcome to {event_data['channel']}, {event_data['nick']}!"
)
Data Persistence
Store and retrieve persistent data for your script:
def load(self):
# Load data when script starts
data_file = os.path.join(self.get_script_data_dir(), "data.json")
try:
if os.path.exists(data_file):
with open(data_file, 'r') as f:
self.data = json.load(f)
except Exception as e:
self.api.log_error(f"Error loading data: {e}")
self.data = {}
def unload(self):
# Save data when script unloads
data_file = os.path.join(self.get_script_data_dir(), "data.json")
try:
os.makedirs(os.path.dirname(data_file), exist_ok=True)
with open(data_file, 'w') as f:
json.dump(self.data, f)
except Exception as e:
self.api.log_error(f"Error saving data: {e}")
ScriptAPIHandler Reference
Method | Description |
---|---|
send_message(target, message) |
Send a message to a channel or user. |
send_action(target, action_text) |
Send a /me action message. |
join_channel(channel_name, key=None) |
Join a channel. |
part_channel(channel_name, reason=None) |
Leave a channel. |
register_command(name, handler, help=None, aliases=None) |
Register a new command. |
subscribe_to_event(event_name, handler) |
Subscribe to an event. |
get_current_context_name() |
Get the name of the current context. |
Available Script Events
Scripts can subscribe to a variety of events. The event_data
dictionary passed to handlers typically includes timestamp
, raw_line
(if applicable from a server message), and client_nick
. Below are key events mentioned in README.md
and their specific additional event_data
keys.
Client Lifecycle Events
CLIENT_CONNECTING
: Fired when a new connection attempt is initiated.event_data
keys:server
(str),port
(int),nick
(str),ssl
(bool).
CLIENT_CONNECTED
: Fired when TCP/IP connection is established and CAP negotiation begins.event_data
keys:server
(str),port
(int),nick
(str),ssl
(bool).
CLIENT_CAP_NEGOTIATION_START
: Fired when capability negotiation begins.event_data
keys:capabilities
(List[str] - requested capabilities).
CLIENT_CAP_NEGOTIATION_COMPLETE
: Fired when capability negotiation completes.event_data
keys:negotiated_capabilities
(List[str]).
CLIENT_AUTHENTICATING
: Fired when SASL authentication begins.event_data
keys:mechanism
(str).
CLIENT_AUTHENTICATED
: Fired upon successful SASL authentication.event_data
keys:account
(str - account name if available).
CLIENT_REGISTERING
: Fired when sending NICK/USER registration commands.event_data
keys:nick
(str),username
(str),realname
(str).
CLIENT_REGISTERED
: Fired upon receiving RPL_WELCOME (001).event_data
keys:nick
(str),server_message
(str),hostname
(str).
CLIENT_READY
: Fired after successful registration and initial auto-join actions.event_data
keys:nick
(str),channels
(List[str]).
CLIENT_DISCONNECTED
: Fired when connection is lost or closed.event_data
keys:server
(str),port
(int),reason
(str),reconnecting
(bool).
CLIENT_RECONNECTING
: Fired when a reconnection attempt is initiated.event_data
keys:attempt
(int),max_attempts
(int).
CLIENT_MESSAGE_ADDED_TO_CONTEXT
: Fired when any message is added to any context's buffer byIRCClient_Logic.add_message
.event_data
keys:context_name
(str),text
(str - the full line added, including timestamp/prefix),color_key_or_attr
(Any),source_full_ident
(Optional[str]),is_privmsg_or_notice
(bool).
CLIENT_NICK_CHANGED
: Fired specifically when tIRC's own nickname successfully changes.event_data
additional keys:old_nick
(str),new_nick
(str).
CLIENT_SHUTDOWN_FINAL
: Fired just before application exit, aftercurses
UI is down (if UI was active).
IRC Message & Command Events
PRIVMSG
: For channel and private messages.event_data
additional keys:nick
(str),userhost
(str),target
(str),message
(str),is_channel_msg
(bool),tags
(Dict[str, Any] - parsed IRCv3 message tags).
NOTICE
: For IRC NOTICEs.event_data
additional keys:nick
(str),userhost
(str),target
(str),message
(str),is_channel_notice
(bool),tags
(Dict[str, Any]).
JOIN
: When a user (including client) joins.event_data
additional keys:nick
(str),userhost
(str - if available from server, e.g. with extended-join),channel
(str),account
(str - if available, e.g. with extended-join),real_name
(str - if available, e.g. with extended-join),is_self
(bool).
CHANNEL_FULLY_JOINED
: Fired after a channel is successfully joined and RPL_ENDOFNAMES (or equivalent) is received.event_data
additional keys:channel_name
(str).
PART
: When a user (including client) parts.event_data
additional keys:nick
(str),userhost
(str),channel
(str),reason
(str),is_self
(bool).
QUIT
: When a user quits.event_data
additional keys:nick
(str),userhost
(str),reason
(str).
NICK
: When any user changes their nickname.event_data
additional keys:old_nick
(str),new_nick
(str),userhost
(str),is_self
(bool).
MODE
: When a mode change occurs (raw event).event_data
additional keys:nick
(str - setter),userhost
(str - setter's host),target
(str - channel/nick affected),modes_and_params
(str - raw mode string),parsed_modes
(List[Dict] - structured mode changes).
CHANNEL_MODE_APPLIED
: Fired after a channel MODE is processed and applied.event_data
additional keys:channel
(str),setter_nick
(str),setter_userhost
(str),mode_changes
(List[Dict] - structured),current_channel_modes
(List[str]).
TOPIC
: When a channel topic is changed or viewed.event_data
additional keys:nick
(str - setter),userhost
(str),channel
(str),topic
(str).
CHGHOST
: When a user's host changes.event_data
additional keys:nick
(str),new_ident
(str),new_host
(str),userhost
(str - old userhost).
Raw IRC Lines & Numerics
RAW_IRC_LINE
: Fired for every complete raw line received from the server before tIRC's internal parsing.event_data
keys:raw_line
(str).
RAW_IRC_NUMERIC
: Fired for all numeric replies from the server.event_data
keys:numeric
(int),source
(str - server name),params_list
(List[str] - full parameters),display_params_list
(List[str] - parameters with client nick removed if first),trailing
(Optional[str]),tags
(Dict[str, Any]).
Example Event Subscription:
async def my_privmsg_handler(event_data: Dict[str, Any]):
nick = event_data.get("nick")
message = event_data.get("message")
# ... do something ...
self.api.subscribe_to_event("PRIVMSG", my_privmsg_handler)
Note: This API reference is based on information available in README.md
and related documentation. The exact method signatures, return types, and available events might have evolved. Always refer to the latest tIRC source code (especially tirc_core/scripting/script_api_handler.py
) for the most accurate and complete API details.