Merge pull request #119 from ASFHyP3/develop

Release v1.4.2

.github/workflows/changelog-check.yml

@@ -13,4 +13,4 @@ on:
 
 jobs:
   call-changelog-check-workflow:
-    uses: ASFHyP3/actions/.github/workflows/reusable-changelog-check.yml@v0.14.0
+    uses: ASFHyP3/actions/.github/workflows/reusable-changelog-check.yml@v0.15.0

.github/workflows/create-jira-issue.yml

@@ -6,7 +6,7 @@ on:
 
 jobs:
   call-create-jira-issue-workflow:
-    uses: ASFHyP3/actions/.github/workflows/reusable-create-jira-issue.yml@v0.14.0
+    uses: ASFHyP3/actions/.github/workflows/reusable-create-jira-issue.yml@v0.15.0
     secrets:
       JIRA_BASE_URL: ${{ secrets.JIRA_BASE_URL }}
       JIRA_USER_EMAIL: ${{ secrets.JIRA_USER_EMAIL }}

.github/workflows/labeled-pr-check.yml

@@ -12,4 +12,4 @@ on:
 
 jobs:
   call-labeled-pr-check-workflow:
-    uses: ASFHyP3/actions/.github/workflows/reusable-labeled-pr-check.yml@v0.14.0
+    uses: ASFHyP3/actions/.github/workflows/reusable-labeled-pr-check.yml@v0.15.0

.github/workflows/release-checklist-comment.yml

@@ -9,7 +9,7 @@ on:
 
 jobs:
   call-release-workflow:
-    uses: ASFHyP3/actions/.github/workflows/reusable-release-checklist-comment.yml@v0.14.0
+    uses: ASFHyP3/actions/.github/workflows/reusable-release-checklist-comment.yml@v0.15.0
     permissions:
       pull-requests: write
     secrets:

.github/workflows/release.yml

@@ -8,7 +8,7 @@ on:
 jobs:
   call-release-workflow:
     # Docs: https://github.com/ASFHyP3/actions
-    uses: ASFHyP3/actions/.github/workflows/reusable-release.yml@v0.14.0
+    uses: ASFHyP3/actions/.github/workflows/reusable-release.yml@v0.15.0
     with:
       release_prefix: burst2safe
     secrets:

.github/workflows/static-analysis.yml

@@ -1,16 +1,16 @@
 name: Static analysis
 
-on: [pull_request]
+on: push
 
 jobs:
   call-secrets-analysis-workflow:
     # Docs: https://github.com/ASFHyP3/actions
-    uses: ASFHyP3/actions/.github/workflows/reusable-secrets-analysis.yml@v0.14.0
+    uses: ASFHyP3/actions/.github/workflows/reusable-secrets-analysis.yml@v0.15.0
 
   call-ruff-workflow:
     # Docs: https://github.com/ASFHyP3/actions
-    uses: ASFHyP3/actions/.github/workflows/reusable-ruff.yml@v0.14.0
+    uses: ASFHyP3/actions/.github/workflows/reusable-ruff.yml@v0.15.0
 
   call-mypy-workflow:
     # Docs: https://github.com/ASFHyP3/actions
-    uses: ASFHyP3/actions/.github/workflows/reusable-mypy.yml@v0.14.0
+    uses: ASFHyP3/actions/.github/workflows/reusable-mypy.yml@v0.15.0

.github/workflows/tag-version.yml

@@ -9,6 +9,6 @@ jobs:
   call-bump-version-workflow:
     # For first-time setup, create a v0.0.0 tag as shown here:
     # https://github.com/ASFHyP3/actions#reusable-bump-versionyml
-    uses: ASFHyP3/actions/.github/workflows/reusable-bump-version.yml@v0.14.0
+    uses: ASFHyP3/actions/.github/workflows/reusable-bump-version.yml@v0.15.0
     secrets:
       USER_TOKEN: ${{ secrets.TOOLS_BOT_PAK }}

CHANGELOG.md

@@ -6,6 +6,11 @@ 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).
 
