Guida rapida all'SDK dei processori

Un processore è un programma esterno alla piattaforma che annota come farebbe una persona. Preleva i job via HTTPS, quindi non ha bisogno di porte in ingresso; decifra solo gli input che gli sono stati concessi, cifra tutto ciò che scrive e pubblica i suoi risultati come proposte che i membri del progetto correggono.

L’SDK Python (yggdrasil-processor, Python 3.10 o successivo) gestisce il protocollo e la crittografia. Tu scrivi una funzione che riceve un Job.

1. Installa

pip install "yggdrasil-processor[volumes]"   # the volumes extra adds numpy, nibabel and pydicom

Per installare invece dai sorgenti:

git clone https://github.com/lorenzobrg/yggdrasil4
pip install "./yggdrasil4/sdk/python-processor[volumes]"

2. Scrivi la funzione

threshold.py

from yggdrasil_processor import Job
from yggdrasil_processor.volumes import labelmap_bytes, load_series


def process(job: Job) -> None:
    for series in job.targets_as("series"):
        volume = load_series(series, job)  # an [i, j, k] numpy array on the canonical geometry
        mask = (volume.data > job.parameters["threshold"]).astype("uint8")
        job.log(f"{int(mask.sum())} voxels", progress=1.0)
        job.add_labelmap("mask", series, labelmap_bytes(volume.geometry, mask), {1: "foreground"})

• job.targets_as("series") elenca le serie del job, con i loro file decifrati e i metadati.
• load_series(series, job) legge il volume di una serie. Una serie con più volumi (echi MR, fasi, direzioni di diffusione, un file NIfTI 4D) restituisce per default il primo; load_series(series, job, volume=1) ne legge un altro, e series_volumes(series) elenca ciò che li distingue. Mappe di etichette e forme stanno sulla geometria della serie, quella del primo volume.
• job.parameters è validato rispetto allo schema dei parametri del descrittore, con i valori predefiniti già inseriti.
• job.log(...) invia una riga di log cifrata e l’avanzamento che i membri del progetto vedono.
• job.add_labelmap, job.add_annotation e job.add_asset aggiungono output. L’SDK verifica ogni risultato rispetto allo schema del suo tipo, cifra e carica gli output, poi pubblica i risultati. Il server non può leggere i risultati, quindi uno non valido fa fallire il job prima che venga pubblicato qualsiasi cosa. job.add_frame_mask aggiunge la maschera di un fotogramma di un video: una propagazione ne scrive una per fotogramma.
• Un’eccezione fa fallire il job con il suo messaggio, cifrato per i membri del progetto. Le funzioni lunghe chiamano job.check() per fermarsi quando il job viene annullato.

3. Descrivilo

Un descriptor.json dichiara input, output, parametri, trigger e accesso del processore. Segue lo schema del descrittore dei processori:

descriptor.json

{
  "id": "ygg.threshold",
  "version": "1.0.0",
  "name": "Threshold segmentation",
  "description": "Labels every voxel above a threshold.",
  "inputs": { "level": "series", "modalities": ["ygg.volume@1"] },
  "outputs": [{ "id": "mask", "type": "ygg.segmentation.labelmap@1", "labels": ["foreground"] }],
  "parameters": {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "type": "object",
    "properties": { "threshold": { "type": "number", "default": 300 } },
    "additionalProperties": false
  },
  "triggers": ["manual", "on-upload"],
  "access": "per-job"
}

L’accesso per-job, quello predefinito, significa che il processore riceve solo le chiavi degli input di ciascun job.

4. Registralo

1. Genera le chiavi del processore. Le chiavi private vengono scritte con modalità 0600; le chiavi pubbliche vengono stampate.

   ygg-processor keygen --keys processor-keys.json

