Generate an ATOM entry for the catalog and perform its ingestion

In this scenario, we generate an ATOM entry containing the metadata of a product and feed it into the catalog (we use a sample product downloaded from a public repository).

1. Set the necessary variables

The following section defines all the necessary information as variables so the code below can be easily reused.

[ ]:
import getpass

# Set the credentials (Ellip username and API key)
username = raw_input("What is your Ellip username? ")
api_key = getpass.getpass("What is your Ellip API key? ")

# Set the name of the destination index on the Terradue catalog
index_name = raw_input("What is the destination index name? (press Enter to confirm default [{0}]) ".format(username))

if not index_name:
    index_name = username

# Set the catalog endpoint URL
endpoint = "{0}".format(index_name)

2. Define a function to generate an EarthObservation extension element

[ ]:
import lxml.etree as etree
import numpy as np
from shapely.wkt import loads

def eop_metadata(metadata):

    # Define namespace URIs
    opt = ''
    om  = ''
    gml = ''
    eop = ''
    sar = ''

    # Define the element structure
    # There are several levels for much of the content and many elements are only containers for other elements;
    # elements that hold actual values are marked as 'content element' below.
    root = etree.Element('{%s}EarthObservation' % opt)

    phenomenon_time = etree.SubElement(root, '{%s}phenomenonTime' % om)
    time_period = etree.SubElement(phenomenon_time, '{%s}TimePeriod' % gml)
    # Content element:
    begin_position = etree.SubElement(time_period, '{%s}beginPosition'  % gml)
    # Content element:
    end_position = etree.SubElement(time_period, '{%s}endPosition'  % gml)

    procedure = etree.SubElement(root, '{%s}procedure' % om)
    earth_observation_equipment = etree.SubElement(procedure, '{%s}EarthObservationEquipment' % eop)
    acquisition = etree.SubElement(earth_observation_equipment, '{%s}acquisitionParameters' % eop)
    # Content element:
    orbit_number = etree.SubElement(acquisition, '{%s}orbitNumber' % eop)
    # Content element:
    wrs_longitude_grid = etree.SubElement(acquisition, '{%s}wrsLongitudeGrid' % eop)
    # Content element:
    orbit_direction = etree.SubElement(acquisition, '{%s}orbitDirection' % eop)

    feature_of_interest = etree.SubElement(root, '{%s}featureOfInterest' % om)
    footprint = etree.SubElement(feature_of_interest, '{%s}Footprint' % eop)
    multi_extentOf = etree.SubElement(footprint, '{%s}multiExtentOf' % eop)
    multi_surface = etree.SubElement(multi_extentOf, '{%s}MultiSurface' % gml)
    surface_members = etree.SubElement(multi_surface, '{%s}surfaceMembers' % gml)
    polygon = etree.SubElement(surface_members, '{%s}Polygon' % gml)
    exterior = etree.SubElement(polygon, '{%s}exterior' % gml)
    linear_ring = etree.SubElement(exterior, '{%s}LinearRing' % gml)
    # Content element:
    poslist = etree.SubElement(linear_ring, '{%s}posList' % gml)

    result = etree.SubElement(root, '{%s}result' % om)
    earth_observation_result = etree.SubElement(result, '{%s}EarthObservationResult' % opt)
    # Content element:
    cloud_cover_percentage = etree.SubElement(earth_observation_result, '{%s}cloudCoverPercentage' % opt)

    metadata_property = etree.SubElement(root, '{%s}metaDataProperty' % eop)
    earth_observation_metadata = etree.SubElement(metadata_property, '{%s}EarthObservationMetaData' % eop)
    # Content element:
    identifier = etree.SubElement(earth_observation_metadata, '{%s}identifier' % eop)
    # Content element:
    product_type = etree.SubElement(earth_observation_metadata, '{%s}productType' % eop)

    vendor_specific = etree.SubElement(earth_observation_metadata, '{%s}vendorSpecific' % eop)
    specific_information = etree.SubElement(vendor_specific, '{%s}SpecificInformation' % eop)
    # Content element:
    local_attribute = etree.SubElement(specific_information, '{%s}localAttribute' % eop)
    # Content element:
    local_value = etree.SubElement(specific_information, '{%s}localValue' % eop)

    # Set values for content elements
    begin_position.text = metadata['startdate']
    end_position.text = metadata['enddate']
    orbit_number.text = metadata['orbitNumber']
    wrs_longitude_grid.text = metadata['wrsLongitudeGrid']
    orbit_direction.text = metadata['orbitDirection']

    coords = np.asarray([t[::-1] for t in list(loads(metadata['wkt']).exterior.coords)]).tolist()
    pos_list = ''
    for elem in coords:
        pos_list += ' '.join(str(e) for e in elem) + ' '

    poslist.attrib['count'] = str(len(coords))
    poslist.text = pos_list

    cloud_cover_percentage.text = metadata['cc']

    identifier.text = metadata['identifier']
    product_type.text = metadata['productType']

    local_attribute.text = 'MY_ATTRIBUTE'
    local_value.text = metadata['MY_ATTRIBUTE']

    return root

