Cleanup source files.

src/openutm_verification/auth/dev_auth.py

@@ -1,14 +1,10 @@
 import datetime
-import urllib.parse
 import uuid
-from typing import Dict, List, Optional
+from abc import ABC, abstractmethod
+from typing import List
 
-import jwcrypto.common
 import jwcrypto.jwk
-import jwcrypto.jws
 import jwcrypto.jwt
-import jwt
-import requests
 
 ALL_SCOPES = [
     "dss.write.identification_service_areas",
@@ -23,47 +19,13 @@
 CLIENT_TIMEOUT = 60  # seconds
 
 
-class AuthAdapter(object):
-    """Base class for an adapter that add JWTs to requests."""
-
-    def __init__(self):
-        self._tokens = {}
+class AuthAdapter(ABC):
+    """Abstract base class for an adapter that adds JWTs to requests."""
 
+    @abstractmethod
     def issue_token(self, intended_audience: str, scopes: List[str]) -> str:
         """Subclasses must return a bearer token for the given audience."""
-
-        raise NotImplementedError()
-
-    def get_headers(self, url: str, scopes: List[str] = None) -> Dict[str, str]:
-        if scopes is None:
-            scopes = ALL_SCOPES
-        intended_audience = urllib.parse.urlparse(url).hostname
-        scope_string = " ".join(scopes)
-        if intended_audience not in self._tokens:
-            self._tokens[intended_audience] = {}
-        if scope_string not in self._tokens[intended_audience]:
-            token = self.issue_token(intended_audience, scopes)
-        else:
-            token = self._tokens[intended_audience][scope_string]
-        payload = jwt.decode(token, options={"verify_signature": False})
-        expires = EPOCH + datetime.timedelta(seconds=payload["exp"])
-        if datetime.datetime.utcnow() > expires - TOKEN_REFRESH_MARGIN:
-            token = self.issue_token(intended_audience, scopes)
-        self._tokens[intended_audience][scope_string] = token
-        return {"Authorization": "Bearer " + token}
-
-    def add_headers(self, request: requests.PreparedRequest, scopes: List[str]):
-        for k, v in self.get_headers(request.url, scopes).items():
-            request.headers[k] = v
-
-    def get_sub(self) -> Optional[str]:
-        """Retrieve `sub` claim from one of the existing tokens"""
-        for _, tokens_by_scope in self._tokens.items():
-            for token in tokens_by_scope.values():
-                payload = jwt.decode(token, options={"verify_signature": False})
-                if "sub" in payload:
-                    return payload["sub"]
-        return None
+        pass
 
 
 class NoAuth(AuthAdapter):

src/openutm_verification/auth/noauth.py

@@ -10,9 +10,6 @@
 class NoAuthCredentialsGetter:
     """Credentials getter that uses dummy authentication for development."""
 
-    def __init__(self):
-        pass
-
     def get_cached_credentials(self, audience: str, scopes: List[str]):
         """Get cached credentials using dummy authentication."""
         if not audience:

src/openutm_verification/core/clients/air_traffic/air_traffic_client.py

@@ -1,6 +1,6 @@
 import json
 import uuid
-from typing import Any, Dict, List, Optional
+from typing import Optional
 from uuid import UUID
 
 from loguru import logger

src/openutm_verification/core/clients/air_traffic/base_client.py

@@ -1,10 +1,5 @@
-from typing import Optional, Tuple
-
-import httpx
-from loguru import logger
 from pydantic_settings import BaseSettings
 
-from openutm_verification.auth.oauth2 import OAuth2Client
 from openutm_verification.core.execution.config_models import get_settings
 
 config = get_settings()

src/openutm_verification/core/clients/opensky/opensky_client.py

@@ -1,4 +1,3 @@
-import json
 from typing import Optional
 
 import pandas as pd
@@ -9,11 +8,7 @@
     OpenSkySettings,
 )
 from openutm_verification.core.execution.scenario_runner import scenario_step
-from openutm_verification.simulator.geo_json_telemetry import (
-    GeoJSONAirtrafficSimulator,
-)
 from openutm_verification.simulator.models.flight_data_types import (
-    AirTrafficGeneratorConfiguration,
     FlightObservationSchema,
 )
 

