TSignalDev/tsignal-pythonPublic

NotificationsYou must be signed in to change notification settings
Fork0
Star24

A Python Signal-Slot library inspired by Qt, featuring thread-safe communication, async support, and automatic connection type detection. Perfect for decoupled event-driven architecture and thread synchronization.

24 stars 0 forks Branches Tags Activity

Star

Notifications

You must be signed in to change notification settings

Branches Tags

Folders and files

Name		Name	Last commit message	Last commit date
Latest commit History 10 Commits
.github/workflows		.github/workflows
docs		docs
examples		examples
src/tsignal		src/tsignal
tests		tests
.gitattribuites		.gitattribuites
.gitignore		.gitignore
CHANGELOG.md		CHANGELOG.md
CONTRIBUTING.md		CONTRIBUTING.md
LICENCSE.md		LICENCSE.md
README.md		README.md
pyproject.toml		pyproject.toml

Repository files navigation

TSignal

Are you looking for a signal/slot pattern in Python without Qt dependencies? TSignal provides a lightweight, thread-safe, and asyncio-compatible implementation that gives you all the power of Qt's signal/slot pattern without the heavyweight dependencies. Perfect for:

Async applications needing event handling
Thread communication in Python applications
Event-driven architectures
Decoupled component communication

Why TSignal?

🚀 Pure Python implementation - no Qt or external dependencies required
⚡ Async/await support out of the box
🔒 Thread-safe signal emission and slot execution
🎯 Simple, decorator-based API similar to Qt
🔄 Automatic thread handling for cross-thread signals

Quick Start

Basic Example

fromtsignalimportt_with_signals,t_signal,t_slot@t_with_signalsclassCounter:def__init__(self):self.count=0@t_signaldefcount_changed(self):passdefincrement(self):self.count+=1self.count_changed.emit(self.count)@t_with_signalsclassDisplay:@t_slotasyncdefon_count_changed(self,value):print(f"Count is now:{value}")# Connect and usecounter=Counter()display=Display()counter.count_changed.connect(display,display.on_count_changed)counter.increment()# Will print: "Count is now: 1"

Async Example

@t_with_signalsclassAsyncDisplay:@t_slotasyncdefon_count_changed(self,value):awaitasyncio.sleep(1)# Simulate async operationprint(f"Count updated to:{value}")# Usage in async contextasyncdefmain():counter=Counter()display=AsyncDisplay()counter.count_changed.connect(display,display.on_count_changed)counter.increment()# Wait for async processingawaitasyncio.sleep(1.1)asyncio.run(main())

Features

Requires Python 3.10+
Easy-to-use signal-slot mechanism with decorators
Support for both synchronous and asynchronous slots
Thread-safe signal emissions
Automatic connection type detection (direct/queued)
Compatible with Python's asyncio

Installation

TSignal requires Python 3.10 or higher. You can install it directly from the repository:

git clone https://github.com/tsignal/tsignal-python.gitcd tsignal-pythonpip install -e.

For development installation (includes test dependencies):

pip install -e".[dev]"

Documentation

Development

Logging

TSignal uses Python's standard logging module. For detailed logging configuration,please seeLogging Guidelines.

Basic usage:

importlogginglogging.getLogger('tsignal').setLevel(logging.INFO)

Testing

TSignal includes a comprehensive test suite using pytest. For basic testing:

# Run all testspytest# Run with verbose outputpytest -v# Run specific test filepytest tests/unit/test_signal.py

For detailed testing instructions and guidelines, seeTesting Guide.

Contributing

Please seeContributing Guidelines for details on how to contribute to this project.

License

This project is licensed under the MIT License - see theLICENSE file for details.

Connecting Signals and Slots

Classic Object-Member Connection

@t_with_signalsclassCounter:@t_signaldefcount_changed(self):pass@t_with_signalsclassDisplay:@t_slotdefon_count_changed(self,value):print(f"Count is now:{value}")counter=Counter()display=Display()counter.count_changed.connect(display,display.on_count_changed)

Function Connection

# Connect to a simple functiondefprint_value(value):print(f"Value:{value}")counter.count_changed.connect(print_value)# Connect to a lambdacounter.count_changed.connect(lambdax:print(f"Lambda received:{x}"))# Connect to an object method without @t_slotclassHandler:defprocess_value(self,value):print(f"Processing:{value}")handler=Handler()counter.count_changed.connect(handler.process_value)

Worker Thread Pattern

TSignal provides a worker thread pattern that combines thread management with signal/slot communication and task queuing:

fromtsignalimportt_with_worker@t_with_workerclassDataProcessor:asyncdefinitialize(self,config=None):# Setup worker (called in worker thread)self.config=configor {}asyncdefprocess_data(self,data):# Heavy processing in worker threadresult=awaitheavy_computation(data)self.processing_done.emit(result)asyncdeffinalize(self):# Cleanup worker (called before thread stops)awaitself.cleanup()@t_signaldefprocessing_done(self):pass# Usageprocessor=DataProcessor()processor.start(config={'threads':4})# Starts worker thread# Queue task in worker threadawaitprocessor.queue_task(processor.process_data(some_data))# Stop workerprocessor.stop()# Graceful shutdown