Add Python interface for socket server

Author: object-Object

Makefile

@@ -23,7 +23,7 @@ $(RUST_PROGRAMS): %: build/rust/%.bin
 # see https://stackoverflow.com/a/61960833
 build/%-0.mlog: build/%.bin scripts/bin_to_mlog.py
 	-rm -f build/$*-[0-9].mlog build/$*-[0-9][0-9].mlog
-	python scripts/bin_to_mlog.py build/$*.bin
+	python -m mlogv32.scripts.bin_to_mlog build/$*.bin
 
 build/rust/%.bin: FORCE | build/rust
 	cd rust/$* && cargo robjcopy ../../build/rust/$*.bin

pyproject.toml

@@ -1,9 +1,120 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
 [project]
 name = "mlogv32"
 version = "0.1.0"
 requires-python = ">=3.12"
 dependencies = [
+    "pydantic>=2.11.5",
     "pymsch>=0.0.11",
     "pyperclip>=1.9.0",
+    "riscof",
+    "riscv-isac",
     "typer>=0.15.4",
 ]
+
+[dependency-groups]
+dev = [
+    "ruff>=0.11.12",
+]
+
+[tool.uv.sources]
+riscof = { git = "https://github.com/riscv/riscof", rev = "9fe3597d75" }
+riscv-isac = { git = "https://github.com/riscv-software-src/riscv-isac", rev = "777d2b4762" }
+
+[tool.hatch.build.targets.wheel]
+packages = [
+    "python/src/mlogv32",
+]
+
+[tool.ruff]
+extend-exclude = [
+    "**/venv",
+    "**/.*",
+    "**/node_modules",
+    "**/__pycache__",
+]
+
+[tool.ruff.lint]
+extend-select = [
+    "I", # import sorting
+    "RUF022", # __all__ sorting
+]
+extend-ignore = [
+    # covered by Pyright
+    "F821", # undefined variable
+    "F401", # imported but unused
+    "F841", # unused variable
+]
+
+[tool.ruff.lint.isort]
+combine-as-imports = true
+
+[tool.pyright]
+pythonVersion = "3.12"
+pythonPlatform = "All"
+
+extraPaths = [
+    "python/src",
+]
+include = [
+    "python/*",
+]
+exclude = [
+    "**/venv",
+    "**/.*",
+    "**/node_modules",
+    "**/__pycache__",
+]
+
+typeCheckingMode = "basic"
+
+strictDictionaryInference = true
+strictListInference = true
+strictSetInference = true
+
+reportAssertAlwaysTrue = "error"
+reportConstantRedefinition = "error"
+reportDeprecated = "error"
+reportFunctionMemberAccess = "error"
+reportIncompatibleMethodOverride = "error"
+reportIncompatibleVariableOverride = "error"
+reportIncompleteStub = "error"
+reportInconsistentConstructor = "error"
+reportInvalidStringEscapeSequence = "error"
+reportInvalidStubStatement = "error"
+reportInvalidTypeVarUse = "error"
+reportMatchNotExhaustive = "error"
+reportMissingParameterType = "error"
+reportOverlappingOverload = "error"
+reportSelfClsParameterName = "error"
+reportTypeCommentUsage = "error"
+reportUnknownParameterType = "error"
+reportUnnecessaryCast = "error"
+reportUnnecessaryContains = "error"
+reportUnsupportedDunderAll = "error"
+reportUntypedBaseClass = "error"
+reportUntypedClassDecorator = "error"
+reportUntypedFunctionDecorator = "error"
+reportUntypedNamedTuple = "error"
+reportWildcardImportFromLibrary = "error"
+
+reportMissingTypeArgument = "warning"
+reportPrivateUsage = "warning"
+reportUnknownLambdaType = "warning"
+reportUnknownMemberType = "warning"
+reportUnnecessaryComparison = "warning"
+reportUnnecessaryIsInstance = "warning"
+reportUnusedClass = "warning"
+reportUnusedExpression = "warning"
+reportUnusedFunction = "warning"
+reportUnusedImport = "warning"
+reportUnusedVariable = "warning"
+
+reportUnknownArgumentType = "information"
+reportUnknownVariableType = "information"
+
+reportMissingTypeStubs = "none"
+reportDuplicateImport = "none"

python/src/mlogv32/__init__.py

@@ -0,0 +1,5 @@
+__all__ = [
+    "ProcessorAccess",
+]
+
+from .processor_access import ProcessorAccess

