-
Notifications
You must be signed in to change notification settings - Fork 120
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #236 from schwehr/sat-extension
Implement SAT extension.
- Loading branch information
Showing
4 changed files
with
260 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,114 @@ | ||
"""Implement the Satellite (SAT) Extension. | ||
https://github.com/radiantearth/stac-spec/tree/dev/extensions/sat | ||
""" | ||
|
||
import enum | ||
from typing import List, Optional | ||
|
||
import pystac | ||
from pystac import Extensions | ||
from pystac import item | ||
from pystac.extensions import base | ||
|
||
ORBIT_STATE: str = 'sat:orbit_state' | ||
RELATIVE_ORBIT: str = 'sat:relative_orbit' | ||
|
||
|
||
class OrbitState(enum.Enum): | ||
ASCENDING: str = 'ascending' | ||
DESCENDING: str = 'descending' | ||
GEOSTATIONARY: str = 'geostationary' | ||
|
||
|
||
class SatItemExt(base.ItemExtension): | ||
"""SatItemExt extends Item to add sat properties to a STAC Item. | ||
Args: | ||
item (Item): The item to be extended. | ||
Attributes: | ||
item (Item): The item that is being extended. | ||
Note: | ||
Using SatItemExt to directly wrap an item will add the 'sat' | ||
extension ID to the item's stac_extensions. | ||
""" | ||
item: pystac.Item | ||
|
||
def __init__(self, an_item: item.Item) -> None: | ||
self.item = an_item | ||
|
||
def apply(self, orbit_state: Optional[OrbitState] = None, relative_orbit: Optional[str] = None): | ||
"""Applies ext extension properties to the extended Item. | ||
Must specify at least one of orbit_state or relative_orbit. | ||
Args: | ||
orbit_state (OrbitState): Optional state of the orbit. Either ascending or descending | ||
for polar orbiting satellites, or geostationary for geosynchronous satellites. | ||
relative_orbit (int): Optional non-negative integer of the orbit number at the time | ||
of acquisition. | ||
""" | ||
if orbit_state is None and relative_orbit is None: | ||
raise pystac.STACError('Must provide at least one of: orbit_state or relative_orbit') | ||
if orbit_state: | ||
self.orbit_state = orbit_state | ||
if relative_orbit: | ||
self.relative_orbit = relative_orbit | ||
|
||
@classmethod | ||
def from_item(cls, an_item: item.Item): | ||
return cls(an_item) | ||
|
||
@classmethod | ||
def _object_links(cls) -> List: | ||
return [] | ||
|
||
@property | ||
def orbit_state(self) -> Optional[OrbitState]: | ||
"""Get or sets an orbit state of the item. | ||
Returns: | ||
OrbitState or None | ||
""" | ||
if ORBIT_STATE not in self.item.properties: | ||
return | ||
return OrbitState(self.item.properties.get(ORBIT_STATE)) | ||
|
||
@orbit_state.setter | ||
def orbit_state(self, v: Optional[OrbitState]) -> None: | ||
if v is None: | ||
if self.relative_orbit is None: | ||
raise pystac.STACError('Must set relative_orbit before clearing orbit_state') | ||
if ORBIT_STATE in self.item.properties: | ||
del self.item.properties[ORBIT_STATE] | ||
else: | ||
self.item.properties[ORBIT_STATE] = v.value | ||
|
||
@property | ||
def relative_orbit(self) -> int: | ||
"""Get or sets a relative orbit number of the item. | ||
Returns: | ||
int or None | ||
""" | ||
return self.item.properties.get(RELATIVE_ORBIT) | ||
|
||
@relative_orbit.setter | ||
def relative_orbit(self, v: int) -> None: | ||
if v is None and self.orbit_state is None: | ||
raise pystac.STACError('Must set orbit_state before clearing relative_orbit') | ||
if v is None: | ||
if RELATIVE_ORBIT in self.item.properties: | ||
del self.item.properties[RELATIVE_ORBIT] | ||
return | ||
if v < 0: | ||
raise pystac.STACError(f'relative_orbit must be >= 0. Found {v}.') | ||
|
||
self.item.properties[RELATIVE_ORBIT] = v | ||
|
||
|
||
SAT_EXTENSION_DEFINITION = base.ExtensionDefinition(Extensions.SAT, [ | ||
base.ExtendedObject(pystac.Item, SatItemExt), | ||
]) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,131 @@ | ||
"""Tests for pystac.extensions.sat.""" | ||
|
||
import datetime | ||
import unittest | ||
|
||
import pystac | ||
from pystac.extensions import sat | ||
|
||
|
||
def make_item() -> pystac.Item: | ||
"""Create basic test items that are only slightly different.""" | ||
asset_id = 'an/asset' | ||
start = datetime.datetime(2018, 1, 2) | ||
item = pystac.Item(id=asset_id, geometry=None, bbox=None, datetime=start, properties={}) | ||
|
||
item.ext.enable(pystac.Extensions.SAT) | ||
return item | ||
|
||
|
||
class SatTest(unittest.TestCase): | ||
def setUp(self): | ||
super().setUp() | ||
self.item = make_item() | ||
|
||
def test_stac_extensions(self): | ||
self.assertEqual([pystac.Extensions.SAT], self.item.stac_extensions) | ||
|
||
def test_no_args_fails(self): | ||
with self.assertRaises(pystac.STACError): | ||
self.item.ext.sat.apply() | ||
|
||
def test_orbit_state(self): | ||
orbit_state = sat.OrbitState.ASCENDING | ||
self.item.ext.sat.apply(orbit_state) | ||
self.assertEqual(orbit_state, self.item.ext.sat.orbit_state) | ||
self.assertNotIn(sat.RELATIVE_ORBIT, self.item.properties) | ||
self.assertFalse(self.item.ext.sat.relative_orbit) | ||
self.item.validate() | ||
|
||
def test_relative_orbit(self): | ||
relative_orbit = 1234 | ||
self.item.ext.sat.apply(None, relative_orbit) | ||
self.assertEqual(relative_orbit, self.item.ext.sat.relative_orbit) | ||
self.assertNotIn(sat.ORBIT_STATE, self.item.properties) | ||
self.assertFalse(self.item.ext.sat.orbit_state) | ||
self.item.validate() | ||
|
||
def test_relative_orbit_no_negative(self): | ||
negative_relative_orbit = -2 | ||
with self.assertRaises(pystac.STACError): | ||
self.item.ext.sat.apply(None, negative_relative_orbit) | ||
|
||
self.item.ext.sat.apply(None, 123) | ||
with self.assertRaises(pystac.STACError): | ||
self.item.ext.sat.relative_orbit = negative_relative_orbit | ||
|
||
def test_both(self): | ||
orbit_state = sat.OrbitState.DESCENDING | ||
relative_orbit = 4321 | ||
self.item.ext.sat.apply(orbit_state, relative_orbit) | ||
self.assertEqual(orbit_state, self.item.ext.sat.orbit_state) | ||
self.assertEqual(relative_orbit, self.item.ext.sat.relative_orbit) | ||
self.item.validate() | ||
|
||
def test_modify(self): | ||
self.item.ext.sat.apply(sat.OrbitState.DESCENDING, 999) | ||
|
||
orbit_state = sat.OrbitState.GEOSTATIONARY | ||
self.item.ext.sat.orbit_state = orbit_state | ||
relative_orbit = 1000 | ||
self.item.ext.sat.relative_orbit = relative_orbit | ||
self.assertEqual(orbit_state, self.item.ext.sat.orbit_state) | ||
self.assertEqual(relative_orbit, self.item.ext.sat.relative_orbit) | ||
self.item.validate() | ||
|
||
def test_from_dict(self): | ||
orbit_state = sat.OrbitState.GEOSTATIONARY | ||
relative_orbit = 1001 | ||
d = { | ||
'type': 'Feature', | ||
'stac_version': '1.0.0-beta.2', | ||
'id': 'an/asset', | ||
'properties': { | ||
'sat:orbit_state': orbit_state.value, | ||
'sat:relative_orbit': relative_orbit, | ||
'datetime': '2018-01-02T00:00:00Z' | ||
}, | ||
'geometry': None, | ||
'links': [], | ||
'assets': {}, | ||
'stac_extensions': ['sat'] | ||
} | ||
item = pystac.Item.from_dict(d) | ||
self.assertEqual(orbit_state, item.ext.sat.orbit_state) | ||
self.assertEqual(relative_orbit, item.ext.sat.relative_orbit) | ||
|
||
def test_to_from_dict(self): | ||
orbit_state = sat.OrbitState.GEOSTATIONARY | ||
relative_orbit = 1002 | ||
self.item.ext.sat.apply(orbit_state, relative_orbit) | ||
d = self.item.to_dict() | ||
self.assertEqual(orbit_state.value, d['properties'][sat.ORBIT_STATE]) | ||
self.assertEqual(relative_orbit, d['properties'][sat.RELATIVE_ORBIT]) | ||
|
||
item = pystac.Item.from_dict(d) | ||
self.assertEqual(orbit_state, item.ext.sat.orbit_state) | ||
self.assertEqual(relative_orbit, item.ext.sat.relative_orbit) | ||
|
||
def test_clear_orbit_state(self): | ||
self.item.ext.sat.apply(sat.OrbitState.DESCENDING, 999) | ||
|
||
self.item.ext.sat.orbit_state = None | ||
self.assertIsNone(self.item.ext.sat.orbit_state) | ||
self.item.validate() | ||
|
||
def test_clear_relative_orbit(self): | ||
self.item.ext.sat.apply(sat.OrbitState.DESCENDING, 999) | ||
|
||
self.item.ext.sat.relative_orbit = None | ||
self.assertIsNone(self.item.ext.sat.relative_orbit) | ||
self.item.validate() | ||
|
||
def test_clear_orbit_state_fail(self): | ||
self.item.ext.sat.apply(sat.OrbitState.DESCENDING) | ||
with self.assertRaises(pystac.STACError): | ||
self.item.ext.sat.orbit_state = None | ||
|
||
def test_clear_orbit_relative_orbit(self): | ||
self.item.ext.sat.apply(None, 1) | ||
with self.assertRaises(pystac.STACError): | ||
self.item.ext.sat.relative_orbit = None |