Merge pull request stac-utils#19 from azavea/sk/itemcollections

add itemcollections

Author: lossyrob

pystac/__init__.py

@@ -9,6 +9,8 @@ class STACError(Exception):
 from pystac.catalog import (Catalog, CatalogType)
 from pystac.collection import (Collection, Extent, SpatialExtent, TemporalExtent, Provider)
 from pystac.item import (Item, Asset)
+from pystac.item_collection import ItemCollection
+from pystac.single_file import SingleFileSTAC
 from pystac.eo import *
 from pystac.label import *
 

pystac/item_collection.py

@@ -0,0 +1,60 @@
+import json
+import os
+
+from pystac.io import STAC_IO
+from pystac.item import Item
+from pystac.link import Link, LinkType
+
+
+class ItemCollection:
+    DEFAULT_FILE_NAME = "item_collection.json"
+
+    def __init__(self, features, links=[], type='FeatureCollection'):
+        self.type = type
+        self.features = features
+        self.links = links
+
+        if [l for l in self.links if l.rel == 'self'] == []:
+            self.links.append(Link('self', './{}'.format(self.DEFAULT_FILE_NAME), link_type=LinkType.RELATIVE))
+
+    def __repr__(self):    
+        return '<ItemCollection item ids={}>'.format([f.id for f in self.features])
+    
+    @staticmethod
+    def from_dict(d):
+        features = [Item.from_dict(feature) for feature in d['features']]
+        links = []
+        if 'links' in d.keys():
+            links = [Link.from_dict(link) for link in d['links']]
+        return ItemCollection(features, links, d['type'])
+
+    @staticmethod
+    def from_file(uri):
+        d = json.loads(STAC_IO.read_text(uri))
+        c = ItemCollection.from_dict(d)
+        return c
+    
+    def to_dict(self, include_self_link=False):
+        links = self.links
+        if not include_self_link:
+            links = filter(lambda l: l.rel != 'self', links)
+
+        d = {
+            'type': self.type,
+            'features': [f.to_dict() for f in self.features],
+            'links': [l.to_dict() for l in links]
+        }
+
+        return d
+    
+    def get_self_href(self):
+        self_link = next((l for l in self.links if l.rel == 'self'), None)
+        if self_link:
+            return self_link.target
+        return self_link
+    
+    def get_items(self):
+        return self.features
+
+    def save(self, include_self_link=True):
+        STAC_IO.save_json(self.get_self_href(), self.to_dict(include_self_link = include_self_link))

pystac/single_file.py

@@ -0,0 +1,61 @@
+import json
+import os
+
+from pystac.collection import Collection
+from pystac.io import STAC_IO
+from pystac.item import Item
+from pystac.item_collection import ItemCollection
+
+
+class SingleFileSTAC(ItemCollection):
+    def __init__(self, type, features, collections, search=None):
+        self.type = type
+        self.features = features
+        self.collections = collections
+        self.search = search
+    
+    def __repr__(self):
+        return '<SingleFile collection ids={}>'.format([c.id for c in self.collections])
+
+    @staticmethod
+    def from_file(uri):
+        d = json.loads(STAC_IO.read_text(uri))
+        c = SingleFileSTAC.from_dict(d)
+        return c
+
+    @staticmethod
+    def from_dict(d):
+        type = d['type']
+        features = [Item.from_dict(feature) for feature in d['features']]
+        collections = [Collection.from_dict(c) for c in d['collections']]
+        search_obj = None
+        if 'search' in d.keys():
+            sd = d['search']
+            search_obj = Search(sd['endpoint'], sd['parameters'])
+        return SingleFileSTAC(type, features, collections, search_obj)
+
+    def to_dict(self):
+        d = {}
+        
+        d['type'] = self.type
+        d['features'] = [f.to_dict() for f in self.features]
+        d['collections'] = [c.to_dict() for c in self.collections]
+        if self.search:
+            d['search'] = self.search.to_dict()
+
+        return d
+    
+    def save(self, uri):
+        STAC_IO.save_json(os.path.abspath(uri), self.to_dict())
+
+class Search:
+    def __init__(self, endpoint=None, parameters=None):
+        self.endpoint = endpoint
+        self.parameters = parameters
+    
+    @staticmethod
+    def from_dict(d):
+        return Search(d['endpoint'], d['parameters'])
+
+    def to_dict(self):
+        return {'endpoint': self.endpoint, 'parameters': self.parameters}