chore: an AI's inadequate pass at adding strict typing

Author: AndyEveritt

examples/custom_http_endpoint.py

@@ -7,23 +7,22 @@
 """
 
 import time
-import os
 import json
 
 from dsf.connections import CommandConnection
-from dsf.http import HttpEndpointConnection, HttpResponseType
+from dsf.http import HttpEndpointConnection, HttpEndpointUnixSocket, HttpResponseType
 from dsf.object_model import HttpEndpointType
 
 
-async def respond_something(http_endpoint_connection: HttpEndpointConnection):
+async def respond_something(http_endpoint_connection: HttpEndpointConnection) -> None:
     r = await http_endpoint_connection.read_request()
     if (len(r.body) > 0):
-        data = json.loads(r.body)
+        json.loads(r.body)
     await http_endpoint_connection.send_response(200, "so happy you asked for it!", HttpResponseType.PlainText)
     http_endpoint_connection.close()
 
 
-def custom_http_endpoint():
+def custom_http_endpoint() -> tuple[CommandConnection, HttpEndpointUnixSocket]:
     cmd_conn = CommandConnection(debug=True)
     cmd_conn.connect()
 
@@ -39,11 +38,14 @@ def custom_http_endpoint():
 
 
 if __name__ == "__main__":
+    cmd_conn: CommandConnection | None = None
+    endpoint: HttpEndpointUnixSocket | None = None
     try:
         cmd_conn, endpoint = custom_http_endpoint()
         # This just simulates doing other things as the new endpoint handler runs async
         time.sleep(1800)
     finally:
         if endpoint is not None:
             endpoint.close()
-        cmd_conn.close()
+        if cmd_conn is not None:
+            cmd_conn.close()

src/dsf/__init__.py

@@ -1,27 +1,27 @@
-__version__ = "3.6.0"
+__version__: str = "3.6.0"
 
 import json
 import os
 
 # Default socket file path
-SOCKET_FILE = "/run/dsf/dcs.sock"
+SOCKET_FILE: str = "/run/dsf/dcs.sock"
 
 # Try to read socket file path from config
 config_path = "/opt/dsf/conf/config.json"
 if os.path.exists(config_path):
     try:
         with open(config_path, 'r') as f:
             config = json.load(f)
-            socket_dir = config.get("SocketDirectory", "/run/dsf")
-            socket_file = config.get("SocketFile", "dcs.sock")
+            socket_dir = str(config.get("SocketDirectory", "/run/dsf"))
+            socket_file = str(config.get("SocketFile", "dcs.sock"))
             SOCKET_FILE = os.path.join(socket_dir, socket_file)
     except (json.JSONDecodeError, IOError):
         pass  # Use default if config file is invalid or inaccessible
 
 # allowed connection per unix server
-DEFAULT_BACKLOG = 4
+DEFAULT_BACKLOG: int = 4
 
 # DSF protocol version
-PROTOCOL_VERSION = 12
+PROTOCOL_VERSION: int = 12
 
 from . import commands, connections, http, object_model

src/dsf/commands/base_command.py

@@ -2,11 +2,13 @@ class BaseCommand:
     """Base class of a command."""
 
     @classmethod
-    def from_json(cls, data):
+    def from_json(cls, data: dict[str, object]) -> "BaseCommand":
         """Deserialize an instance of this class from a JSON deserialized dictionary"""
-        return cls(**data)
+        command = data.get("command", "")
+        kwargs = {key: value for key, value in data.items() if key != "command"}
+        return cls(command=str(command), **kwargs)
 
-    def __init__(self, command: str, **kwargs):
+    def __init__(self, command: str, **kwargs: object) -> None:
         self.command = command
         for key, value in kwargs.items():
             self.__dict__[key] = value

src/dsf/commands/code.py

@@ -13,15 +13,23 @@ class Code(BaseCommand):
     """A parsed representation of a generic G/M/T-code"""
 
     @classmethod
-    def from_json(cls, data):
+    def from_json(cls, data: dict[str, object] | str) -> "Code":
         """Deserialize an instance of this class from JSON deserialized dictionary"""
-        data["result"] = None if data["result"] is None else list(map(Message.from_json, data["result"]))
-        data["parameters"] = list(map(CodeParameter.from_json, data["parameters"]))
+        if isinstance(data, str):
+            raise TypeError("Code.from_json expects a decoded JSON dictionary")
+        data = dict(data)
+        result = data.get("result")
+        parameters = data.get("parameters")
+        data["result"] = None if not isinstance(result, list) else [Message.from_json(item)
+                                                                    for item in result if isinstance(item, dict)]
+        data["parameters"] = [] if not isinstance(parameters, list) else [CodeParameter.from_json(
+            item) for item in parameters if isinstance(item, dict)]
         if "channel" in data:
             data["channel"] = CodeChannel(data["channel"])
-        return cls(**data)
+        command = str(data.pop("command", ""))
+        return cls(command=command, **data)
 
-    def __init__(self, **kwargs):
+    def __init__(self, command: str = "", **kwargs: object):
         # The connection ID this code was received from. If this is 0, the code originates from an internal DCS task
         # Usually there is no need to populate this property.
         # It is internally overwritten by the control server on receipt
@@ -73,14 +81,14 @@ def __init__(self, **kwargs):
         # List of parsed code parameters
         self.parameters: List[CodeParameter] = []
 
-        super().__init__(**kwargs)
+        super().__init__(command=command, **kwargs)
 
     @property
     def is_from_file_channel(self) -> bool:
         """Check if this code is from a file channel"""
         return self.channel is CodeChannel.File or self.channel is CodeChannel.File2
 
-    def parameter(self, letter: str, default=None):
+    def parameter(self, letter: str, default: object = None) -> CodeParameter | None:
         """Retrieve the parameter whose letter equals c or generate a default parameter"""
         letter = letter.upper()
         param = [param for param in self.parameters if param.letter.upper() == letter]
@@ -90,7 +98,7 @@ def parameter(self, letter: str, default=None):
             return CodeParameter.simple_param(letter, default)
         return None
 
-    def get_unprecedented_string(self, quote: bool = False):
+    def get_unprecedented_string(self, quote: bool = False) -> str:
         """
         Reconstruct an unprecedented string from the parameter list or
         retrieve the parameter which does not have a letter assigned.
