How to use the pystac.CatalogType function in pystac

To help you get started, we’ve selected a few pystac examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github azavea / pystac / tests / test_writing.py View on Github external
def validate_catalog_link_type(href, link_type, should_include_self):
            cat_dict = STAC_IO.read_json(href)
            cat = STACObject.from_file(href)
            for link in cat.get_links():
                if not link.rel == 'self':
                    self.assertEqual(link.link_type, link_type)

            rels = set([l['rel'] for l in cat_dict['links']])
            self.assertEqual('self' in rels, should_include_self)

            for child_link in cat.get_child_links():
                child_href = make_absolute_href(child_link.target, href)
                validate_catalog_link_type(child_href, link_type,
                                           catalog_type == CatalogType.ABSOLUTE_PUBLISHED)

            for item_link in cat.get_item_links():
                item_href = make_absolute_href(item_link.target, href)
                validate_item_link_type(item_href, link_type,
                                        catalog_type == CatalogType.ABSOLUTE_PUBLISHED)
github azavea / pystac / tests / extensions / test_label.py View on Github external
def test_validate_label(self):
        with open(self.label_example_1_uri) as f:
            label_example_1_dict = json.load(f)
        self.validator.validate_dict(label_example_1_dict, "ITEM")

        with TemporaryDirectory() as tmp_dir:
            cat_dir = os.path.join(tmp_dir, 'catalog')
            catalog = TestCases.test_case_1()
            catalog.normalize_and_save(cat_dir, catalog_type=CatalogType.SELF_CONTAINED)

            cat_read = Catalog.from_file(os.path.join(cat_dir, 'catalog.json'))
            label_item_read = cat_read.get_item("area-2-2-labels", recursive=True)
            self.validator.validate_object(label_item_read)
github azavea / pystac / tests / test_collection.py View on Github external
'eo:gsd': 0.3,
            'eo:platform': 'Maxar',
            'eo:instrument': 'WorldView3'
        }

        collection = Collection(id='test',
                                description='test',
                                extent=collection_extent,
                                properties=common_properties,
                                license='CC-BY-SA-4.0')

        collection.add_items([item1, item2])

        with TemporaryDirectory() as tmp_dir:
            collection.normalize_hrefs(tmp_dir)
            collection.save(catalog_type=CatalogType.SELF_CONTAINED)

            read_col = Collection.from_file('{}/collection.json'.format(tmp_dir))
            items = list(read_col.get_all_items())

            self.assertEqual(len(items), 2)
            self.assertIsInstance(items[0], EOItem)
            self.assertIsInstance(items[1], EOItem)
github azavea / pystac / tests / test_writing.py View on Github external
for link in cat.get_links():
                if not link.rel == 'self':
                    self.assertEqual(link.link_type, link_type)

            rels = set([l['rel'] for l in cat_dict['links']])
            self.assertEqual('self' in rels, should_include_self)

            for child_link in cat.get_child_links():
                child_href = make_absolute_href(child_link.target, href)
                validate_catalog_link_type(child_href, link_type,
                                           catalog_type == CatalogType.ABSOLUTE_PUBLISHED)

            for item_link in cat.get_item_links():
                item_href = make_absolute_href(item_link.target, href)
                validate_item_link_type(item_href, link_type,
                                        catalog_type == CatalogType.ABSOLUTE_PUBLISHED)
github azavea / pystac / tests / test_catalog.py View on Github external
def asset_mapper(key, asset):
            if 'geotiff' in asset.media_type:
                asset.title = 'NEW TITLE'
                changed_assets.append(key)
                return ('{}-modified'.format(key), asset)
            else:
                return asset

        with TemporaryDirectory() as tmp_dir:
            catalog = TestCases.test_case_2()

            new_cat = catalog.map_assets(asset_mapper)

            new_cat.normalize_hrefs(os.path.join(tmp_dir, 'cat'))
            new_cat.save(catalog_type=CatalogType.ABSOLUTE_PUBLISHED)

            result_cat = Catalog.from_file(os.path.join(tmp_dir, 'cat', 'catalog.json'))

            found = False
            not_found = False
            for item in result_cat.get_all_items():
                for key, asset in item.assets.items():
                    if key.replace('-modified', '') in changed_assets:
                        found = True
                        self.assertEqual(asset.title, 'NEW TITLE')
                    else:
                        not_found = True
                        self.assertNotEqual(asset.title, 'NEW TITLE')

            self.assertTrue(found)
            self.assertTrue(not_found)
github azavea / pystac / tests / test_catalog.py View on Github external
def test_create_and_read(self):
        with TemporaryDirectory() as tmp_dir:
            cat_dir = os.path.join(tmp_dir, 'catalog')
            catalog = TestCases.test_case_1()

            catalog.normalize_and_save(cat_dir, catalog_type=CatalogType.ABSOLUTE_PUBLISHED)

            read_catalog = Catalog.from_file('{}/catalog.json'.format(cat_dir))

            collections = catalog.get_children()
            self.assertEqual(len(list(collections)), 2)

            items = read_catalog.get_all_items()

            self.assertEqual(len(list(items)), 8)
github azavea / pystac / tests / test_catalog.py View on Github external
def test_full_copy_4(self):
        with TemporaryDirectory() as tmp_dir:
            root_cat = TestCases.test_case_2()
            root_cat.normalize_hrefs(os.path.join(tmp_dir, 'catalog-full-copy-4-source'))
            root_cat.save(catalog_type=CatalogType.ABSOLUTE_PUBLISHED)
            cat2 = root_cat.full_copy()
            cat2.normalize_hrefs(os.path.join(tmp_dir, 'catalog-full-copy-4-dest'))
            cat2.save(catalog_type=CatalogType.ABSOLUTE_PUBLISHED)

            self.check_catalog(root_cat, 'source')
            self.check_catalog(cat2, 'dest')

            # Check that the relative asset link was saved correctly in the copy.
            item = cat2.get_item('cf73ec1a-d790-4b59-b077-e101738571ed', recursive=True)

            href = item.assets['cf73ec1a-d790-4b59-b077-e101738571ed'].get_absolute_href()
            self.assertTrue(os.path.exists(href))