fix: mypy assignment (TDE-277)

pyproject.toml

@@ -51,7 +51,6 @@ disable_error_code = [
     "no-untyped-call",
     "operator",
     "misc",
-    "assignment",
     "import",
     "var-annotated",
     "comparison-overlap",

topo_processor/cog/command.py

@@ -12,7 +12,7 @@ class CommandDocker(TypedDict):
 class Command:
     use_docker: bool
 
-    def __init__(self, command: str, docker_ref: CommandDocker = None) -> None:
+    def __init__(self, command: str, docker_ref: Optional[CommandDocker] = None) -> None:
         self.command = command
         self.arguments = []
         self.volumes = []

topo_processor/data/data_transformers/data_transformer.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 
 if TYPE_CHECKING:
     from topo_processor.stac import Item
@@ -10,8 +10,8 @@
 class DataTransformer(ABC):
     @property
     @abstractmethod
-    def name(self) -> None:
-        str
+    def name(self) -> Optional[str]:
+        pass
 
     @abstractmethod
     def is_applicable(self, item: Item) -> bool:

topo_processor/metadata/metadata_loaders/metadata_loader.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 
 if TYPE_CHECKING:
     from topo_processor.stac import Asset
@@ -10,8 +10,8 @@
 class MetadataLoader(ABC):
     @property
     @abstractmethod
-    def name(self) -> None:
-        str
+    def name(self) -> Optional[str]:
+        pass
 
     @abstractmethod
     def is_applicable(self, asset: Asset) -> bool:

topo_processor/metadata/metadata_loaders/metadata_loader_imagery_historic.py

@@ -126,8 +126,8 @@ def read_csv(self, metadata_file: str = "") -> None:
         if not os.path.isfile(csv_path):
             raise Exception(f'Cannot find "{csv_path}"')
 
-        with open(csv_path, "r") as csv_path:
-            reader = csv.DictReader(csv_path, delimiter=",")
+        with open(csv_path, "r") as csv_text:
+            reader = csv.DictReader(csv_text, delimiter=",")
             for row in reader:
                 if row["raw_filename"]:
                     raw_filename = row["raw_filename"]

topo_processor/metadata/metadata_validators/metadata_validator.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 
 if TYPE_CHECKING:
     from topo_processor.stac import Item
@@ -10,8 +10,8 @@
 class MetadataValidator(ABC):
     @property
     @abstractmethod
-    def name(self) -> None:
-        str
+    def name(self) -> Optional[str]:
+        pass
 
     @abstractmethod
     def is_applicable(self, item: Item) -> bool:

topo_processor/stac/asset.py

@@ -1,6 +1,6 @@
 from mimetypes import MimeTypes
 from os import path
-from typing import TYPE_CHECKING, Any, Dict, Union
+from typing import TYPE_CHECKING, Any, Dict, Optional, Union
 
 import pystac
 
@@ -14,23 +14,18 @@
 
 class Asset(Validity):
     source_path: str  # The raw file location on disk
-    target: str  # New file name used for uploading
-    content_type: str
-    needs_upload = bool
+    target: Optional[str] = None  # New file name used for uploading
+    content_type: Optional[str] = None
+    needs_upload: bool = True
     href: str
     properties: Dict[str, Any]
-    item: "Item"
-    key_name: AssetKey
+    item: Optional["Item"] = None
+    key_name: Optional[AssetKey] = None
 
     def __init__(self, source_path: str):
         super().__init__()
         self.source_path = source_path
-        self.content_type = None
-        self.target = None
-        self.needs_upload = True
         self.properties = {}
-        self.item = None
-        self.key_name = None
 
     def file_ext(self) -> str:
         return path.splitext(self.target if self.target else self.source_path)[1]

topo_processor/stac/collection.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import os
-from datetime import date, datetime
+from datetime import datetime
 from shutil import rmtree
 from tempfile import mkdtemp
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple
@@ -36,12 +36,12 @@ class Collection(Validity):
     items: Dict[str, "Item"]
     linz_providers: List[Dict[str, Any]]
     providers: List[pystac.Provider]
-    schema: str
+    schema: Optional[str]
     extra_fields: Dict[str, Any]
     linz_geospatial_type: str
 
     stac_extensions: Set[str]
-    summaries: Summaries
+    summaries: Summaries = Summaries.empty()
 
     def __init__(self, title: str):
         super().__init__()
@@ -50,7 +50,6 @@ def __init__(self, title: str):
         self.title = title
         self.items = {}
         self.schema = DefaultSchemaUriMap().get_object_schema_uri(pystac.STACObjectType.COLLECTION, pystac.get_stac_version())
-        self.stac_extensions = set([])
         self.extra_fields = dict(
             {
                 "linz:security_classification": "unclassified",
@@ -63,7 +62,6 @@ def __init__(self, title: str):
         self.linz_providers = []
         self.stac_extensions = set([StacExtensions.file.value])
         self.providers = [Providers.TTW.value]
-        self.summaries = Summaries.empty()
 
     def add_item(self, item: Item) -> None:
         if item.collection is not None and item.collection != self:

topo_processor/stac/item.py

@@ -1,5 +1,5 @@
-from datetime import datetime
-from typing import Any, Dict, List, Set
+import datetime as dt
+from typing import Any, Dict, List, Optional, Set
 
 import shapely.geometry
 from pystac import pystac
@@ -15,29 +15,25 @@
 class Item(Validity):
 
     id: str
-    geometry_poly: shapely.geometry.Polygon
-    linz_geospatial_type: str
-    datetime: datetime
+    geometry_poly: Optional[shapely.geometry.Polygon] = None
+    linz_geospatial_type: str = ""
+    datetime: Optional[dt.datetime] = None
     properties: Dict[str, Any]
     stac_extensions: Set[str]
-    collection: Collection
     assets: List[Asset]
-    schema: str
+    collection: Optional[Collection] = None
+    schema: Optional[str]
 
     def __init__(self, item_id: str):
         super().__init__()
         self.id = item_id
-        self.datetime = None
         self.properties = {
             # TODO: [TDE-237] to generate release versioning
             "processing:software": {"Topo Processor": "0.1.0"},
             # TODO: decision to be made on version ref comments [TDE-230] hardcode to '1' for now
             "version": "1",
         }
         self.stac_extensions = set([StacExtensions.file.value])
-        self.collection = None
-        self.geometry_poly = None
-        self.linz_geospatial_type = ""
         self.assets = []
         self.schema = DefaultSchemaUriMap().get_object_schema_uri(pystac.STACObjectType.ITEM, pystac.get_stac_version())
 
@@ -57,6 +53,8 @@ def add_asset(self, asset: Asset) -> None:
 
     def add_extension(self, ext: str, add_to_collection: bool = True) -> None:
         self.stac_extensions.add(ext)
+        if not self.collection:
+            return
         if add_to_collection:
             self.collection.add_extension(ext)
 

topo_processor/stac/linz_provider.py

@@ -2,9 +2,10 @@
 from typing import Any, Dict, List, Optional
 
 from pystac import Provider
+from pystac.utils import StringEnum
 
 
-class LinzProviderRole(str, Enum):
+class LinzProviderRole(StringEnum):
     """Enumerates the allows values of the LinzProvider "role" field."""
 
     MANAGER = "manager"
@@ -13,9 +14,11 @@ class LinzProviderRole(str, Enum):
 
 class LinzProvider(Provider):
 
-    roles: Optional[List[LinzProviderRole]]
+    roles: Optional[List[LinzProviderRole]]  # type:ignore
     """Optional roles of the provider. Any of manager or custodian.
-    LINZ override of pystac.ProviderRole Enum."""
+    LINZ override of pystac.ProviderRole Enum.
+    Type ignored due to: https://github.com/radiantearth/stac-spec/issues/1147
+    """
 
     def __init__(
         self,

topo_processor/util/aws_files.py

@@ -55,7 +55,7 @@ def load_file_content(bucket_name: str, object_path: str) -> Dict[str, Any]:
         json_result: Dict[str, Any] = json.loads(object_content.get()["Body"].read())
         return json_result
 
-    result: Dict[str, Any] = object_content.get()["Body"].read()
+    result: Dict[str, Any] = json.loads(object_content.get()["Body"].read().decode("utf-8"))
     return result
 
 

topo_processor/util/configuration.py

@@ -1,10 +1,11 @@
 from os import environ
 from tempfile import mkdtemp
+from typing import cast
 
 from dotenv import load_dotenv
 
 load_dotenv()
-lds_cache_bucket: str = environ.get("LDS_CACHE_BUCKET")
-aws_role_config_path: str = environ.get("AWS_ROLES_CONFIG")
-aws_profile: str = environ.get("AWS_PROFILE")
+lds_cache_bucket: str = cast(str, environ.get("LDS_CACHE_BUCKET"))
+aws_role_config_path: str = cast(str, environ.get("AWS_ROLES_CONFIG"))
+aws_profile: str = cast(str, environ.get("AWS_PROFILE"))
 temp_folder: str = mkdtemp()

topo_processor/util/gzip.py

@@ -1,5 +1,5 @@
 import gzip
-import os
+from typing import Optional
 
 from linz_logger.logger import get_log
 
@@ -11,7 +11,7 @@ def is_gzip_file(file_path: str) -> bool:
 
 
 def decompress_file(file_path: str) -> None:
-    input: gzip.GzipFile = None
+    input: Optional[gzip.GzipFile] = None
 
     try:
         input = gzip.GzipFile(file_path, "rb")
@@ -20,7 +20,8 @@ def decompress_file(file_path: str) -> None:
         get_log().error("File decompression failed", file=file_path, error=e)
         raise e
     finally:
-        input.close()
+        if input:
+            input.close()
 
     output = open(file_path, "wb")
     output.write(s)

topo_processor/util/valid.py

@@ -1,13 +1,16 @@
+from typing import Optional
+
+
 class Validity:
     def __init__(self) -> None:
         self.log = []
         self._valid = True
 
-    def add_error(self, msg: str, cause: str, e: Exception = None) -> None:
+    def add_error(self, msg: str, cause: str, e: Optional[Exception] = None) -> None:
         self.log.append({"msg": msg, "level": "error", "cause": cause, "error": e})
         self._valid = False
 
-    def add_warning(self, msg: str, cause: str, e: Exception = None) -> None:
+    def add_warning(self, msg: str, cause: str, e: Optional[Exception] = None) -> None:
         self.log.append({"msg": msg, "level": "warning", "cause": cause, "error": e})
 
     def is_valid(self) -> bool: