Python API

Dixel API

class diana.dixel.Dixel(meta=NOTHING, tags=NOTHING, level=<DicomLevel.STUDIES: 2>, parent=None)

“Dixels” are DICOMish-elements (following pixels, voxels, and texels). They may include metadata, tags, a file, and a pixel array. All Dixels have an id, typically following the Orthanc format, and a DicomLevel (study, series, or instance).

DIANA endpoints handle and store dixel instances. Some functions may take a dixel identifier and return the dixel instance.

children = None

Stores information about sub-dixels (series for study, instances for series)

file = None

Stores binary file representation

fn

Filename alias for meta[‘Filename’]

static from_montage_csv(data: Mapping)

Generate a dixel from a line in a Montage csv download

static from_montage_json(data: Mapping)

Generate a dixel from a Montage JSON result (as returned by the Montage Endpoint.

Metadata includes Montage-mapped CPT codes; to dereference them to real CPT codes and body parts, call Montage().get_meta(dixel)

static from_orthanc(meta: Mapping = None, tags: Mapping = None, level: diana.utils.dicom.levels.DicomLevel = <DicomLevel.STUDIES: 2>, file=None)

Generate a dixel from an Orthanc json tag dictionary

static from_pydicom(ds: pydicom.dataset.Dataset, fn: str = None, file=None)

Generate a Dixel from a pydicom dataset

image_base_fn

Filename for image instance

level = None

Study, series, instance

meta = None

Metadata dict

oid()

Compute Orthanc ID

parent = None

Stores reference to parent dixel (study for series, series for instances)

pixels = None

Stores pixel array representation

report = None

Stores study report as RadiologyReport

sid()

Serializer id alias to tags[‘AccessionNumber’]

tags = None

Dicom tag dict

class diana.dixel.DixelView

Some endpoints can only generate certain DixelViews, others can return Dixel instances with data from multiple views.

Use DixelView.TAGS in view to test for different flags.

FILE = 8

In Orthanc, this is the data from /<level>/<oid>/file

META = 1

In Orthanc, this is the data from /<level>/<oid>

METAKV = 2

In Orthanc, this is the data from /<level>/<oid>/metadata/*

PIXELS = 16

Optional for pydicom readers

TAGS = 4

In Orthanc, this is the data from /<level>/<oid>/tags

TAGS_FILE = 12

Dixels also implement the Serializable interface.

Full-Dixel Endpoint APIs

All Dixel Endpoints implement the Serializable and Endpoint interfaces.

Diana-CLI provides shortcuts for calling basic endpoint functions (put, get, find, delete) on arbitrary services.

class diana.utils.endpoint.Endpoint(name: str = 'Endpoint', ctype=None)

Generic CRUD endpoint API.

check() → bool

Check endpoint health

delete(item: Union[NewType.<locals>.new_type, Item], **kwargs) → bool

Remove an item from the endpoint

exists(item: Union[NewType.<locals>.new_type, Item, NewType.<locals>.new_type], **kwargs) → bool

Check if an item exists by id or query

find(item: Union[NewType.<locals>.new_type, Item, NewType.<locals>.new_type], **kwargs) → Union[Sequence[NewType.<locals>.new_type], Sequence[Item]]

Identify items and optionally retrieve data by query

get(item: Union[NewType.<locals>.new_type, Item], **kwargs) → Item

Retrieve item data

handle(item: Union[NewType.<locals>.new_type, Item], method: str, *args, **kwargs)

Call a class-specific method

put(item: Item, **kwargs) → NewType.<locals>.new_type

Add or replace an item in the collection

update(item: Union[NewType.<locals>.new_type, Item], data: Mapping, **kwargs) → NewType.<locals>.new_type

Update data for an item in the endpoint

Orthanc

class diana.apis.Orthanc(ctype=None, name='Orthanc', protocol='http', host='localhost', port=8042, path=None, aet='ORTHANC', peername='orthanc', user='orthanc', password='passw0rd!', meta_keys=<class 'list'>)

check()

Check crud health

delete(item: Union[str, diana.dixel.dixel.Dixel], level=<DicomLevel.STUDIES: 2>, **kwargs)

Remove an item from the crud

find(item: Union[Mapping, diana.dixel.dixel.Dixel], level=<DicomLevel.STUDIES: 2>, **kwargs) → list

Identify items and optionally retrieve data by query

get(item: Union[str, diana.dixel.dixel.Dixel], level: diana.utils.dicom.levels.DicomLevel = <DicomLevel.STUDIES: 2>, view: diana.dixel.views.DixelView = <DixelView.TAGS: 4>, **kwargs)

Retrieve item data

put(item: diana.dixel.dixel.Dixel, **kwargs)

Add or replace an item in the collection

Osimis-flavor Orthanc provides additional services including a webviewer and annotations. Importing osimis_extras extends the Orthanc base-class.

diana.apis.osimis_extras.get_annotation(source: diana.apis.orthanc.Orthanc, item: diana.dixel.dixel.Dixel) → Optional[Mapping]

Method to summarize collections of Osimis-style ROI metadata and monkey-patch for diana.apis.Orthanc

Considered implementing this as an annotation “view” for Orthanc.get, but it did not need to return a dixel as I was using it. So monkey-patching seemed easier for a one-off application.

>>> from diana.apis import Orthanc, osimis_extras
>>> o = Orthanc()
>>> a = o.get_annotation(my_study)

ProxiedDicom

class diana.apis.ProxiedDicom(ctype=None, name='ProxiedDicom', proxy_desc=NOTHING, proxy_domain: str = 'remote')

check()

Check crud health

delete(item: Union[str, diana.dixel.dixel.Dixel], level=<DicomLevel.STUDIES: 2>)

Remove an item from the crud

exists(item, level=<DicomLevel.STUDIES: 2>)

Check if an item exists by id or query

find(item: Union[Mapping, diana.dixel.dixel.Dixel], level=<DicomLevel.STUDIES: 2>, retrieve: bool = False, **kwargs)

Identify items and optionally retrieve data by query

get(item: Union[str, diana.dixel.dixel.Dixel], level=<DicomLevel.STUDIES: 2>, view=<DixelView.TAGS: 4>)

Retrieve item data

DcmDir

class diana.apis.DcmDir(ctype=None, name='DcmDir', path='.', subpath_width=2, subpath_depth=0, anonymizing=False, recurse_style='UNSTRUCTURED')

check()

Check crud health

delete(item: Union[str, diana.dixel.dixel.Dixel], **kwargs)

Remove an item from the crud

exists(item: Union[str, diana.dixel.dixel.Dixel])

Check if an item exists by id or query

get(item: Union[str, diana.dixel.dixel.Dixel], view=<DixelView.TAGS: 4>, **kwargs)

Retrieve item data

class orthanc_subdirs(base_dir=None, low=0, high=65535)

Generates 1024 Orthanc-style nested subdirs

put(item: diana.dixel.dixel.Dixel, **kwargs)

Add or replace an item in the collection

subdirs()

Generator for nested sub-directories.

Performed lazily because this can be expensive for UNSTRUCTURED directories.

class unstructured_subdirs(base_dir)

Generates subdirs with os.walk, this remains _very_ slow for large datasets!

update(fn: str, item: diana.dixel.dixel.Dixel, **kwargs)

Update data for an item in the crud

Dixel-Summary Endpoint APIs

Montage

class diana.apis.Montage(ctype=None, name='Montage', protocol='http', host='localhost', port=80, path='api/v1', user='montage', password='montage')

check()

Check crud health

find(query: Mapping, index='rad', ignore_errs=True, get_meta=False)

Identify items and optionally retrieve data by query

CsvFile

class diana.apis.CsvFile(ctype=None, fp=None, level=<DicomLevel.STUDIES: 2>, name='CsvFile')

exists(item: diana.dixel.dixel.Dixel, **kwargs)

Check if an item exists by id or query

put(item: diana.dixel.dixel.Dixel, force_write=False)

Add or replace an item in the collection

Redis

class diana.apis.Redis(ctype=None, name='Redis', host='localhost', port=6379, db=0, password=None)

add_to_collection(item: diana.dixel.dixel.Dixel, prefix: str = '', collection_key: str = 'AccessionNumber', item_key: str = 'FilePath', path=None)

It is non-obvious how to check the total number of objects across all sets. This works and its fast for even large data sets.

> EVAL “local total = 0 for _, key in ipairs(redis.call(‘keys’, ARGV[1])) do total = total + redis.call(‘scard’, key) end return total” 0 prefix-*

See <https://stackoverflow.com/questions/34563144/redis-multiple-key-set-counts>

check()

Check crud health

delete(item: Union[str, diana.dixel.dixel.Dixel], **kwargs)

Remove an item from the crud

find(query: Mapping, **kwargs)

Identify items and optionally retrieve data by query

get(item: Union[str, crud.abc.serializable.AttrSerializable], **kwargs) → Any

Retrieve item data

put(item: Union[str, crud.abc.serializable.AttrSerializable, Any], **kwargs) → str

Add or replace an item in the collection

update(item: Union[str, crud.abc.serializable.AttrSerializable], data: Union[Any, diana.dixel.dixel.Dixel], **kwargs)

Update data for an item in the crud

Splunk

SMTP Messenger

Daemon APIs

Diana daemons are higher order constructs that combine multiple APIs into pipelines.

Routing

A routing deamon may be created by instantiating a Watcher with ObservableEndpoitns and adding Dixel routing rules.

Diana-CLI provides a shortcut for creating a watcher service and attaching observables and triggers.

class diana.utils.endpoint.Watcher(action_interval=1.0)

Generic event handler. Setup is through Trigger objects that map sources and event types to a partial function call. Events then arrive via event_queues from Observable sources.

Any endpoint that implements “Observable” is an eligible source.

add_trigger(trigger: diana.utils.endpoint.watcher.Trigger)

fire(event: crud.abc.observable.Event)

run()

stop()

Observables implement the Observable API.

class diana.apis.ObservableOrthanc(ctype=None, protocol='http', host='localhost', port=8042, path=None, aet='ORTHANC', peername='orthanc', user='orthanc', password='passw0rd!', meta_keys=<class 'list'>, polling_interval=1.0, name='ObsOrthanc', start_from_change: int = None, persist_file=NOTHING)

Orthanc service that implements a DicomEvents “changes” function for polling.

changes(**kwargs)

persist_current_change()

persist_last_change()

set_current_change()

set_persist_file()

class diana.apis.ObservableProxiedDicom(ctype=None, proxy_desc=NOTHING, proxy_domain: str = 'remote', name='ObsPrxDcm', polling_interval=180.0, query=NOTHING, qlevel: str = <DicomLevel.STUDIES: 2>, qperiod=600, history_len=200)

ProxiedDicom service that implements a DicomEvents “changes” function for polling. Restricted to observing recent events, from now to qperiod seconds in the past.

It is intended that qperiod >> polling_interval to avoid missing events that may have delivered slowly or would be missed by keeping a near-0 overlap between polling and query windows. A history queue tracks the n most recently observed events and suppresses multiple notifications.

changes(**kwargs)

history_len = None

Set to at least 10 mins of studies

polling_interval = None

Poll every 3 mins by default

qperiod = None

Check last 10 mins by default

setup_history()

setup_query()

class diana.apis.ObservableDcmDir(ctype=None, path='.', subpath_width=2, subpath_depth=0, anonymizing=False, recurse_style='UNSTRUCTURED', polling_interval=1.0, name='ObsDcmDir')

DcmDir service that implements a watchdog polling system and converts file events into DicomEvents.

class WatchdogEventReceiver(source)

on_any_event(wd_event: watchdog.events.FileSystemEvent)

Catch-all event handler.

Parameters:event (FileSystemEvent) – The event object representing the file system event.

changes()

poll_events()

Dixel routes are instantiated with (source, dest, handler) tuples and attached to the Watcher’s routing table.

diana.daemons.routes.index_item(item: Mapping, level: diana.utils.dicom.levels.DicomLevel, source: crud.abc.endpoint.Endpoint, dest: crud.abc.endpoint.Endpoint, index=None, token=None)

diana.daemons.routes.mk_route(hname, source_desc, dest_desc=None)

diana.daemons.routes.pull_and_save_item(item: diana.dixel.dixel.Dixel, source: diana.apis.proxied_dicom.ProxiedDicom, dest: diana.apis.dcmdir.DcmDir, anonymize=False) → str

diana.daemons.routes.put_item(item: str, source: crud.abc.endpoint.Endpoint, dest: crud.abc.endpoint.Endpoint, **kwargs)

diana.daemons.routes.query_and_index(query: Mapping, level: diana.utils.dicom.levels.DicomLevel, source: diana.apis.orthanc.Orthanc, domain: str, dest: crud.endpoints.splunk.Splunk, token: str, index: str)

diana.daemons.routes.say(item: str, suffix: str = None)

diana.daemons.routes.send_item(oid: str, level: diana.utils.dicom.levels.DicomLevel, source: diana.apis.orthanc.Orthanc, dest: Union[diana.apis.orthanc.Orthanc, str], remove_src=True, anonymize=False, remove_anon=True)

diana.daemons.routes.upload_item(item: Mapping, source: diana.apis.dcmdir.DcmDir, dest: diana.apis.orthanc.Orthanc, anonymizing=False)

Collector

A collector is similar to a Watcher, but it processes historical data based on an input list of studies.

Collect a list of accession numbers, process them, and save them

1. Get patient info and StudyUIDs for each a/n
2. Copy each item and anonymize it
3. Send each item to the destination orthanc/path for review

class diana.daemons.collector.Collector(pool_size=0)

create_pool()

handle_worklist(items: Iterable, source: diana.apis.orthanc.Orthanc, domain: str, dest: Union[diana.apis.orthanc.Orthanc, diana.apis.dcmdir.DcmDir], anonymize)

make_key(ids, source: diana.apis.orthanc.Orthanc, domain: str) → set

pull_and_save(items: Iterable, source: diana.apis.orthanc.Orthanc, domain: str, dest: diana.apis.dcmdir.DcmDir, anonymize=False)

pull_and_send(items: Iterable, source: diana.apis.orthanc.Orthanc, domain: str, dest: diana.apis.orthanc.Orthanc, anonymize=False)

run(project: str, data_path: pathlib.Path, source: diana.apis.orthanc.Orthanc, domain: str, dest: Union[diana.apis.orthanc.Orthanc, diana.apis.dcmdir.DcmDir], anonymize=False)

File Indexer

class diana.daemons.file_indexer.FileIndexer(pool_size=0)

Create a registry for all files in a DICOM directory and subdirs

create_pool()

index_path(basepath, registry, rex='*.dcm', recurse_style='UNSTRUCTURED')

static items_on_path(basepath, registry)

static prefix_for_path(basepath)

upload_collection(collection, basepath, registry, dest)

upload_path(basepath, registry: diana.apis.legacy.redis.Redis, dest: diana.apis.orthanc.Orthanc)

diana.daemons.file_indexer.index_file(fn, path=None, reg=None, prefix=None, _checked: multiprocessing.context.BaseContext.Value = <Synchronized wrapper for c_int(0)>, _registered: multiprocessing.context.BaseContext.Value = <Synchronized wrapper for c_int(0)>)

diana.daemons.file_indexer.put_inst(fn, dest, _uploaded: multiprocessing.context.BaseContext.Value = <Synchronized wrapper for c_int(0)>)

Mock Data Generation

Reasonably well-formed DICOM header data for mock studies can be generated on a site, service, or device basis.

Diana-CLI provides a shortcut for defining and running a mock service.

class diana.daemons.mock_site.MockDevice(site_name='Mock Facility', station_name='Imaging Device', modality='CT', studies_per_hour=6, action_interval=5.0)

Generates studies on a schedule

gen_study(study_datetime=None)

poll(dest: diana.apis.orthanc.Orthanc = None)

class diana.daemons.mock_site.MockService(site_name='Mock Facility', name='Imaging Service', modality='CT', num_devices=1, studies_per_hour=6)

Set of similar modality imaging devices

setup_devices()

class diana.daemons.mock_site.MockSite(name='Mock Facility')

Set of imaging services

class Factory

classmethod create(desc: dict)

add_service(name, modality, devices, studies_per_hour)

devices()

run(pacs)

class diana.dixel.mock_dixel.MockInstance(meta=NOTHING, tags=NOTHING, parent=None, inst_num=0)

as_pydicom_ds()

Return a pydicom dataset suitable for writing to disk. This is primarily useful for dixel-mocking.

gen_file(fn=None)

set_inst_datetime()

set_instuid()

class diana.dixel.mock_dixel.MockSeries(meta=NOTHING, tags=NOTHING, parent=None, ser_num=0, ser_desc='DICOM Series', n_instances=1)

set_series_datetime()

set_seruid()

class diana.dixel.mock_dixel.MockStudy(meta=NOTHING, tags=NOTHING, parent=None, study_datetime: datetime.datetime = NOTHING, site_name: str = 'Mock Site', station_name: str = 'Scanner', modality: str = 'CT')

This is a study-level root dixel

set_study_description()

diana.dixel.mock_dixel.reset_mock_seed()

Reset the mock seed and mock id for testing

DICOM Utilities

class diana.utils.dicom.DicomLevel

Enumeration of DICOM levels, ordered by general < specific

INSTANCES = 4

PATIENTS = 1

SERIES = 3

STUDIES = 2

from_label = <function DicomLevel.from_label>

diana.utils.dicom.dicom_simplify(tags, ignore_errors=True)

Simplify a DICOM tag set:

• Standardize dates and times as Python datetime objects
• Identify a sensible creation datetime
• Flatten and simplify ContentSequences in the manner of Orthanc’s ‘simplify’ parameter
• Add sensible defaults for missing station names
• Add sensible defaults for exposure data in dose reports

class diana.utils.dicom.DicomUIDMint(app_id='dicom')

Minting reproducible UIDs is important for anticipating the OIDs of anonymized studies and for ensuring studies are linked by StudyUID even when individual series are anonymized with different conditions.

hierarchical_suffix(PatientID: str, StudyInstanceUID: str, SeriesInstanceUID=None, SOPInstanceUID=None)

A hierarchical asset uid has the form:

prefix.app.patient.study.series.instance

Where

• prefix = 25 digits 25
• app = stop + 2 digits 3 = 28
• pt, study = stop + 12 digits each 26 = 54

• series, instance = stop + 4 digits each (optional)

  10 = 64

Total length is 64

prefix = '1.2.826.0.1.3680043.10.43'

Prefix parts:

• 1 - iso
• 2 - ansi
• 840 - us
• 0.1.3680043.10.43 - sub-organization within medicalconnections.co.uk

A 25 character prefix leaves 39 digits and stops available (64 chars max)

uid(PatientID: str = None, StudyInstanceUID: str = None, SeriesInstanceUID=None, SOPInstanceUID=None)

app fields immediately following prefix with 2 digits are asset or common uids (pt, st, ser, inst).

asset_uid takes up to 4 parameters (pt, st, ser, inst) that will be converted to strings and hashed.

Non-asset uids will have app fields >2 digits.

GUID Utilities

class diana.utils.guid.GUIDMint

Mint for generating reproducible pseudo-identities, random is isolated and should not affect other calls to the PRNG

classmethod get_hash(name: str, dob: datetime.date = None, age: int = None, reference_date: datetime.date = None, gender=<GUIDGender.UNKNOWN: 'U'>) → [<built-in function openssl_sha1>, <class 'datetime.date'>]

Use the study date as a reference date for reproducible dob given age only

classmethod get_id(h: _hashlib.openssl_sha1) → str

classmethod get_name(id: str, gender=<GUIDGender.UNKNOWN: 'U'>) → list

classmethod get_new_dob(id: str, dob: datetime.date)

classmethod get_sham_id(name: str, dob: Union[datetime.date, str] = None, age: int = None, reference_date: Union[datetime.date, str] = None, gender='U')

classmethod get_time_offset(id: str)

names = {}

REST Gateway Utilities

Orthanc

class diana.utils.gateways.Orthanc(protocol='http', host='localhost', path=None, name='OrthancGateway', port=8042, user='orthanc', password='passw0rd!', aet='ORTHANC')

Diana-agnostic API and helpers for Orthanc, with no endpoint or dixel dependencies.

anonymize(oid, level, replacement_map)

changes(current=0, limit=10)

delete(oid, level)

find(query)

get(oid: str, level: diana.utils.dicom.levels.DicomLevel, view: str = 'tags')

get_metadata(oid: str, level: diana.utils.dicom.levels.DicomLevel, key: str)

inventory(level=<DicomLevel.STUDIES: 2>)

modify(oid, level, replacement_map)

put(file)

put_metadata(oid: str, level: diana.utils.dicom.levels.DicomLevel, key: str, value: str)

recho(domain)

reset()

rfind(query, domain, retrieve=False)

send(oid: str, dest: str, dest_type)

dest_type should be either ‘peers’ or ‘modalities’

statistics()

diana.utils.gateways.orthanc_id(PatientID: str, StudyInstanceUID: str, SeriesInstanceUID=None, SOPInstanceUID=None) → str

Montage

class diana.utils.gateways.Montage(protocol='http', host='localhost', port=80, password='passw0rd!', name='MontageGateway', user='montage', path='apis/v1', index='rad')

Diana-agnostic API for Montage, with no endpoint or dixel dependencies

There is no montage-sdk for Python afaik; this gateway provides a minimal functionality to ‘find’ events and collect some additional metadata about body part and cpt code.

classmethod clean_text(text)

Clean up text from the RIH report templates. Recent variants are returned with r indicators, older variants are not, so we insert newlines based on whether the next line is a continuation or an obviously new section.

find(query: Mapping, index: str = None) → list

lookup_body_part(montage_cpts: list) → list

Build up cached lookup tables for body part from exam code.

lookup_cpts(montage_cpts: list) → list

Build up cached lookup table for cpt codes from exam code.

Splunk

File Gateway Utilities

class diana.utils.gateways.DcmFileHandler(path='.', subpath_width=2, subpath_depth=0, name='DcmFileHandler')

get(fn: str, get_pixels=False, force=False) → pydicom.dataset.Dataset

static is_dicom(item) → bool

put(fn: str, data)

class diana.utils.gateways.TextFileHandler(name='FileHandler', path='.', subpath_width=2, subpath_depth=0)

put(fn: str, data: str)

class diana.utils.gateways.ImageFileHandler(name='FileHandler', path='.', subpath_width=2, subpath_depth=0)

get(fn: str)

put(fn: str, data, max_size=1024)

static squash_to_8bit(data: numpy.array)