+## [1.4.2]
+
+### Changed
+* Upgraded the `reusable-mypy` action to [v0.15.0](https://github.com/ASFHyP3/actions/releases/tag/v0.15.0) and replaced the `--ignore-missing-imports` option with `disable_error_code = ["import-untyped"]` as recommended by <https://github.com/ASFHyP3/actions/issues/225>, then ignored or fixed the resulting `mypy` errors.
+
 ## [1.4.1]
 
 ### Added

pyproject.toml

@@ -99,3 +99,7 @@ warn_unused_ignores = true
 warn_unreachable = true
 strict_equality = true
 check_untyped_defs = true
+install_types = true
+non_interactive = true
+pretty = true
+disable_error_code = ["import-untyped"]

src/burst2safe/base.py

@@ -1,3 +1,4 @@
+# mypy: disable-error-code="union-attr"
 import hashlib
 from copy import deepcopy
 from datetime import datetime, timedelta
@@ -14,7 +15,7 @@
 
 class ListOfListElements:
     def __init__(
-        self, inputs: List[ET.Element], start_line: Optional[int] = None, slc_lengths: Optional[List[int]] = None
+        self, inputs: List[ET._Element], start_line: Optional[int] = None, slc_lengths: Optional[List[int]] = None
     ):
         """Initialize the ListOfListElements object.
 
@@ -23,7 +24,7 @@ def __init__(
             start_line: The starting line number of the first element.
             slc_lengths: The total line lengths of the SLCs corresponding to each element.
         """
-        self.inputs: list[ET.Element] = inputs
+        self.inputs: list[ET._Element] = inputs
         self.start_line: Optional[int] = start_line
         self.slc_lengths: Optional[list[int]] = slc_lengths
 
@@ -51,7 +52,7 @@ def __init__(
         self.inputs = sorted(self.inputs, key=self.get_first_time)
         self.has_line = elements[0].find('line') is not None
 
-    def get_first_time(self, element: ET.Element) -> datetime:
+    def get_first_time(self, element: ET._Element) -> datetime:
         """Get first time in List element
 
         Args:
@@ -60,32 +61,32 @@ def get_first_time(self, element: ET.Element) -> datetime:
         Returns:
             The first time in the element.
         """
-        first_time = min([datetime.fromisoformat(sub.find(self.time_field).text) for sub in element])
+        first_time = min([datetime.fromisoformat(sub.find(self.time_field).text) for sub in element])  # type: ignore[arg-type]
         return first_time
 
-    def get_unique_elements(self) -> List[ET.Element]:
+    def get_unique_elements(self) -> List[ET._Element]:
         """Get the elements without duplicates. Adjust line number if present.
 
         Returns:
             The list of elements without duplicates.
         """
         list_of_element_lists = [item.findall('*') for item in self.inputs]
 
-        last_time = datetime.fromisoformat(list_of_element_lists[0][-1].find(self.time_field).text)
+        last_time = datetime.fromisoformat(list_of_element_lists[0][-1].find(self.time_field).text)  # type: ignore[arg-type]
         uniques = [deepcopy(element) for element in list_of_element_lists[0]]
         if self.has_line:
             assert self.slc_lengths is not None
             previous_line_count = self.slc_lengths[0]
 
         for i, element_list in enumerate(list_of_element_lists[1:]):
-            times = [datetime.fromisoformat(element.find(self.time_field).text) for element in element_list]
+            times = [datetime.fromisoformat(element.find(self.time_field).text) for element in element_list]  # type: ignore[arg-type]
             keep_index = [index for index, time in enumerate(times) if time > last_time]
             to_keep = [deepcopy(element_list[index]) for index in keep_index]
 
             if self.has_line:
-                new_lines = [int(elem.find('line').text) + previous_line_count for elem in to_keep]
+                new_lines = [int(elem.find('line').text) + previous_line_count for elem in to_keep]  # type: ignore[arg-type]
                 for elem, line in zip(to_keep, new_lines):
-                    set_text(elem.find('line'), line)
+                    set_text(elem.find('line'), line)  # type: ignore[arg-type]
                 assert self.slc_lengths is not None
                 previous_line_count += self.slc_lengths[i]
 
@@ -95,7 +96,7 @@ def get_unique_elements(self) -> List[ET.Element]:
         return uniques
 
     @staticmethod
-    def filter_by_line(element_list: List[ET.Element], line_bounds: tuple[float, float]) -> List[ET.Element]:
+    def filter_by_line(element_list: List[ET._Element], line_bounds: tuple[float, float]) -> List[ET._Element]:
         """Filter elements by line number.
 
         Args:
@@ -106,24 +107,24 @@ def filter_by_line(element_list: List[ET.Element], line_bounds: tuple[float, flo
         """
         new_list = []
         for elem in element_list:
-            if line_bounds[0] <= int(elem.find('line').text) <= line_bounds[1]:
+            if line_bounds[0] <= int(elem.find('line').text) <= line_bounds[1]:  # type: ignore[arg-type,operator]
                 new_list.append(deepcopy(elem))
         return new_list
 
-    def update_line_numbers(self, elements: List[ET.Element]) -> None:
+    def update_line_numbers(self, elements: List[ET._Element]) -> None:
         """Update the line numbers of the elements.
 
         Args:
             elements: The list of elements to update.
         """
         for element in elements:
-            standard_line = int(element.find('line').text)
+            standard_line = int(element.find('line').text)  # type: ignore[arg-type]
             assert self.start_line is not None
             element.find('line').text = str(standard_line - self.start_line)
 
     def filter_by_time(
-        self, elements: List[ET.Element], anx_bounds: tuple[datetime, datetime], buffer: timedelta
-    ) -> List[ET.Element]:
+        self, elements: List[ET._Element], anx_bounds: tuple[datetime, datetime], buffer: timedelta
+    ) -> List[ET._Element]:
         """Filter elements by time.
 
         Args:
@@ -138,7 +139,7 @@ def filter_by_time(
         max_anx_bound = anx_bounds[1] + buffer
         filtered_elements = []
         for element in elements:
-            azimuth_time = datetime.fromisoformat(element.find(self.time_field).text)
+            azimuth_time = datetime.fromisoformat(element.find(self.time_field).text)  # type: ignore[arg-type]
             if min_anx_bound < azimuth_time < max_anx_bound:
                 filtered_elements.append(deepcopy(element))
 
@@ -149,7 +150,7 @@ def create_filtered_list(
         anx_bounds: tuple[datetime, datetime],
         buffer: timedelta = timedelta(seconds=3),
         line_bounds: Optional[tuple[float, float]] = None,
-    ) -> ET.Element:
+    ) -> ET._Element:
         """Filter elements by time/line. Adjust line number if present.
 
         Args:
@@ -172,12 +173,13 @@ def create_filtered_list(
             filtered_elements = self.filter_by_line(filtered_elements, line_bounds)
 
         new_element = ET.Element(self.name)
-        [new_element.append(element) for element in filtered_elements]
+        for element in filtered_elements:
+            new_element.append(element)
         new_element.set('count', str(len(filtered_elements)))
         return new_element
 
 
-def create_content_unit(simple_name: str, unit_type: str, rep_id: str) -> ET.Element:
+def create_content_unit(simple_name: str, unit_type: str, rep_id: str) -> ET._Element:
     """Create a content unit element for a manifest.safe file.
 
     Args:
@@ -195,7 +197,7 @@ def create_content_unit(simple_name: str, unit_type: str, rep_id: str) -> ET.Ele
     return content_unit
 
 
-def create_metadata_object(simple_name: str) -> ET.Element:
+def create_metadata_object(simple_name: str) -> ET._Element:
     """Create a metadata object element for a manifest.safe file.
 
     Args:
@@ -214,7 +216,7 @@ def create_metadata_object(simple_name: str) -> ET.Element:
 
 def create_data_object(
     simple_name: str, relative_path: Union[Path, str], rep_id: str, mime_type: str, size_bytes: int, md5: str
-) -> ET.Element:
+) -> ET._Element:
     """Create a data object element for a manifest.safe file.
 
     Args:
@@ -273,14 +275,14 @@ def __init__(self, burst_infos: list[BurstInfo], metadata_type: str, ipf_version
         products = [get_subxml_from_metadata(path, 'product', self.swath, self.pol) for path in self.metadata_paths]
         slc_lengths = []
         for annotation in products:
-            n_bursts = int(annotation.find('.//burstList').get('count'))
-            burst_length = int(annotation.find('.//linesPerBurst').text)
+            n_bursts = int(annotation.find('.//burstList').get('count'))  # type: ignore[arg-type]
+            burst_length = int(annotation.find('.//linesPerBurst').text)  # type: ignore[arg-type]
             slc_lengths.append(n_bursts * burst_length)
         self.slc_lengths = slc_lengths
 
         # annotation components to be extended by subclasses
-        self.ads_header = None
-        self.xml: Optional[ET.Element] = None
+        self.ads_header: Optional[ET._Element] = None
+        self.xml: Union[ET._Element, ET._ElementTree, None] = None
 
         # these attributes are updated when the annotation is written to a file
         self.size_bytes: Optional[int] = None
@@ -294,7 +296,7 @@ def create_ads_header(self):
         ads_header.find('imageNumber').text = f'{self.image_number:03d}'
         self.ads_header = ads_header
 
-    def merge_lists(self, list_name: str, line_bounds: Optional[tuple[int, int]] = None) -> ET.Element:
+    def merge_lists(self, list_name: str, line_bounds: Optional[tuple[int, int]] = None) -> ET._Element:
         """Merge lists of elements into a single list.
 
         Args:
@@ -303,7 +305,7 @@ def merge_lists(self, list_name: str, line_bounds: Optional[tuple[int, int]] = N
         Returns:
             The merged list element.
         """
-        list_elements = [input_xml.find(list_name) for input_xml in self.inputs]
+        list_elements: list = [input_xml.find(list_name) for input_xml in self.inputs]
         list_of_list_elements = ListOfListElements(list_elements, self.start_line, self.slc_lengths)
         merged_list = list_of_list_elements.create_filtered_list((self.min_anx, self.max_anx), line_bounds=line_bounds)
         return merged_list
@@ -331,6 +333,7 @@ def __str__(self, **kwargs):
         Args:
             kwargs: Keyword arguments to pass to the lxml
         """
+        assert self.xml is not None
         xml_str = ET.tostring(self.xml, pretty_print=True, **kwargs)
         return xml_str.decode()
 

src/burst2safe/calibration.py

@@ -1,4 +1,5 @@
 from copy import deepcopy
+from typing import Optional
 
 import lxml.etree as ET
 
@@ -18,12 +19,12 @@ def __init__(self, burst_infos: list[BurstInfo], ipf_version: str, image_number:
             image_number: Image number.
         """
         super().__init__(burst_infos, 'calibration', ipf_version, image_number)
-        self.calibration_information = None
+        self.calibration_information: Optional[ET._Element] = None
         self.calibrattion_vector_list = None
 
     def create_calibration_information(self):
         """Create the calibration information."""
-        calibration_information = [calibration.find('calibrationInformation') for calibration in self.inputs][0]
+        calibration_information = [calibration.find('calibrationInformation') for calibration in self.inputs][0]  # type: ignore[union-attr]
         self.calibration_information = deepcopy(calibration_information)
 
     def create_calibration_vector_list(self):
@@ -37,6 +38,8 @@ def assemble(self):
         self.create_calibration_vector_list()
 
         calibration = ET.Element('calibration')
+        assert self.ads_header is not None
+        assert self.calibration_information is not None
         calibration.append(self.ads_header)
         calibration.append(self.calibration_information)
         calibration.append(self.calibration_vector_list)

src/burst2safe/local2safe.py

@@ -1,3 +1,4 @@
+# mypy: disable-error-code="union-attr"
 """Generate a SAFE file from local burst extractor outputs"""
 
 import argparse
@@ -29,13 +30,15 @@ def burst_info_from_local(
     manifest = utils.get_subxml_from_metadata(xml_path, 'manifest', swath, polarization)
     xml_orbit_path = './/{*}metadataObject[@ID="measurementOrbitReference"]/metadataWrap/xmlData/{*}orbitReference'
     meta_orbit = manifest.find(xml_orbit_path)
-    abs_orbit_start, abs_orbit_stop = [int(x.text) for x in meta_orbit.findall('{*}orbitNumber')]
-    rel_orbit_start, rel_orbit_stop = [int(x.text) for x in meta_orbit.findall('{*}relativeOrbitNumber')]
+    abs_orbit_start, abs_orbit_stop = [int(x.text) for x in meta_orbit.findall('{*}orbitNumber')]  # type: ignore[arg-type]
+    rel_orbit_start, rel_orbit_stop = [int(x.text) for x in meta_orbit.findall('{*}relativeOrbitNumber')]  # type: ignore[arg-type]
     direction = meta_orbit.find('{*}extension/{*}orbitProperties/{*}pass').text.upper()
 
     product = utils.get_subxml_from_metadata(xml_path, 'product', swath, polarization)
     sensing_time_str = product.findall('swathTiming/burstList/burst')[burst_index].find('sensingTime').text
     anx_time_str = meta_orbit.find('{*}extension/{*}orbitProperties/{*}ascendingNodeTime').text
+    assert sensing_time_str is not None
+    assert anx_time_str is not None
     burst_id, rel_orbit = calculate_burstid(sensing_time_str, anx_time_str, rel_orbit_start, rel_orbit_stop, swath)
     info = utils.BurstInfo(
         granule='',