3. Define a class for ATOM manipulation

This class allows us to add basic elements such as an identifier, a title, an enclosure link and a product date; and to append the EarthObservation extension created above.

[ ]:
import lxml.etree as etree
import sys
import os
import string
import hashlib
import urllib2
import base64
import time

class Atom:
    tree = None
    root = None
    entry = None

    def __init__(self, root):
        self.root = root
        self.tree = root
        self.links = self.root.xpath('/a:feed/a:entry/a:link', namespaces={'a':''})
        entries = self.root.xpath('/a:feed/a:entry', namespaces={'a':''})
        if len(entries) > 0:
            self.entry = entries[0]

    def from_template():
        template = """<?xml version="1.0"?>
<feed xmlns="">
    <title type="text"></title>
    <summary type="html"></summary>
    <link rel="enclosure" type="application/octet-stream" href=""/>
    <date xmlns=""></date>
    <identifier xmlns=""></identifier>
        tree = etree.fromstring(template)
        return Atom(tree)

    def load(url, username=None, api_key=None):
        """Load and return the atom file at the location url

        request = urllib2.Request(url)

        if ( username != None ):
            base64string = base64.b64encode('%s:%s' % (username, api_key))
            request.add_header("Authorization", "Basic %s" % base64string)
        fp = urllib2.urlopen(request)
        tree = etree.parse(fp)
        if ( tree.getroot().tag != "{}feed" ):
            raise ValueError('not an Atom feed')

        return Atom(tree)

    def set_identifier(self, identifier):
        """Set first atom entry identifier

        el_identifier = self.root.xpath('/a:feed/a:entry/d:identifier',

        el_identifier[0].text = identifier

    def get_identifier(self):
        el_identifier = self.root.xpath('/a:feed/a:entry/d:identifier',

        if (len(el_identifier) == 0):
            return None

        return el_identifier[0].text;

    def get_total_results(self, create=False):
        # get OS total results in feed
        totalResults = self.root.xpath('/a:feed/os:totalResults', namespaces={'a':'', 'os':''})

        if (len(totalResults) == 0):
            return None

        return int(totalResults[0].text)

    def get_title(self, create=False):
        # get or create title
        titles = self.root.xpath('/a:feed/a:entry/a:title', namespaces={'a':''})

        if (len(titles) == 0):
            if (create):
                titles = [etree.SubElement(self.entry, "{}title")]
                return titles[0]
            return None

        return titles[0]

    def set_title_text(self, text):
        """Set first atom entry title

        el_title = self.root.xpath('/a:feed/a:entry/a:title',

        el_title[0].text = text

    def get_summary(self, create=False):
        # get or create summary
        summaries = self.root.xpath('/a:feed/a:entry/a:summary', namespaces={'a':''})

        if (len(summaries) == 0):
            if (create):
                summaries = [etree.SubElement(self.entry, "{}summary")]
                return summaries[0]
            return None

        return summaries[0]

    def set_summary_text(self, text):
        # get or create summary
        summary = self.get_summary(True)

        summary.text = text

    def get_links(self, rel_type):
        # get links
        return self.root.xpath('/a:feed/a:entry/a:link[@rel = "{0}"]'.format(rel_type), namespaces={'a':''})

    def set_enclosure_link(self, href, title):

        el_enclosure_link = self.root.xpath('/a:feed/a:entry/a:link[@rel="enclosure" and (@href="" or @href="{0}")]'.format(href),

        if (len(el_enclosure_link) > 0):
            link = el_enclosure_link[0]
            link.attrib['href'] = href
            link = self.add_enclosure_link(href, title)

    def add_enclosure_link(self, href, title):

        xml_string = '<link rel="enclosure" type="application/octet-stream" title="%s" href="%s"/>' % (title, href.replace('&', '&amp;'))

        link = etree.fromstring(xml_string)

        return link

    def add_extension(self, xml_ext):

        el_entry = self.root.xpath('/a:feed/a:entry/a:link',


    def add_link(self, href, rel, title=None, type=None):

        link = etree.SubElement(self.root.xpath('/a:feed/a:entry',
                      namespaces={'a':''})[0], "{}link")

        link.attrib['href'] = href
        link.attrib['rel'] = rel
        if title:
            link.attrib['title'] = title
        if type:
            link.attrib['type'] = type

    def remove_link(self, rel, link_title=None, link_type=None, link_url=None):
        links = self.get_links(rel)
        filter = None
        value = None

        if link_title:
            filter = 'title'
            value = link_title
        elif link_type:
            filter = 'type'
            value = link_type
        elif link_url:
            filter = 'url'
            value = link_url
            raise Exception("Required parameter link_title, link_type or link_url")

        for link in links:
            if link.attrib[filter] == value:

    def get_offering_elements(self, offering_code):

        return self.root.xpath('/a:feed/a:entry/b:offering[@code="{0}"]'.format(offering_code),

    def get_operation_elements(offering_element, operation_code=None):

        xpath = 'b:operation'
        if (operation_code):
            xpath += '[@code="{0}"]'.format(operation_code)
        return offering_element.xpath(xpath, namespaces={'b':''})

    def add_offering(self, offering):

        self.root.xpath('/a:feed/a:entry', namespaces={'a':''})[0].append(offering)

    def add_offerings(self, offerings):

        for offering in offerings:

    def get_dctspatial(self, create=False):

        # get or create summary
        spatials = self.root.xpath('/a:feed/a:entry/c:spatial',

        if (len(spatials) == 0):
            if (create):
                spatials = [etree.SubElement(self.entry, "{}spatial")]
                return spatials[0]
            return None

        return spatials[0]

    def set_dctspatial(self, wkt):

        el_spatial = self.get_dctspatial(True)

        el_spatial.text = wkt

    def get_dcdate(self, create):

        # get or create dcdate
        el_dates = self.root.xpath('/a:feed/a:entry/d:date',

        if (len(el_dates) == 0):
            if (create):
                el_dates = [etree.SubElement(self.entry, "{}date")]
                return el_dates[0]
            return None

        return el_dates[0]

    def set_dcdate(self, date):

        # get or create dcdate
        dcdate = self.get_dcdate(True)

        dcdate.text = date

    def set_published(self, published):

        el_published = self.root.xpath('/a:feed/a:entry/a:published',
        el_published[0].text = published

    def get_category_by_scheme(self, scheme):

        categories = self.root.xpath('/a:feed/a:entry/a:category[@scheme="{0}"]'.format(scheme), namespaces={'a':''})
        if (len(categories) == 0):
            return None

        return categories[0]

    def get_categories(self, term, scheme=None):

        # get categories
        filter = '@term="{0}"'.format(term)
        if scheme != None:
            filter = '{0} and @scheme="{1}"'.format(filter, scheme)

        return self.root.xpath('/a:feed/a:entry/a:category[{0}]'.format(filter), namespaces={'a':''})

    def remove_category(self, term, scheme=None):

        # get and remove category
        for category in self.get_categories(term, scheme):

    def remove_category_by_scheme(self, scheme):

        # get categories
        filter = '@scheme="{0}"'.format(scheme)

        categories = self.root.xpath('/a:feed/a:entry/a:category[{0}]'.format(filter), namespaces={'a':''})
        for category in categories:

    def set_category(self, term, label=None, scheme=None):

        categories = self.get_categories(term, scheme)

        if (len(categories) == 0):
            categories = [etree.SubElement(self.entry, "{}category")]

        categories[0].attrib['term']  = term
        if label != None:
            categories[0].attrib['label']  = label
        if scheme != None:
            categories[0].attrib['scheme']  = scheme

    def set_generator(self, uri, version, text):

       # get or create generator
        el_generator = self.root.xpath('/a:feed/a:entry/a:generator', namespaces={'a':''})

        if (len(el_generator) == 0):
            el_generator = [etree.SubElement(self.root.xpath('/a:feed/a:entry',
                      namespaces={'a':''})[0], "{}generator")]

        el_generator[0].attrib['uri'] = uri
        el_generator[0].attrib['version'] = version
        el_generator[0].text = text

    def append_summary_html(self, text):
        """Append atom summary with text

        html_summary = self.get_summary(True).text
        html_summary += "<p>%s</p>" % text


    def to_string(self, pretty_print = True):

        return etree.tostring(self.tree, pretty_print=pretty_print)

    def clear_enclosures(self):

        links = self.get_links("enclosure")
        for link in links:

    def get_extensions(self, name, namespace):

        return self.root.xpath('/a:feed/a:entry/e:{0}'.format(name),

4. Build the EarthObservation extension element

We define a dictionary containing the metadata and use it as argument for the function defined above.

[ ]:
metadata = {'startdate':       '2019-01-02T03:04:05.678Z',
            'enddate':         '2019-01-02T03:05:06.789Z',
            'orbitNumber':     '99',
            'orbitDirection':  'DESCENDING',
            'wkt':             'POLYGON((10.1 10.2,20.3 10.4,20.5 20.6,10.7 20.8,10.1 10.2))',
            'cc':              '55',
            'identifier' :     'MY_PRODUCT',
            'productType':     'MY_TYPE',
            'MY_ATTRIBUTE':    'MY_VALUE'

# Build the element
eo = eop_metadata(metadata)

Show the EarthObservation element just created:

[ ]:
print(etree.tostring(eo, pretty_print=True))

5. Build an ATOM feed

We create an ATOM feed with one entry to which we append the extension created above.

[ ]:
import datetime

atom = Atom.from_template()
atom.set_title_text("Title for MY_PRODUCT")
atom.set_summary_text("This is the summary for MY_PRODUCT")
atom.set_dcdate("{0}/{1}".format(metadata['startdate'], metadata['enddate']))


atom.set_enclosure_link("", "Location on storage")

Show the resulting ATOM feed:

[ ]:
print(etree.tostring(atom.root, pretty_print=True))

6. Post the ATOM feed

We post the ATOM feed to an index on the catalog (variables are defined in the first step).

[ ]:
import requests

request =,
                        headers={"Content-Type": "application/atom+xml", "Accept": "application/xml"},
                        auth=(username, api_key),

if request.status_code == 200:
    print('Data item updated at {0}/search?uid={1}&apikey={2} ({3})'.format(endpoint, atom.get_identifier(), api_key, str(request.status_code)))
    print('Data item NOT updated at {0}/search?uid={1}&apikey={2} ({3})'.format(endpoint, atom.get_identifier(), api_key, str(request.status_code)))

If a product URL with status 200 is displayed, the ATOM feed has been successfully uploaded and the product information is available on the Terradue catalog.