src/openutm_verification/core/reporting/reporting_models.py

@@ -3,9 +3,9 @@
 """
 
 from enum import StrEnum
-from typing import Any, Dict, List, Literal, Optional
+from typing import Any, Dict, List, Optional
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel
 
 from openutm_verification.core.execution.config_models import DeploymentDetails
 

src/openutm_verification/drip/auth_decoder.py

@@ -32,10 +32,7 @@
 
 For more information about the DRIP protocol and the Authentication message format, refer to the ASTM F3411 specification.
 """
-
 import ctypes
-import struct
-from ctypes import POINTER, Structure, c_char, c_double, c_float, c_uint8, c_uint16, c_uint32, c_void_p, sizeof
 
 import openutm_verification.drip.drip_messages as common
 

src/openutm_verification/drip/drip_messages.py

@@ -32,8 +32,7 @@
 """
 
 import ctypes
-import struct
-from ctypes import POINTER, Structure, c_char, c_double, c_float, c_uint8, c_uint16, c_uint32, c_void_p, sizeof
+from ctypes import Structure, c_char, c_double, c_float, c_uint8, c_uint16, c_uint32, c_void_p
 
 # Size constants for DRIP messages
 DRIP_ID_SIZE = 20

src/openutm_verification/drip/import_drip_decoder.py

@@ -29,7 +29,6 @@
 import argparse
 import ctypes
 import os
-import struct
 
 import openutm_verification.drip.drip_messages as common
 from openutm_verification.drip.auth_decoder import AuthDecoder
@@ -48,7 +47,6 @@ def decode_drone_id(uas_data, raw_data):
 
     if common.DRIP_MESSAGE_SIZE <= message_size:
         msg_type = raw_data[0] >> 4
-        message_type_bytes = struct.pack("B", raw_data[0])
 
         if msg_type == common.DRIP_MESSAGE_BASIC_ID:
             print("DRIP_MESSAGE_BASIC_ID")
@@ -112,7 +110,7 @@ def decodeMessagePack(data):
     uasData = common.DRIP_UAS_Data()
     pack = common.DRIP_MessagePack_encoded()
     ctypes.memmove(ctypes.addressof(pack), data, ctypes.sizeof(pack))
-    ret = decode_message_pack(uasData, pack, data)
+    decode_message_pack(uasData, pack, data)
 
     return uasData
 

src/openutm_verification/drip/location_decoder.py

@@ -32,7 +32,6 @@
 For more information about the DRIP protocol and the Location message format, refer to the ASTM F3411 specification.
 """
 
-import ctypes
 import struct
 
 import openutm_verification.drip.drip_messages as common

src/openutm_verification/drip/operator_id_decoder.py

@@ -30,8 +30,6 @@
 For more information about the DRIP protocol and the Operator ID message format, refer to the ASTM F3411 specification.
 """
 
-import ctypes
-
 from openutm_verification.drip import drip_messages as common
 
 

src/openutm_verification/drip/self_id_decoder.py

@@ -30,8 +30,6 @@
 For more information about the DRIP protocol and the Self ID message format, refer to the ASTM F3411 specification.
 """
 
-import ctypes
-
 from openutm_verification.drip import drip_messages as common
 
 

src/openutm_verification/drip/system_decoder.py

@@ -28,8 +28,6 @@
 For more information about the DRIP protocol and the System message format, refer to the ASTM F3411 specification.
 """
 
-import ctypes
-
 from openutm_verification.drip import drip_messages as common
 
 

src/openutm_verification/importers/amqp_consumer.py

@@ -7,8 +7,6 @@
 import pika
 from dotenv import find_dotenv, load_dotenv
 
-load_dotenv(find_dotenv())
-
 
 def parse_args(argv: List[str]):
     parser = argparse.ArgumentParser(description="Retrieve an access token")
@@ -24,6 +22,7 @@ def parse_args(argv: List[str]):
 
 
 def main(queue_id):
+    load_dotenv(find_dotenv())
     amqp_connection_url = env.get("AMQP_URL", "localhost")
     params = pika.URLParameters(amqp_connection_url)
     connection = pika.BlockingConnection(params)