Merge pull request #342 from ASFHyP3/opera-rtc

Add OPERA RTC as job type

Author: williamh890

CHANGELOG.md

@@ -6,6 +6,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [PEP 440](https://www.python.org/dev/peps/pep-0440/)
 and uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [7.4.0]
+
+### Added
+* Added `HyP3.submit_opera_rtc_s1_job` and `HyP3.prepare_opera_rtc_s1_job` methods for submitting OPERA RTC-S1 jobs to HyP3. This job type is under development and is currently only available in `https://hyp3-test-api.asf.alaska.edu`
+* Added `hyp3_sdk.exceptions.ServiceUnavailableError` for handling 503 errors from HyP3.
+
 ## [7.3.0]
 
 ### Added

README.md

@@ -23,8 +23,8 @@ A python wrapper around the HyP3 API
 ## Install
 
 In order to easily manage dependencies, we recommend using dedicated project environments
-via [Anaconda/Miniconda](https://docs.conda.io/projects/conda/en/latest/user-guide/install/index.html) 
-or [Python virtual environments](https://docs.python.org/3/tutorial/venv.html). 
+via [Anaconda/Miniconda](https://docs.conda.io/projects/conda/en/latest/user-guide/install/index.html)
+or [Python virtual environments](https://docs.python.org/3/tutorial/venv.html).
 
 The HyP3 SDK can be installed into a conda environment with
 
@@ -68,6 +68,7 @@ insar_burst_job = hyp3.submit_insar_isce_burst_job('reference_granule_id', 'seco
 insar_multi_burst_job = hyp3.submit_insar_isce_multi_burst_job(['ref_id_1', 'ref_id_2'], ['sec_id_1', 'sec_id_2'], 'job_name')
 autorift_job = hyp3.submit_autorift_job('reference_granule_id', 'secondary_granule_id', 'job_name')
 aria_s1_gunw_job = hyp3.submit_aria_s1_gunw_job(['ref_id_1', 'ref_id_2'], ['sec_id_1', 'sec_id_2'], 'frame_id', 'job_name')
+opera_rtc_s1_job = hyp3.submit_opera_rtc_s1_job('granule_id')
 ```
 Each of these functions will return an instance of the `Job` class that represents a new HyP3 job request.
 
@@ -76,7 +77,7 @@ To find HyP3 jobs that were run previously, you can use the `hyp3.find_jobs()`
 ```python
 batch = hyp3.find_jobs()
 ```
-This will return a `Batch` instance representing all jobs owned by you. You can also pass parameters to 
+This will return a `Batch` instance representing all jobs owned by you. You can also pass parameters to
 query to a specific set of jobs
 
 

src/hyp3_sdk/exceptions.py

@@ -24,12 +24,19 @@ class AuthenticationError(HyP3SDKError):
     """Raise when authentication does not succeed"""
 
 
+class ServiceUnavailableError(HyP3SDKError):
+    """Raise when the HyP3 API is unavailable"""
+
+
 def _raise_for_hyp3_status(response: Response):
     try:
         response.raise_for_status()
     except HTTPError:
         if 400 <= response.status_code < 500:
             raise HyP3Error(f'{response} {response.json()["detail"]}')
+        if response.status_code == 503:
+            raise ServiceUnavailableError(f'{response} {response.json()["detail"]}')
+
         raise ServerError
 
 

src/hyp3_sdk/hyp3.py

@@ -607,6 +607,44 @@ def prepare_aria_s1_gunw_job(
             job_dict['name'] = name
         return job_dict
 
+    def submit_opera_rtc_s1_job(self, granule: str, name: str | None = None) -> Batch:
+        """Submit an OPERA RTC-S1 job.
+
+        Args:
+            granule: The name of the S1 burst to use
+            name: A name for the job (optional)
+
+        Returns:
+            A Batch object containing the OPERA RTC-S1 job
+        """
+        arguments = locals().copy()
+        arguments.pop('self')
+        job_dict = self.prepare_opera_rtc_s1_job(**arguments)
+        return self.submit_prepared_jobs(prepared_jobs=job_dict)
+
+    @classmethod
+    def prepare_opera_rtc_s1_job(cls, granule: str, name: str | None = None) -> dict:
+        """Prepare an OPERA RTC-S1 job.
+
+        Args:
+            granule: The name of the S1 burst to use
+            name: A name for the job
+
+        Returns:
+            A dictionary containing the prepared OPERA RTC-S1 job
+        """
+        job_parameters = locals().copy()
+        for key in ['cls', 'name', 'granule']:
+            job_parameters.pop(key)
+
+        job_dict = {
+            'job_parameters': {'granules': [granule], **job_parameters},
+            'job_type': 'OPERA_RTC_S1',
+        }
+        if name is not None:
+            job_dict['name'] = name
+        return job_dict
+
     def my_info(self) -> dict:
         """Returns:
         Your user information

tests/test_exceptions.py

@@ -24,6 +24,15 @@ def test_raise_for_hyp3_status():
     exceptions._raise_for_hyp3_status(response)
 
 
+def test_service_unavailable_error():
+    response = Response()
+    response.status_code = 503
+    response._content = b'{ "detail" : "foo" }'
+    with pytest.raises(exceptions.ServiceUnavailableError) as e:
+        exceptions._raise_for_hyp3_status(response)
+    assert 'foo' in str(e)
+
+
 def test_raise_for_search_status():
     response = Response()
     response.status_code = 400

tests/test_hyp3.py

@@ -347,6 +347,22 @@ def test_prepare_insar_isce_multi_burst_job():
     }
 
 
+def test_prepare_opera_rtc_s1_job():
+    assert HyP3.prepare_opera_rtc_s1_job(granule='g1') == {
+        'job_type': 'OPERA_RTC_S1',
+        'job_parameters': {
+            'granules': ['g1'],
+        },
+    }
+    assert HyP3.prepare_opera_rtc_s1_job(granule='g2', name='my_name') == {
+        'job_type': 'OPERA_RTC_S1',
+        'name': 'my_name',
+        'job_parameters': {
+            'granules': ['g2'],
+        },
+    }
+
+
 def test_prepare_aria_s1_gunw_job():
     assert HyP3.prepare_aria_s1_gunw_job(
         reference=['ref_granule1', 'ref_granule2'], secondary=['sec_granule1', 'sec_granule2'], frame_id=100
@@ -448,6 +464,16 @@ def test_submit_aria_s1_gunw_job(get_mock_hyp3, get_mock_job):
     assert batch.jobs[0] == job
 
 
+@responses.activate
+def test_submit_opera_rtc_s1_job(get_mock_hyp3, get_mock_job):
+    job = get_mock_job('OPERA_RTC_S1', job_parameters={'granules': ['g1']})
+    api_response = {'jobs': [job.to_dict()]}
+    api = get_mock_hyp3()
+    responses.add(responses.POST, urljoin(api.url, '/jobs'), json=api_response)
+    batch = api.submit_opera_rtc_s1_job(['g1'])
+    assert batch.jobs[0] == job
+
+
 @responses.activate
 def test_resubmit_previous_job(get_mock_hyp3, get_mock_job):
     job = get_mock_job()