tiny_pacs.events

Event definitions for the tiny_pacs event system.

Events are built on top of trolleybus: every event is a class carrying its payload type and listener result type. Events are never instantiated — the event class itself is broadcast together with a single payload value. Multi-value payloads are packed into small dataclasses.

Lifecycle events are provided by trolleybus itself (trolleybus.OnStart, trolleybus.OnStarted and trolleybus.OnExit) and are emitted by trolleybus.EventBus.start() / trolleybus.EventBus.stop().

Module Attributes

StoredFile

Single stored file: SOP Class UID, Transfer Syntax UID and either a file name, a dataset or a file object.

Classes

Assoc()	Incoming association request.
AssocPayload(asce, assoc)	Payload of the Assoc event.
Atomic()	Request an atomic transaction context manager.
Commitment()	Storage Commitment request.
DeviceByAE()	Request device settings by AE Title.
DeviceConfigs()	Request the configured remote devices.
Find()	Incoming C-FIND request.
FindPayload(context, ds)	Payload of the Find event.
Get()	Incoming C-GET request.
GetClient()	Request a DICOM client for the provided AE Title.
GetFile()	Request a file object to store incoming dataset.
GetFilePayload(context, command_set)	Payload of the GetFile event.
GetFiles()	Request stored files by a list of SOP Instance UIDs.
GetPayload(context, ds)	Payload of the Get event.
MainAET()	Request the main AE Title of the service.
Migrations()	Request schema migrations from components.
Move()	Incoming C-MOVE request.
MovePayload(context, ds, destination)	Payload of the Move event.
Store()	Incoming C-STORE request.
StoreDone()	Dataset has been stored successfully.
StoreFailure()	Dataset storage has failed.
StorePayload(context, ds)	Payload of the Store event.
StoreVerify()	Verify that provided SOP Instance UIDs are stored.
StringAgg()	Request the string aggregate SQL function for the current DB driver.
Tables()	Request a list of database tables from components.

tiny_pacs.events.StoredFile

Single stored file: SOP Class UID, Transfer Syntax UID and either a file name, a dataset or a file object. UID values can come either as pydicom.uid.UID objects or as plain strings (e.g. from the database).

alias of tuple[str | UID, str | UID, str | Dataset | BinaryIO]

class tiny_pacs.events.AssocPayload(asce: AssociationAcceptor, assoc: AAssociateRqPDU)

Bases: object

Payload of the Assoc event.

asce: AssociationAcceptor

Association acceptor handling the incoming connection

assoc: AAssociateRqPDU

Association request parameters

class tiny_pacs.events.Assoc

Bases: Event[AssocPayload, None]

Incoming association request.

event_id: ClassVar[uuid.UUID] = UUID('811eadee-935e-41fc-ba4c-ed2c0f1f4b73')

name: ClassVar[str] = 'Assoc'

class tiny_pacs.events.StorePayload(context: PContextDef, ds: BinaryIO | bytes)

Bases: object

Payload of the Store event.

context: PContextDef

Presentation context

ds: BinaryIO | bytes

Incoming dataset: file object (when SOP Class is stored in file) or raw encoded bytes

class tiny_pacs.events.Store

Bases: Event[StorePayload, Status]

Incoming C-STORE request. Handling result is a C-STORE status.

event_id: ClassVar[uuid.UUID] = UUID('28d1903f-c58d-41fe-903a-1666060d7e09')

name: ClassVar[str] = 'Store'

class tiny_pacs.events.FindPayload(context: PContextDef, ds: Dataset)

Bases: object

Payload of the Find event.

context: PContextDef

Presentation context

ds: Dataset

C-FIND request dataset

class tiny_pacs.events.Find

Bases: Event[FindPayload, Iterable[tuple[Dataset, Status]]]

Incoming C-FIND request.

Handling result is an iterable of (dataset, status) tuples.

event_id: ClassVar[uuid.UUID] = UUID('1f5db3f1-89a1-4b26-be2b-70afc908f8d3')

name: ClassVar[str] = 'Find'

class tiny_pacs.events.MovePayload(context: PContextDef, ds: Dataset, destination: str)

Bases: object

Payload of the Move event.

context: PContextDef

Presentation context

ds: Dataset

C-MOVE request dataset

destination: str

C-MOVE destination AE title

class tiny_pacs.events.Move

Bases: Event[MovePayload, list[tuple[str | UID, str | UID, str | Dataset | BinaryIO]]]

Incoming C-MOVE request. Handling result is a list of stored files.

event_id: ClassVar[uuid.UUID] = UUID('c0297596-40ae-4493-80fc-0da0793abe7e')

name: ClassVar[str] = 'Move'

class tiny_pacs.events.GetPayload(context: PContextDef, ds: Dataset)

Bases: object

Payload of the Get event.

context: PContextDef

Presentation context

ds: Dataset

C-GET request dataset

class tiny_pacs.events.Get

Bases: Event[GetPayload, list[tuple[str | UID, str | UID, str | Dataset | BinaryIO]]]

Incoming C-GET request. Handling result is a list of stored files.

event_id: ClassVar[uuid.UUID] = UUID('49316fdd-d210-49eb-b55d-1e61ab2ec7f5')

name: ClassVar[str] = 'Get'

class tiny_pacs.events.Commitment