@@ -103,7 +111,7 @@ def get_unprecedented_string(self, quote: bool = False):
                 str_list.append(f"{param.letter}{param.string_value}")
         return " ".join(str_list)
 
-    def __str__(self):
+    def __str__(self) -> str:
         """Convert the parsed code back to a text-based G/M/T-code"""
         if self.keyword != KeywordType.KeywordNone:
             if self.keywordArgument is not None:
@@ -129,7 +137,7 @@ def __str__(self):
 
         return "".join(str_list)
 
-    def short_str(self):
+    def short_str(self) -> str:
         """Convert only the command portion to a text-based G/M/T-code (e.g. G28)"""
         if self.type == CodeType.Comment:
             return "(comment)"
@@ -143,7 +151,7 @@ def short_str(self):
 
         return f"{prefix}{self.type}"
 
-    def keyword_to_str(self):
+    def keyword_to_str(self) -> str:
         """Convert the keyword to a string"""
         return {
             KeywordType.If: "if",
@@ -157,7 +165,7 @@ def keyword_to_str(self):
             KeywordType.Set: "set",
             KeywordType.Echo: "echo",
             KeywordType.Global: "global",
-        }.get(self.keyword)
+        }[self.keyword]
 
-    def is_flag_set(self, flag: CodeFlags):
+    def is_flag_set(self, flag: CodeFlags) -> bool:
         return self.flags & flag != 0

src/dsf/commands/code_channel.py

@@ -54,5 +54,5 @@ class CodeChannel(str, Enum):
     DEFAULT_CHANNEL = SBC
 
     @staticmethod
-    def list():
-        return list(map(lambda cc: cc.value, CodeChannel))
+    def list() -> list["CodeChannel"]:
+        return list(CodeChannel)