2. Un amministratore del progetto aggiunge il processore al manifest del progetto (configurazione, passaggio Processori) con il file del descrittore, oppure lo sceglie tra i processori offerti dall’installazione (il catalogo che un superamministratore gestisce nella console; i byte del descrittore sono quelli del catalogo, e il badge “Offerto da …” indica l’istituzione dell’installazione, non una verifica), pubblica, poi lo registra in Processori con il descrittore e le chiavi pubbliche. Il browser dell’amministratore firma la registrazione; vengono mostrati l’ID della registrazione, un token e la radice del progetto. Il token viene mostrato una sola volta. Un processore con accesso all’intero progetto riceve la chiave del progetto, quindi l’amministratore lo conferma prima di firmare.

3. Verifica la registrazione e il log firmato del progetto senza prendere in carico un job:

   export YGG_PROCESSOR_URL=https://yggdrasil.example.org   # the instance's API
   export YGG_PROCESSOR_ID=<registration ID>
   export YGG_PROCESSOR_TOKEN=<token>
   export YGG_PROCESSOR_ROOT=<project root>                 # the trust anchor
   ygg-processor check

   La radice del progetto è l’ancora di fiducia del processore: il processore rifiuta una storia del progetto con un’altra radice, quindi il server non può mostrargli una storia contraffatta. Ottienila dall’amministratore attraverso un canale di cui ti fidi. Senza una radice il processore non si avvia. --tofu (YGG_PROCESSOR_TOFU=1) si fida invece della prima storia che vede, cosa che un server potrebbe sfruttare; usalo solo per i test.

5. Eseguilo

ygg-processor run threshold:process

Il runner prende in carico job finché non viene fermato; --once termina dopo un job. --jobs N esegue fino a N job contemporaneamente in un solo processo (1 per default), e --transfers N imposta quanti download o caricamenti un job esegue contemporaneamente (4 per default). Un download interrotto a metà riprende da dove si era fermato. Il primo SIGINT o SIGTERM lascia terminare i job in corso; un secondo ferma tutto subito e i job tornano in coda quando i loro lease scadono.

Prima di decifrare o sigillare qualsiasi cosa, l’SDK riesegue il log firmato del progetto, verifica che la sua radice sia quella configurata, la fissa insieme alla testa più avanzata vista, e verifica che la registrazione indichi queste chiavi e esattamente questo descrittore. Le chiavi concesse al processore sono legate alla radice del progetto, quindi si aprono solo con la storia reale. Ogni output scritto dall’SDK (maschere, risultati, righe di log, dettagli dei fallimenti) è firmato con la chiave del processore; i browser dei membri verificano quella firma rispetto alla registrazione prima di mostrare o accettare l’output, e concedono a un job solo input che hanno verificato essere all’interno del suo target. Conserva lo stato di fiducia (--state) su uno storage persistente, per esempio un volume Docker. Gli input decifrati vengono scritti in una directory di lavoro che viene eliminata al termine del job, quindi esegui i processori su macchine ritenute affidabili per i dati del progetto.

Nuove serie

Un processore può anche scrivere una nuova serie accanto a quella su cui ha girato, della stessa modalità o di un’altra: una panoramica proiettata da una CBCT, un volume ricampionato, un’immagine di un vetrino. Il suo descrittore dichiara un output series con il plugin di modalità che produce, e il manifest del progetto lega quell’output a una delle sue voci di modalità:

{ "id": "projection", "kind": "series", "modality": "ygg.image2d@1", "formats": ["png"] }

def process(job: Job) -> None:
    (series,) = job.targets_as("series")
    derived = job.add_series(series, "projection", {"description": "Maximum-intensity projection"})
    derived.add_asset(png_bytes, {"format": "png", "mediaType": "image/png", "image": {"width": w, "height": h}})

