add chia dev command to generate api peer schema classes

.pre-commit-config.yaml

@@ -13,6 +13,13 @@ repos:
         entry: ./activated.py python chia/_tests/build-init-files.py -v --root .
         language: system
         pass_filenames: false
+  - repo: local
+    hooks:
+      - id: generate-service-peer-schemas
+        name: Generate service peer schemas
+        entry: ./activated.py chia dev generate-service-peer-schemas
+        language: system
+        pass_filenames: false
   - repo: local
     hooks:
       - id: ruff_format

chia/apis/__init__.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from chia.apis.data_layer_api_schema import DataLayerApiSchema
+from chia.apis.farmer_api_schema import FarmerApiSchema
+from chia.apis.full_node_api_schema import FullNodeApiSchema
+from chia.apis.harvester_api_schema import HarvesterApiSchema
+from chia.apis.introducer_api_schema import IntroducerApiSchema
+from chia.apis.solver_api_schema import SolverApiSchema
+from chia.apis.timelord_api_schema import TimelordApiSchema
+from chia.apis.wallet_api_schema import WalletNodeApiSchema
+from chia.protocols.outbound_message import NodeType
+from chia.server.api_protocol import ApiSchemaProtocol
+
+ApiProtocolRegistry: dict[NodeType, type[ApiSchemaProtocol]] = {
+    NodeType.DATA_LAYER: DataLayerApiSchema,
+    NodeType.FARMER: FarmerApiSchema,
+    NodeType.FULL_NODE: FullNodeApiSchema,
+    NodeType.HARVESTER: HarvesterApiSchema,
+    NodeType.INTRODUCER: IntroducerApiSchema,
+    NodeType.SOLVER: SolverApiSchema,
+    NodeType.TIMELORD: TimelordApiSchema,
+    NodeType.WALLET: WalletNodeApiSchema,
+}

chia/apis/data_layer_api_schema.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, ClassVar, cast
+
+from chia.server.api_protocol import ApiMetadata, ApiSchemaProtocol
+
+
+class DataLayerApiSchema:
+    if TYPE_CHECKING:
+        _protocol_check: ApiSchemaProtocol = cast("DataLayerApiSchema", None)
+
+    metadata: ClassVar[ApiMetadata] = ApiMetadata()

chia/apis/farmer_api_schema.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, ClassVar, Optional, cast
+
+from chia.protocols import farmer_protocol, harvester_protocol
+from chia.protocols.harvester_protocol import (
+    PartialProofsData,
+    PlotSyncDone,
+    PlotSyncPathList,
+    PlotSyncPlotList,
+    PlotSyncStart,
+)
+from chia.protocols.outbound_message import Message
+from chia.protocols.solver_protocol import SolverResponse
+from chia.server.api_protocol import ApiMetadata, ApiSchemaProtocol
+from chia.server.ws_connection import WSChiaConnection
+
+
+class FarmerApiSchema:
+    if TYPE_CHECKING:
+        _protocol_check: ApiSchemaProtocol = cast("FarmerApiSchema", None)
+
+    metadata: ClassVar[ApiMetadata] = ApiMetadata()
+
+    @metadata.request(peer_required=True)
+    async def new_proof_of_space(
+        self, new_proof_of_space: harvester_protocol.NewProofOfSpace, peer: WSChiaConnection
+    ) -> None: ...
+
+    @metadata.request(peer_required=True)
+    async def partial_proofs(self, partial_proof_data: PartialProofsData, peer: WSChiaConnection) -> None: ...
+
+    @metadata.request()
+    async def solution_response(self, response: SolverResponse, peer: WSChiaConnection) -> None: ...
+
+    @metadata.request()
+    async def respond_signatures(self, response: harvester_protocol.RespondSignatures) -> None: ...
+
+    @metadata.request()
+    async def new_signage_point(self, new_signage_point: farmer_protocol.NewSignagePoint) -> None: ...
+
+    @metadata.request()
+    async def request_signed_values(
+        self, full_node_request: farmer_protocol.RequestSignedValues
+    ) -> Optional[Message]: ...
+
+    @metadata.request(peer_required=True)
+    async def farming_info(self, request: farmer_protocol.FarmingInfo, peer: WSChiaConnection) -> None: ...
+
+    @metadata.request(peer_required=True)
+    async def respond_plots(self, _: harvester_protocol.RespondPlots, peer: WSChiaConnection) -> None: ...
+
+    @metadata.request(peer_required=True)
+    async def plot_sync_start(self, message: PlotSyncStart, peer: WSChiaConnection) -> None: ...
+
+    @metadata.request(peer_required=True)
+    async def plot_sync_loaded(self, message: PlotSyncPlotList, peer: WSChiaConnection) -> None: ...
+
+    @metadata.request(peer_required=True)
+    async def plot_sync_removed(self, message: PlotSyncPathList, peer: WSChiaConnection) -> None: ...
+
+    @metadata.request(peer_required=True)
+    async def plot_sync_invalid(self, message: PlotSyncPathList, peer: WSChiaConnection) -> None: ...
+
+    @metadata.request(peer_required=True)
+    async def plot_sync_keys_missing(self, message: PlotSyncPathList, peer: WSChiaConnection) -> None: ...
+
+    @metadata.request(peer_required=True)
+    async def plot_sync_duplicates(self, message: PlotSyncPathList, peer: WSChiaConnection) -> None: ...
+
+    @metadata.request(peer_required=True)
+    async def plot_sync_done(self, message: PlotSyncDone, peer: WSChiaConnection) -> None: ...