Bases: Event[list[tuple[UID, UID]], tuple[list[tuple[UID, UID]], list[tuple[UID, UID]]]]

Storage Commitment request.

Payload is a list of (SOP Class UID, SOP Instance UID) tuples; handling result is a tuple of two lists — successfully stored instances and failures.

event_id: ClassVar[uuid.UUID] = UUID('e0aff352-53bb-42c1-8733-1f382d8ed699')

name: ClassVar[str] = 'Commitment'

class tiny_pacs.events.GetFilePayload(context: PContextDef, command_set: Dataset)

Bases: object

Payload of the GetFile event.

context: PContextDef

Presentation context

command_set: Dataset

Command dataset of the received message

class tiny_pacs.events.GetFile

Bases: Event[GetFilePayload, tuple[BinaryIO, int]]

Request a file object to store incoming dataset.

Handling result is a tuple of the file object and the starting position of the dataset stream within it.

event_id: ClassVar[uuid.UUID] = UUID('cfe34465-dd2d-4d63-a4f7-f3e2ec1f01a5')

name: ClassVar[str] = 'GetFile'

class tiny_pacs.events.MainAET

Bases: Event[None, str]

Request the main AE Title of the service.

event_id: ClassVar[uuid.UUID] = UUID('1a9dadc1-cfbb-4b37-8dfd-882b2a38e083')

name: ClassVar[str] = 'MainAET'

class tiny_pacs.events.StoreDone

Bases: Event[Dataset, None]

Dataset has been stored successfully. Payload is the stored dataset.

event_id: ClassVar[uuid.UUID] = UUID('13c4eee8-3e8c-4b8b-8976-69dac47a3256')

name: ClassVar[str] = 'StoreDone'

class tiny_pacs.events.StoreFailure

Bases: Event[Dataset, None]

Dataset storage has failed. Payload is the dataset that failed.

event_id: ClassVar[uuid.UUID] = UUID('2ae23bb2-d3b0-40e1-be75-fd2d722b8d4b')

name: ClassVar[str] = 'StoreFailure'

class tiny_pacs.events.GetFiles

Bases: Event[list[str], Iterable[tuple[str | UID, str | UID, str | Dataset | BinaryIO]]]

Request stored files by a list of SOP Instance UIDs.

event_id: ClassVar[uuid.UUID] = UUID('14ce2495-447c-4f41-99e8-856baa0a0df4')

name: ClassVar[str] = 'GetFiles'

class tiny_pacs.events.StoreVerify

Bases: Event[list[tuple[UID, UID]], tuple[frozenset[tuple[UID, UID]], frozenset[tuple[UID, UID]]]]

Verify that provided SOP Instance UIDs are stored.

Payload is a list of (SOP Class UID, SOP Instance UID) tuples; handling result is a tuple of two frozensets — successes and failures.

event_id: ClassVar[uuid.UUID] = UUID('6687a638-9f4c-47b9-8a18-f4375d9bf08c')

name: ClassVar[str] = 'StoreVerify'

class tiny_pacs.events.Atomic

Bases: Event[None, Any]

Request an atomic transaction context manager.

event_id: ClassVar[uuid.UUID] = UUID('2c2d9782-8527-469c-9697-7999e3ffe88c')

name: ClassVar[str] = 'Atomic'

class tiny_pacs.events.Tables

Bases: Event[None, list[type[Model]]]

Request a list of database tables from components.

event_id: ClassVar[uuid.UUID] = UUID('049ef1f1-d50c-4250-8131-e4ed5a31f650')

name: ClassVar[str] = 'Tables'

class tiny_pacs.events.Migrations

Bases: Event[None, schema.ComponentMigrations]

Request schema migrations from components.

Handling result is the component’s schema name, its tables and its migration list (see tiny_pacs.schema.ComponentMigrations).

event_id: ClassVar[uuid.UUID] = UUID('7b07c1ed-d6a2-49b4-8b2f-59e767779c7f')

name: ClassVar[str] = 'Migrations'

class tiny_pacs.events.StringAgg

Bases: Event[None, Callable[[…], Function]]

Request the string aggregate SQL function for the current DB driver.

event_id: ClassVar[uuid.UUID] = UUID('2056869d-677f-4fb8-a22d-25df88d584ac')

name: ClassVar[str] = 'StringAgg'

class tiny_pacs.events.DeviceByAE

Bases: Event[str, devices.DeviceConfig | None]

Request device settings by AE Title.

event_id: ClassVar[uuid.UUID] = UUID('60a56c9e-6c59-4752-b98d-c8f6d5d5db5e')

name: ClassVar[str] = 'DeviceByAE'

class tiny_pacs.events.DeviceConfigs

Bases: Event[None, dict[str, 'devices.DeviceConfig']]

Request the configured remote devices.

Answered by device registry components with a mapping of AE titles to device configurations, e.g. so database-backed registries can import the YAML-configured devices.

event_id: ClassVar[uuid.UUID] = UUID('36a8a653-3597-4d16-8cbf-567d66e6507d')

name: ClassVar[str] = 'DeviceConfigs'

class tiny_pacs.events.GetClient

Bases: Event[str, client.DICOMClient]

Request a DICOM client for the provided AE Title.

event_id: ClassVar[uuid.UUID] = UUID('890278b1-5a68-44f2-89a4-6603a038b5dc')

name: ClassVar[str] = 'GetClient'