python/src/mlogv32/processor_access.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+import socket
+from pathlib import Path
+from typing import Annotated, Any, Literal
+
+from pydantic import (
+    BaseModel as _BaseModel,
+    ConfigDict,
+    Field,
+    TypeAdapter,
+    alias_generators,
+)
+
+
+class ProcessorAccess:
+    def __init__(self, hostname: str, port: int):
+        self.hostname = hostname
+        self.port = port
+        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+
+    def socket_connect(self):
+        self.socket.connect((self.hostname, self.port))
+
+    def socket_close(self):
+        self.socket.close()
+
+    def flash(self, path: str | Path):
+        path = Path(path)
+        if not path.is_absolute():
+            raise ValueError("Path must be absolute.")
+
+        self._send_request(FlashRequest(path=path))
+        return self._recv_response()
+
+    def dump(self, path: str | Path):
+        path = Path(path)
+        if not path.is_absolute():
+            raise ValueError("Path must be absolute.")
+
+        self._send_request(DumpRequest(path=path))
+        return self._recv_response()
+
+    def start(self, wait: bool):
+        self._send_request(StartRequest(wait=wait))
+        return self._recv_response()
+
+    def stop(self):
+        self._send_request(StopRequest())
+        return self._recv_response()
+
+    def _send_request(self, request: Request) -> None:
+        message = request.model_dump_json() + "\n"
+        self.socket.sendall(message.encode("utf-8"))
+
+    def _recv_response(self) -> Response:
+        with self.socket.makefile("r", encoding="utf-8") as f:
+            line = f.readline()
+        return response_ta.validate_json(line)
+
+    def __enter__(self):
+        self.socket_connect()
+        return self
+
+    def __exit__(self, *_: Any):
+        self.socket_close()
+        return False  # propagate exceptions
+
+
+class BaseModel(_BaseModel):
+    model_config = ConfigDict(
+        alias_generator=alias_generators.to_camel,
+    )
+
+
+class FlashRequest(BaseModel):
+    type: Literal["flash"] = "flash"
+    path: Path
+
+
+class DumpRequest(BaseModel):
+    type: Literal["dump"] = "dump"
+    path: Path
+
+
+class StartRequest(BaseModel):
+    type: Literal["start"] = "start"
+    wait: bool
+
+
+class StopRequest(BaseModel):
+    type: Literal["stop"] = "stop"
+
+
+type Request = Annotated[
+    FlashRequest | DumpRequest | StartRequest | StopRequest,
+    Field(discriminator="type"),
+]
+
+
+class SuccessResponse(BaseModel):
+    type: Literal["success"]
+    message: str
+
+
+class ErrorResponse(BaseModel):
+    type: Literal["error"]
+    message: str
+
+
+type Response = Annotated[
+    SuccessResponse | ErrorResponse,
+    Field(discriminator="type"),
+]
+
+response_ta = TypeAdapter[Response](Response)

python/src/mlogv32/py.typed

@@ -74,7 +74,7 @@ def flush_output(is_done: bool):
         # we didn't break the loop, so flush the rest of the output
         flush_output(is_done=True)
 
-    print(f"Generated {output_index} mlog file{'' if output_index == 1 else 's'}.")
+    print(f"Generated {output_index} mlog file{'' if output_index == 1 else 's'}.")  # pyright: ignore[reportUnnecessaryComparison]
 
 
 if __name__ == "__main__":

python/src/mlogv32/scripts/__init__.py

@@ -24,7 +24,7 @@ def main(
         amount = ByteUtils.pop_int(data, 2, True)
         output[f"@{ctype}Count"] = amount
 
-        for i in range(0, amount):
+        for _ in range(0, amount):
             name = ByteUtils.pop_UTF(data)
             output[ctype].append(name)
 
@@ -37,12 +37,12 @@ class ByteUtils:
     @staticmethod
     def pop_bytes(data: bytearray, byte_count: int):
         out_bytes = bytearray()
-        for i in range(byte_count):
+        for _ in range(byte_count):
             out_bytes.append(data.pop(0))
         return out_bytes
 
     @staticmethod
-    def pop_int(data: bytearray, byte_count: int, signed=False):
+    def pop_int(data: bytearray, byte_count: int, signed: bool = False):
         return int.from_bytes(ByteUtils.pop_bytes(data, byte_count), signed=signed)
 
     @staticmethod