WARNING

Nextflow es una herramienta orientada a casos de uso mucho más interesantes pero este ejemplo creo que puede servir para prácticar un poco y ver algunas funcionalidades de este DSL

En este post vamos a desarrollar un pipeline de Nextflow para organizar las fotos de un directorio y ordernarlas por el sitio que fueron tomadas respecto de un punto que indiquemos.

Para ello el pipeline accederá a la meta información de cada foto y buscará si tiene la posición donde fue tomada. Si la foto no cuenta con esta info, la foto será ignorada.

Una vez obtenida la información de todas las fotos, el pipeline las ordenará según lo lejos que se encuentren de un punto dado (en formato "latitud, longitud"). Si no se proporciona ningun punto se tomará el punto [0,0] como referencia.

La idea es obtener al final del proceso un directorio con las imágenes copiadas pero renombradas como

0-imagexxxx.jpg 1-imageyyyy.png 2-imagezzzz.jpg etc

Groovy util

Para mejorar la legibilidad del pipeline (y separar la "lógica de negocio" del pipeline) vamos a crear una clase Groovy con 2 métodos estáticos:

Uno servirá para extraer la posición donde fue tomada la foto

static def extractCoord( path ){
    def metadata = Imaging.getMetadata(path.toFile())
    if( !metadata || !metadata.metaClass.getMetaMethod("getExif") )
        return null

    def latitude = metadata.exif?.GPS?.latitudeAsDegreesNorth
    def longitude = metadata.exif?.GPS?.longitudeAsDegreesEast

    [latitude, longitude]
}

El otro método servirá para ordenar dos posiciones geográficas:

static double metersTo( a, b) {
    double lat1 = a[0] as double
    double lng1 = a[1] as double
    double lat2 = b[0] as double
    double lng2 = b[1] as double
    double radioTierra = 6371;
    double dLat = Math.toRadians(lat2 - lat1);
    double dLng = Math.toRadians(lng2 - lng1);
    double sindLat = Math.sin(dLat / 2);
    double sindLng = Math.sin(dLng / 2);
    double va1 = Math.pow(sindLat, 2) + Math.pow(sindLng, 2) * Math.cos(Math.toRadians(lat1)) * Math.cos(Math.toRadians(lat2));
    double va2 = 2 * Math.atan2(Math.sqrt(va1), Math.sqrt(1 - va1));
    double meters = radioTierra * va2;
    meters;
}

Image

Para mejorar la legibilidad del código (y no estar usando mapas genéricos donde guardar la información) vamos a crear una clase Image

class Image{
    def file
    def coord
    def order
}

Nos servirá para guardar referencia a la imagen original, las coordenadas y su posicion en la lista

Process

El pipeline se va a componer de dos procesos:

Uno para iterar sobre las fotos y extraer su posicion

process EXIF_GPS{
    input:
        path fImage
    output:
        val image
    exec:
        coord = Images.extractCoord( file("$params.directory/$fImage") )
        image = new Image(file:fImage, coord: coord)
}

y otro para copiar la imagen original al destino, usando la posicion como nombre

process PROCESS_IMAGE{
    input:
        each img
    output:
        val target
    exec:
        target = file("$params.directory/$img.file").copyTo(file("$params.outputDir/$img.order-$img.file"))
}

Pipeline

Por último el pipeline a ejecutar:

• Para todas las fotos que existan en un directorio

• Extraeremos la posicion

• Filtraremos las que tengan información de la posición

• Las ordenaremos segun el punto de origin

• Copiaremos a directorio de salida
  

workflow{
    def images = Channel.fromPath("$params.directory/*.jpg")

    images //read all images
        | EXIF_GPS // extract gpf information
        | branch { // diferenciate if exif information or not
            no_info: !it.coord?[0]
            with_info: it.coord
        }
        | set{ gps_images } // send to new channel

    gps_images.with_info // only images with gps info
        | toSortedList{ a, b-> // sort respect how far are from origin
            Images.metersTo(origin,a.coord) <=> Images.metersTo(origin, b.coord)
        }
        | map { // assign the index
            it.eachWithIndex{ img, idx-> img.order = idx}
        }
        | set{ images_sorted } //send to new channel

    images_sorted
        | PROCESS_IMAGE
        | view

}

Este workflow tiene algunas cosas interesantes como:

branch que nos permite crear un canal con canales "hijos" según el criterio que queramos

set que nos permite crear canales "al vuelo". En este caso creamos un canal images_sortedalimentándolo con un ArrayList de images y así poder ejecutar en pararelo cada imagen

Ejecutando

Si tienes instalado nextflow y tienes un directorio con imágenes puedes ejecutar

nextflow run -r main https://github.com/jagedn/nextflow-images.git --directory "/MI/DIRECTORIO/ORIGEN" --outputDir "/MI/DIRECTORIO/SALIDA"

Como ves Nextflow es capaz de descargar desde un repositorio git un pipeline completo e incluso de poder especificarle qué rama del mismo queremos ejecutar (main en mi caso)

Si quieres ordenar las fotos por algún punto que no sea [0,0] añade al final del comando --origin "20.23,12.13123" o las coordenadas que quieras

Conclusión

Probablemente no sea un pipeline muy útil pero me ha servido para practicar un poco el cómo encadenar canales y procesos