From 504a4298c92f798ce3f1467333c6336f3362f29c Mon Sep 17 00:00:00 2001 From: Vlad Iftime Date: Sun, 13 Apr 2025 16:17:51 +0200 Subject: [PATCH 1/5] File structure scaffolding Signed-off-by: Vlad Iftime --- pyproject.toml | 12 +- src/s2python/authorization/client.py | 0 src/s2python/authorization/fastapi_service.py | 0 src/s2python/authorization/flask_service.py | 0 src/s2python/authorization/server.py | 0 src/s2python/authorization/service.py | 0 src/s2python/s2_connection.py | 55 +++---- .../s2-pairing/s2-over-ip-pairing.yaml | 136 ++++++++++++++++++ 8 files changed, 164 insertions(+), 39 deletions(-) create mode 100644 src/s2python/authorization/client.py create mode 100644 src/s2python/authorization/fastapi_service.py create mode 100644 src/s2python/authorization/flask_service.py create mode 100644 src/s2python/authorization/server.py create mode 100644 src/s2python/authorization/service.py create mode 100644 src/s2python/specification/s2-pairing/s2-over-ip-pairing.yaml diff --git a/pyproject.toml b/pyproject.toml index 7fd26b9..30b0dc7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,3 +1,13 @@ [build-system] requires = ["setuptools"] -build-backend = "setuptools.build_meta" \ No newline at end of file +build-backend = "setuptools.build_meta" + +[project] +name = "s2-python" +description = "S2 Protocol Python Implementation" +version = "0.5.0" + +[project.optional-dependencies] +ws = ["websockets"] +fastapi = ["fastapi"] +flask = ["Flask"] diff --git a/src/s2python/authorization/client.py b/src/s2python/authorization/client.py new file mode 100644 index 0000000..e69de29 diff --git a/src/s2python/authorization/fastapi_service.py b/src/s2python/authorization/fastapi_service.py new file mode 100644 index 0000000..e69de29 diff --git a/src/s2python/authorization/flask_service.py b/src/s2python/authorization/flask_service.py new file mode 100644 index 0000000..e69de29 diff --git a/src/s2python/authorization/server.py b/src/s2python/authorization/server.py new file mode 100644 index 0000000..e69de29 diff --git a/src/s2python/authorization/service.py b/src/s2python/authorization/service.py new file mode 100644 index 0000000..e69de29 diff --git a/src/s2python/s2_connection.py b/src/s2python/s2_connection.py index 897344c..08b10cb 100644 --- a/src/s2python/s2_connection.py +++ b/src/s2python/s2_connection.py @@ -8,8 +8,11 @@ from dataclasses import dataclass from typing import Any, Optional, List, Type, Dict, Callable, Awaitable, Union -import websockets -from websockets.asyncio.client import ClientConnection as WSConnection, connect as ws_connect +try: + import websockets + from websockets.asyncio.client import ClientConnection as WSConnection, connect as ws_connect +except ImportError: + raise ImportError("You need to run 'pip install s2-python[ws]' to use this feature.") from s2python.common import ( ReceptionStatusValues, @@ -51,13 +54,9 @@ class AssetDetails: # pylint: disable=too-many-instance-attributes firmware_version: Optional[str] = None serial_number: Optional[str] = None - def to_resource_manager_details( - self, control_types: List[S2ControlType] - ) -> ResourceManagerDetails: + def to_resource_manager_details(self, control_types: List[S2ControlType]) -> ResourceManagerDetails: return ResourceManagerDetails( - available_control_types=[ - control_type.get_protocol_control_type() for control_type in control_types - ], + available_control_types=[control_type.get_protocol_control_type() for control_type in control_types], currency=self.currency, firmware_version=self.firmware_version, instruction_processing_delay=self.instruction_processing_delay, @@ -298,9 +297,7 @@ async def wait_till_connection_restart() -> None: self._eventloop.create_task(wait_till_connection_restart()), ] - (done, pending) = await asyncio.wait( - background_tasks, return_when=asyncio.FIRST_COMPLETED - ) + (done, pending) = await asyncio.wait(background_tasks, return_when=asyncio.FIRST_COMPLETED) if self._current_control_type: self._current_control_type.deactivate(self) self._current_control_type = None @@ -333,9 +330,7 @@ async def _connect_ws(self) -> None: connection_kwargs["ssl"].verify_mode = ssl.CERT_NONE if self._bearer_token: - connection_kwargs["additional_headers"] = { - "Authorization": f"Bearer {self._bearer_token}" - } + connection_kwargs["additional_headers"] = {"Authorization": f"Bearer {self._bearer_token}"} self.ws = await ws_connect(uri=self.url, **connection_kwargs) except (EOFError, OSError) as e: @@ -343,21 +338,15 @@ async def _connect_ws(self) -> None: async def _connect_as_rm(self) -> None: await self.send_msg_and_await_reception_status_async( - Handshake( - message_id=uuid.uuid4(), role=self.role, supported_protocol_versions=[S2_VERSION] - ) + Handshake(message_id=uuid.uuid4(), role=self.role, supported_protocol_versions=[S2_VERSION]) ) logger.debug("Send handshake to CEM. Expecting Handshake and HandshakeResponse from CEM.") await self._handle_received_messages() - async def handle_handshake( - self, _: "S2Connection", message: S2Message, send_okay: Awaitable[None] - ) -> None: + async def handle_handshake(self, _: "S2Connection", message: S2Message, send_okay: Awaitable[None]) -> None: if not isinstance(message, Handshake): - logger.error( - "Handler for Handshake received a message of the wrong type: %s", type(message) - ) + logger.error("Handler for Handshake received a message of the wrong type: %s", type(message)) return logger.debug( @@ -401,12 +390,8 @@ async def handle_select_control_type_as_rm( logger.debug("CEM selected control type %s. Activating control type.", message.control_type) - control_types_by_protocol_name = { - c.get_protocol_control_type(): c for c in self.control_types - } - selected_control_type: Optional[S2ControlType] = control_types_by_protocol_name.get( - message.control_type - ) + control_types_by_protocol_name = {c.get_protocol_control_type(): c for c in self.control_types} + selected_control_type: Optional[S2ControlType] = control_types_by_protocol_name.get(message.control_type) if self._current_control_type is not None: await self._eventloop.run_in_executor(None, self._current_control_type.deactivate, self) @@ -424,9 +409,7 @@ async def _receive_messages(self) -> None: to any calls of `send_msg_and_await_reception_status`. """ if self.ws is None: - raise RuntimeError( - "Cannot receive messages if websocket connection is not yet established." - ) + raise RuntimeError("Cannot receive messages if websocket connection is not yet established.") logger.info("S2 connection has started to receive messages.") @@ -470,9 +453,7 @@ async def _receive_messages(self) -> None: async def _send_and_forget(self, s2_msg: S2Message) -> None: if self.ws is None: - raise RuntimeError( - "Cannot send messages if websocket connection is not yet established." - ) + raise RuntimeError("Cannot send messages if websocket connection is not yet established.") json_msg = s2_msg.to_json() logger.debug("Sending message %s", json_msg) @@ -532,9 +513,7 @@ def send_msg_and_await_reception_status_sync( self, s2_msg: S2Message, timeout_reception_status: float = 5.0, raise_on_error: bool = True ) -> ReceptionStatus: return asyncio.run_coroutine_threadsafe( - self.send_msg_and_await_reception_status_async( - s2_msg, timeout_reception_status, raise_on_error - ), + self.send_msg_and_await_reception_status_async(s2_msg, timeout_reception_status, raise_on_error), self._eventloop, ).result() diff --git a/src/s2python/specification/s2-pairing/s2-over-ip-pairing.yaml b/src/s2python/specification/s2-pairing/s2-over-ip-pairing.yaml new file mode 100644 index 0000000..e3bf691 --- /dev/null +++ b/src/s2python/specification/s2-pairing/s2-over-ip-pairing.yaml @@ -0,0 +1,136 @@ +openapi: 3.0.3 +info: + version: "0.1" + title: s2-over-ip pairing and connection initiation + description: "Description of the pairing process over IP for S2" +paths: + /requestPairing: + post: + description: Initiate pairing + requestBody: + description: TODO + content: + application/json: + schema: + $ref: '#/components/schemas/PairingRequest' + responses: + '200': + description: TODO + content: + application/json: + schema: + $ref: '#/components/schemas/PairingResponse' + /requestConnection: + post: + description: TODO + requestBody: + description: TODO + content: + application/json: + schema: + $ref: '#/components/schemas/ConnectionRequest' + responses: + '200': + description: TODO + content: + application/json: + schema: + $ref: '#/components/schemas/ConnectionDetails' +components: + schemas: + PairingInfo: + type: object + properties: + pairingUri: + type: string + format: uri + token: + $ref: "#/components/schemas/PairingToken" + validUntil: + type: string + format: date-time + PairingRequest: + type: object + properties: + token: + $ref: "#/components/schemas/PairingToken" + publicKey: + type: string + format: byte + s2ClientNodeId: + type: string + format: uuid + s2ClientNodeDescription: + $ref: "#/components/schemas/S2NodeDescription" + supportedProtocols: + type: array + items: + $ref: "#/components/schemas/Protocols" + PairingResponse: + type: object + properties: + s2ServerNodeId: + type: string + format: uuid + serverNodeDescription: + $ref: "#/components/schemas/S2NodeDescription" + requestConnectionUri: + type: string + format: uri + ConnectionRequest: + type: object + properties: + s2ClientNodeId: + type: string + format: uuid + supportedProtocols: + type: array + items: + $ref: "#/components/schemas/Protocols" + ConnectionDetails: + type: object + properties: + selectedProtocol: + $ref: "#/components/schemas/Protocols" + challenge: + type: string + format: byte + connectionUri: + type: string + format: uri + S2NodeDescription: + type: object + description: TODO nog even over nadenken + properties: + brand: + type: string + logoUri: + type: string + format: uri + type: + type: string + modelName: + type: string + userDefinedName: + type: string + role: + $ref: "#/components/schemas/S2Role" + deployment: + $ref: "#/components/schemas/Deployment" + Protocols: + type: string + enum: + - WebSocketSecure + S2Role: + type: string + enum: + - CEM + - RM + Deployment: + type: string + enum: + - WAN + - LAN + PairingToken: + type: string + pattern: "^[0-9a-zA-Z]{32}$" \ No newline at end of file From 34e2e51344ed79c6059d19132e9a67be9c4f443e Mon Sep 17 00:00:00 2001 From: Vlad Iftime Date: Sun, 13 Apr 2025 16:39:00 +0200 Subject: [PATCH 2/5] Added the generated dataclasses from the openapi specs. Also added serialisation and deserealisation Signed-off-by: Vlad Iftime --- src/s2python/authorization/client.py | 73 ++++++++++ src/s2python/generated/gen_s2_pairing.py | 173 +++++++++++++++++++++++ 2 files changed, 246 insertions(+) create mode 100644 src/s2python/generated/gen_s2_pairing.py diff --git a/src/s2python/authorization/client.py b/src/s2python/authorization/client.py index e69de29..bf76be5 100644 --- a/src/s2python/authorization/client.py +++ b/src/s2python/authorization/client.py @@ -0,0 +1,73 @@ +from abc import ABC, abstractmethod +from typing import Any, Dict + + +class AbstractConnectionClient(ABC): + """Abstract class for handling the /requestConnection endpoint.""" + + def request_connection(self) -> Any: + """Orchestrate the connection request flow: build → execute → handle.""" + request_data = self.build_connection_request() + response_data = self.execute_connection_request(request_data) + return self.handle_connection_response(response_data) + + @abstractmethod + def build_connection_request(self) -> Dict: + """ + Build the payload for the ConnectionRequest schema. + Returns a dictionary with keys: s2ClientNodeId, supportedProtocols. + """ + pass + + @abstractmethod + def execute_connection_request(self, request_data: Dict) -> Dict: + """ + Execute the POST request to /requestConnection. + Implementations should send the request_data to the endpoint + and return the JSON response as a dictionary. + """ + pass + + @abstractmethod + def handle_connection_response(self, response_data: Dict) -> Any: + """ + Process the ConnectionDetails response (e.g., extract challenge and connection URI). + The response_data contains keys: selectedProtocol, challenge, connectionUri. + """ + pass + + +class AbstractPairingClient(ABC): + """Abstract class for handling the /requestPairing endpoint.""" + + def request_pairing(self) -> Any: + """Orchestrate the pairing request flow: build → execute → handle.""" + request_data = self.build_pairing_request() + response_data = self.execute_pairing_request(request_data) + return self.handle_pairing_response(response_data) + + @abstractmethod + def build_pairing_request(self) -> Dict: + """ + Build the payload for the PairingRequest schema. + Returns a dictionary with keys: token, publicKey, s2ClientNodeId, + s2ClientNodeDescription, supportedProtocols. + """ + pass + + @abstractmethod + def execute_pairing_request(self, request_data: Dict) -> Dict: + """ + Execute the POST request to /requestPairing. + Implementations should send the request_data to the endpoint + and return the JSON response as a dictionary. + """ + pass + + @abstractmethod + def handle_pairing_response(self, response_data: Dict) -> Any: + """ + Process the PairingResponse (e.g., extract server details). + The response_data contains keys: s2ServerNodeId, serverNodeDescription, requestConnectionUri. + """ + pass diff --git a/src/s2python/generated/gen_s2_pairing.py b/src/s2python/generated/gen_s2_pairing.py new file mode 100644 index 0000000..32979b9 --- /dev/null +++ b/src/s2python/generated/gen_s2_pairing.py @@ -0,0 +1,173 @@ +""" +Generated classes based on s2-over-ip-pairing.yaml OpenAPI schema. +This file is auto-generated and should not be modified directly. +""" + +import uuid +from dataclasses import dataclass +from datetime import datetime +from enum import Enum, auto +from typing import List, Optional + + +class Protocols(str, Enum): + """Supported protocol types.""" + + WebSocketSecure = "WebSocketSecure" + + +class S2Role(str, Enum): + """Roles in the S2 protocol.""" + + CEM = "CEM" + RM = "RM" + + +class Deployment(str, Enum): + """Deployment types.""" + + WAN = "WAN" + LAN = "LAN" + + +@dataclass +class S2NodeDescription: + """Description of an S2 node.""" + + brand: Optional[str] = None + logoUri: Optional[str] = None + type: Optional[str] = None + modelName: Optional[str] = None + userDefinedName: Optional[str] = None + role: Optional[S2Role] = None + deployment: Optional[Deployment] = None + + +class PairingToken(str): + """A token used for pairing. + + Must match pattern: ^[0-9a-zA-Z]{32}$ + """ + + def __new__(cls, content: str): + import re + + if not re.match(r"^[0-9a-zA-Z]{32}$", content): + raise ValueError("PairingToken must be 32 alphanumeric characters") + return super().__new__(cls, content) + + +@dataclass +class PairingInfo: + """Information about a pairing.""" + + pairingUri: Optional[str] = None + token: Optional[PairingToken] = None + validUntil: Optional[datetime] = None + + +@dataclass +class PairingRequest: + """Request to initiate pairing.""" + + token: Optional[PairingToken] = None + publicKey: Optional[bytes] = None + s2ClientNodeId: Optional[uuid.UUID] = None + s2ClientNodeDescription: Optional[S2NodeDescription] = None + supportedProtocols: Optional[List[Protocols]] = None + + +@dataclass +class PairingResponse: + """Response to a pairing request.""" + + s2ServerNodeId: Optional[uuid.UUID] = None + serverNodeDescription: Optional[S2NodeDescription] = None + requestConnectionUri: Optional[str] = None + + +@dataclass +class ConnectionRequest: + """Request to establish a connection.""" + + s2ClientNodeId: Optional[uuid.UUID] = None + supportedProtocols: Optional[List[Protocols]] = None + + +@dataclass +class ConnectionDetails: + """Details for establishing a connection.""" + + selectedProtocol: Optional[Protocols] = None + challenge: Optional[bytes] = None + connectionUri: Optional[str] = None + + +# Serialization/Deserialization functions + + +def _is_dataclass_instance(obj): + """Check if an object is a dataclass instance.""" + from dataclasses import is_dataclass + + return is_dataclass(obj) and not isinstance(obj, type) + + +def to_dict(obj): + """Convert a dataclass instance to a dictionary.""" + if isinstance(obj, datetime): + return obj.isoformat() + elif isinstance(obj, uuid.UUID): + return str(obj) + elif isinstance(obj, bytes): + import base64 + + return base64.b64encode(obj).decode("ascii") + elif isinstance(obj, Enum): + return obj.value + elif isinstance(obj, list): + return [to_dict(item) for item in obj] + elif _is_dataclass_instance(obj): + result = {} + for field in obj.__dataclass_fields__: + value = getattr(obj, field) + if value is not None: + result[field] = to_dict(value) + return result + else: + return obj + + +def from_dict(cls, data): + """Create a dataclass instance from a dictionary.""" + if data is None: + return None + + if cls is datetime: + return datetime.fromisoformat(data) + elif cls is uuid.UUID: + return uuid.UUID(data) + elif cls is bytes: + import base64 + + return base64.b64decode(data.encode("ascii")) + elif issubclass(cls, Enum): + return cls(data) + elif issubclass(cls, PairingToken): + return PairingToken(data) + elif hasattr(cls, "__dataclass_fields__"): + fieldtypes = cls.__annotations__ + instance_data = {} + + for field, field_type in fieldtypes.items(): + if field in data and data[field] is not None: + # Handle List[Type] annotations + if hasattr(field_type, "__origin__") and field_type.__origin__ is list: + item_type = field_type.__args__[0] + instance_data[field] = [from_dict(item_type, item) for item in data[field]] + else: + instance_data[field] = from_dict(field_type, data[field]) + + return cls(**instance_data) + else: + return data From 93c7f0f1b542d70e6d91fb9356156fab3b269107 Mon Sep 17 00:00:00 2001 From: Vlad Iftime Date: Tue, 15 Apr 2025 11:47:10 +0200 Subject: [PATCH 3/5] Added the generated dataclasses from the openapi specs. Also added serialisation and deserealisation Signed-off-by: Vlad Iftime --- .gitignore | 1 + .../s2-pairing/s2-over-ip-pairing.yaml | 136 ------------------ 2 files changed, 1 insertion(+), 136 deletions(-) delete mode 100644 src/s2python/specification/s2-pairing/s2-over-ip-pairing.yaml diff --git a/.gitignore b/.gitignore index 5baf340..3f96451 100644 --- a/.gitignore +++ b/.gitignore @@ -13,3 +13,4 @@ venv dist/ build/ %LOCALAPPDATA% +s2-python/src/s2python/specification/s2-pairing/s2-over-ip-pairing.yaml diff --git a/src/s2python/specification/s2-pairing/s2-over-ip-pairing.yaml b/src/s2python/specification/s2-pairing/s2-over-ip-pairing.yaml deleted file mode 100644 index e3bf691..0000000 --- a/src/s2python/specification/s2-pairing/s2-over-ip-pairing.yaml +++ /dev/null @@ -1,136 +0,0 @@ -openapi: 3.0.3 -info: - version: "0.1" - title: s2-over-ip pairing and connection initiation - description: "Description of the pairing process over IP for S2" -paths: - /requestPairing: - post: - description: Initiate pairing - requestBody: - description: TODO - content: - application/json: - schema: - $ref: '#/components/schemas/PairingRequest' - responses: - '200': - description: TODO - content: - application/json: - schema: - $ref: '#/components/schemas/PairingResponse' - /requestConnection: - post: - description: TODO - requestBody: - description: TODO - content: - application/json: - schema: - $ref: '#/components/schemas/ConnectionRequest' - responses: - '200': - description: TODO - content: - application/json: - schema: - $ref: '#/components/schemas/ConnectionDetails' -components: - schemas: - PairingInfo: - type: object - properties: - pairingUri: - type: string - format: uri - token: - $ref: "#/components/schemas/PairingToken" - validUntil: - type: string - format: date-time - PairingRequest: - type: object - properties: - token: - $ref: "#/components/schemas/PairingToken" - publicKey: - type: string - format: byte - s2ClientNodeId: - type: string - format: uuid - s2ClientNodeDescription: - $ref: "#/components/schemas/S2NodeDescription" - supportedProtocols: - type: array - items: - $ref: "#/components/schemas/Protocols" - PairingResponse: - type: object - properties: - s2ServerNodeId: - type: string - format: uuid - serverNodeDescription: - $ref: "#/components/schemas/S2NodeDescription" - requestConnectionUri: - type: string - format: uri - ConnectionRequest: - type: object - properties: - s2ClientNodeId: - type: string - format: uuid - supportedProtocols: - type: array - items: - $ref: "#/components/schemas/Protocols" - ConnectionDetails: - type: object - properties: - selectedProtocol: - $ref: "#/components/schemas/Protocols" - challenge: - type: string - format: byte - connectionUri: - type: string - format: uri - S2NodeDescription: - type: object - description: TODO nog even over nadenken - properties: - brand: - type: string - logoUri: - type: string - format: uri - type: - type: string - modelName: - type: string - userDefinedName: - type: string - role: - $ref: "#/components/schemas/S2Role" - deployment: - $ref: "#/components/schemas/Deployment" - Protocols: - type: string - enum: - - WebSocketSecure - S2Role: - type: string - enum: - - CEM - - RM - Deployment: - type: string - enum: - - WAN - - LAN - PairingToken: - type: string - pattern: "^[0-9a-zA-Z]{32}$" \ No newline at end of file From ae9a506cf1076c89c6651cf67994b2a99c27b998 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Tue, 15 Apr 2025 17:02:46 +0200 Subject: [PATCH 4/5] style: pylint W0107: Unnecessary pass statement (unnecessary-pass) Signed-off-by: F.N. Claessen --- src/s2python/authorization/client.py | 6 ------ 1 file changed, 6 deletions(-) diff --git a/src/s2python/authorization/client.py b/src/s2python/authorization/client.py index bf76be5..90e3a17 100644 --- a/src/s2python/authorization/client.py +++ b/src/s2python/authorization/client.py @@ -17,7 +17,6 @@ def build_connection_request(self) -> Dict: Build the payload for the ConnectionRequest schema. Returns a dictionary with keys: s2ClientNodeId, supportedProtocols. """ - pass @abstractmethod def execute_connection_request(self, request_data: Dict) -> Dict: @@ -26,7 +25,6 @@ def execute_connection_request(self, request_data: Dict) -> Dict: Implementations should send the request_data to the endpoint and return the JSON response as a dictionary. """ - pass @abstractmethod def handle_connection_response(self, response_data: Dict) -> Any: @@ -34,7 +32,6 @@ def handle_connection_response(self, response_data: Dict) -> Any: Process the ConnectionDetails response (e.g., extract challenge and connection URI). The response_data contains keys: selectedProtocol, challenge, connectionUri. """ - pass class AbstractPairingClient(ABC): @@ -53,7 +50,6 @@ def build_pairing_request(self) -> Dict: Returns a dictionary with keys: token, publicKey, s2ClientNodeId, s2ClientNodeDescription, supportedProtocols. """ - pass @abstractmethod def execute_pairing_request(self, request_data: Dict) -> Dict: @@ -62,7 +58,6 @@ def execute_pairing_request(self, request_data: Dict) -> Dict: Implementations should send the request_data to the endpoint and return the JSON response as a dictionary. """ - pass @abstractmethod def handle_pairing_response(self, response_data: Dict) -> Any: @@ -70,4 +65,3 @@ def handle_pairing_response(self, response_data: Dict) -> Any: Process the PairingResponse (e.g., extract server details). The response_data contains keys: s2ServerNodeId, serverNodeDescription, requestConnectionUri. """ - pass From e72847e6e16815c267d5d7cb20acfe33e1ed2718 Mon Sep 17 00:00:00 2001 From: Felix Claessen <30658763+Flix6x@users.noreply.github.com> Date: Tue, 15 Apr 2025 19:17:35 +0200 Subject: [PATCH 5/5] Fix/use datamodel codegen (#105) * docs: add instruction for generating gen_s2_pairing.py Signed-off-by: F.N. Claessen * feat: add generated classes for S2 pairing & authentication Signed-off-by: F.N. Claessen --------- Signed-off-by: F.N. Claessen --- ci/generate_s2.sh | 1 + src/s2python/generated/gen_s2_pairing.py | 189 ++++++----------------- 2 files changed, 44 insertions(+), 146 deletions(-) diff --git a/ci/generate_s2.sh b/ci/generate_s2.sh index f1ee694..0df831c 100755 --- a/ci/generate_s2.sh +++ b/ci/generate_s2.sh @@ -3,3 +3,4 @@ . .venv/bin/activate datamodel-codegen --input specification/openapi.yml --input-file-type openapi --output-model-type pydantic_v2.BaseModel --output src/s2python/generated/gen_s2.py --use-one-literal-as-default +# datamodel-codegen --input specification/s2-over-ip-pairing.yaml --input-file-type openapi --output-model-type pydantic_v2.BaseModel --output src/s2python/generated/gen_s2_pairing.py --use-one-literal-as-default diff --git a/src/s2python/generated/gen_s2_pairing.py b/src/s2python/generated/gen_s2_pairing.py index 32979b9..15a6b05 100644 --- a/src/s2python/generated/gen_s2_pairing.py +++ b/src/s2python/generated/gen_s2_pairing.py @@ -1,173 +1,70 @@ -""" -Generated classes based on s2-over-ip-pairing.yaml OpenAPI schema. -This file is auto-generated and should not be modified directly. -""" - -import uuid -from dataclasses import dataclass -from datetime import datetime -from enum import Enum, auto -from typing import List, Optional +# generated by datamodel-codegen: +# filename: s2-over-ip-pairing.yaml +# timestamp: 2025-04-15T14:41:29+00:00 +from __future__ import annotations -class Protocols(str, Enum): - """Supported protocol types.""" +from enum import Enum +from typing import List, Optional +from uuid import UUID - WebSocketSecure = "WebSocketSecure" +from pydantic import AnyUrl, AwareDatetime, BaseModel, RootModel, constr -class S2Role(str, Enum): - """Roles in the S2 protocol.""" +class Protocols(Enum): + WebSocketSecure = 'WebSocketSecure' - CEM = "CEM" - RM = "RM" +class S2Role(Enum): + CEM = 'CEM' + RM = 'RM' -class Deployment(str, Enum): - """Deployment types.""" - WAN = "WAN" - LAN = "LAN" +class Deployment(Enum): + WAN = 'WAN' + LAN = 'LAN' -@dataclass -class S2NodeDescription: - """Description of an S2 node.""" +class PairingToken(RootModel[constr(pattern=r'^[0-9a-zA-Z]{32}$')]): + root: constr(pattern=r'^[0-9a-zA-Z]{32}$') - brand: Optional[str] = None - logoUri: Optional[str] = None - type: Optional[str] = None - modelName: Optional[str] = None - userDefinedName: Optional[str] = None - role: Optional[S2Role] = None - deployment: Optional[Deployment] = None +class PairingInfo(BaseModel): + pairingUri: Optional[AnyUrl] = None + token: Optional[PairingToken] = None + validUntil: Optional[AwareDatetime] = None -class PairingToken(str): - """A token used for pairing. - - Must match pattern: ^[0-9a-zA-Z]{32}$ - """ - - def __new__(cls, content: str): - import re - if not re.match(r"^[0-9a-zA-Z]{32}$", content): - raise ValueError("PairingToken must be 32 alphanumeric characters") - return super().__new__(cls, content) +class ConnectionRequest(BaseModel): + s2ClientNodeId: Optional[UUID] = None + supportedProtocols: Optional[List[Protocols]] = None -@dataclass -class PairingInfo: - """Information about a pairing.""" +class ConnectionDetails(BaseModel): + selectedProtocol: Optional[Protocols] = None + challenge: Optional[str] = None + connectionUri: Optional[AnyUrl] = None - pairingUri: Optional[str] = None - token: Optional[PairingToken] = None - validUntil: Optional[datetime] = None +class S2NodeDescription(BaseModel): + brand: Optional[str] = None + logoUri: Optional[AnyUrl] = None + type: Optional[str] = None + modelName: Optional[str] = None + userDefinedName: Optional[str] = None + role: Optional[S2Role] = None + deployment: Optional[Deployment] = None -@dataclass -class PairingRequest: - """Request to initiate pairing.""" +class PairingRequest(BaseModel): token: Optional[PairingToken] = None - publicKey: Optional[bytes] = None - s2ClientNodeId: Optional[uuid.UUID] = None + publicKey: Optional[str] = None + s2ClientNodeId: Optional[UUID] = None s2ClientNodeDescription: Optional[S2NodeDescription] = None supportedProtocols: Optional[List[Protocols]] = None -@dataclass -class PairingResponse: - """Response to a pairing request.""" - - s2ServerNodeId: Optional[uuid.UUID] = None +class PairingResponse(BaseModel): + s2ServerNodeId: Optional[UUID] = None serverNodeDescription: Optional[S2NodeDescription] = None - requestConnectionUri: Optional[str] = None - - -@dataclass -class ConnectionRequest: - """Request to establish a connection.""" - - s2ClientNodeId: Optional[uuid.UUID] = None - supportedProtocols: Optional[List[Protocols]] = None - - -@dataclass -class ConnectionDetails: - """Details for establishing a connection.""" - - selectedProtocol: Optional[Protocols] = None - challenge: Optional[bytes] = None - connectionUri: Optional[str] = None - - -# Serialization/Deserialization functions - - -def _is_dataclass_instance(obj): - """Check if an object is a dataclass instance.""" - from dataclasses import is_dataclass - - return is_dataclass(obj) and not isinstance(obj, type) - - -def to_dict(obj): - """Convert a dataclass instance to a dictionary.""" - if isinstance(obj, datetime): - return obj.isoformat() - elif isinstance(obj, uuid.UUID): - return str(obj) - elif isinstance(obj, bytes): - import base64 - - return base64.b64encode(obj).decode("ascii") - elif isinstance(obj, Enum): - return obj.value - elif isinstance(obj, list): - return [to_dict(item) for item in obj] - elif _is_dataclass_instance(obj): - result = {} - for field in obj.__dataclass_fields__: - value = getattr(obj, field) - if value is not None: - result[field] = to_dict(value) - return result - else: - return obj - - -def from_dict(cls, data): - """Create a dataclass instance from a dictionary.""" - if data is None: - return None - - if cls is datetime: - return datetime.fromisoformat(data) - elif cls is uuid.UUID: - return uuid.UUID(data) - elif cls is bytes: - import base64 - - return base64.b64decode(data.encode("ascii")) - elif issubclass(cls, Enum): - return cls(data) - elif issubclass(cls, PairingToken): - return PairingToken(data) - elif hasattr(cls, "__dataclass_fields__"): - fieldtypes = cls.__annotations__ - instance_data = {} - - for field, field_type in fieldtypes.items(): - if field in data and data[field] is not None: - # Handle List[Type] annotations - if hasattr(field_type, "__origin__") and field_type.__origin__ is list: - item_type = field_type.__args__[0] - instance_data[field] = [from_dict(item_type, item) for item in data[field]] - else: - instance_data[field] = from_dict(field_type, data[field]) - - return cls(**instance_data) - else: - return data + requestConnectionUri: Optional[AnyUrl] = None