L’originale non viene mai modificato. La nuova serie è cifrata e firmata come ogni output, e i suoi metadati firmati indicano la serie di origine, il processore e il job, che i browser dei membri verificano. I membri la vedono quando il job è riuscito, nella worklist con il segno “Derivata” e nell’area di lavoro con un link alla serie di origine; si apre nel visualizzatore della sua modalità, riceve le attività di quella modalità ed è esportata come ogni serie. È un dato come gli altri, non una proposta: chi può eliminare serie la elimina, ed eliminare la serie di origine elimina anche lei. Anche i processori che girano al caricamento girano sulle nuove serie (concatenamento), avviati dal browser di un membro che può eseguire processori mentre l’area di lavoro è aperta; ciascuno gira una volta per serie, mai su una serie derivata da un proprio output. examples/projection proietta così i volumi in immagini.

Processori in cascata

Un processore può leggere ciò che un altro ha scritto. Un classificatore multi-classe che ha bisogno della segmentazione di una CBCT, per esempio, gira dopo un segmentatore di denti: il suo descrittore nomina l’input con un ID, il tipo che legge e, se serve, gli stati che contano (per impostazione predefinita le annotazioni inviate; accepted in un descrittore indica le inviate da quando la correzione ha sostituito la revisione):

"inputs": {
  "level": "series",
  "modalities": ["ygg.volume@1"],
  "annotations": [
    { "id": "segmentation", "type": "ygg.segmentation.labelmap@1", "description": "The series' segmentation" }
  ]
}

Il manifest del progetto lega quell’input a un’attività: quella in cui scrive l’output del segmentatore. Il wizard di configurazione lo lega all’unica attività del suo tipo (o ne crea una) e lo mostra come “Richiede segmentation dall’attività …”, che l’amministratore può cambiare:

"processors": [
  { "id": "segmenter", "trigger": "on-upload", "access": "per-job", "outputs": [{ "output": "mask", "task": "teeth" }] },
  {
    "id": "classifier", "trigger": "on-upload", "access": "per-job",
    "inputs": [{ "input": "segmentation", "task": "teeth" }],
    "outputs": [{ "output": "burden", "task": "caries" }]
  }
]

La funzione legge l’input dal suo ID e apre la labelmap sulla sua serie:

def process(job: Job) -> None:
    for series in job.targets_as("series"):
        total_ml = 0.0
        for annotation in job.inputs("segmentation", series):  # quelle inviate dell'attività legata
            total_ml += sum(job.read_labelmap(annotation).volumes_ml().values())
        job.add_annotation("burden", series, {"choice": "large" if total_ml >= 1 else "small"})

Entrambi girano al caricamento, e il classificatore aspetta: una nuova CBCT avvia il segmentatore, un membro corregge o conferma la sua segmentazione (o l’attività prende i suoi risultati così come sono), e allora il browser di un membro che può eseguire processori avvia il classificatore su quella serie, una volta. La pagina di caricamento e la finestra di esecuzione dicono cosa aspetta un processore automatico, e la pagina Processori mostra ogni catena in una riga, “CBCT → Segmentazione denti (Segmentatore denti) → Carie (Classificatore carie)”. Una modifica successiva della segmentazione non riesegue il classificatore: eseguilo a mano. examples/classify-from-segmentation è questo classificatore.

Altro

• Il README dell’SDK elenca ogni impostazione, codice di uscita e controllo.
• docker/Dockerfile impacchetta un processore come immagine.
• examples/totalsegmentator è il processore di riferimento: TotalSegmentator, con un descrittore per la segmentazione TC total body (total) e uno per le strutture craniofacciali su TC e CBCT.
• examples/sam2-video propaga lungo il video le maschere di fotogrammi di un lettore con SAM 2.1, scrivendo una proposta di maschera su ogni frame annotato, che il lettore adotta su tutti i frame in una volta.
• I processori interattivi rispondono a un membro mentre lavora, con ygg-processor serve module:handler: examples/region-grow segmenta a partire da clic, e examples/scripted-captions mostra la dettatura dal vivo con transcription_handler (un processore di trascrizione riceve la voce del lettore e nessun dato del progetto; non deve conservare audio).
• L’API HTTP dei processori è il contratto, per processori scritti in altri linguaggi.