Add market offer scanning and GPU discovery example

examples/find-gpu-nodes.py

@@ -0,0 +1,41 @@
+import asyncio
+import json
+from tabulate import tabulate  # You may need to install this: pip install tabulate
+
+from golem.node import GolemNode
+
+
+async def do_scan():
+    golem = GolemNode(app_key="o")
+    await golem.start()
+
+    print("Scanning for GPU nodes...")
+
+    # Store the data for tabulation
+    nodes_data = []
+
+    async for it in golem.scan(constraints="(golem.runtime.name=vm-nvidia)", quick_scan=True):
+        provider_id = it.providerId
+        subnet = it.properties.get("golem.node.debug.subnet", "N/A")
+        name = it.properties.get("golem.node.id.name", "N/A")
+        gpu_model = it.properties.get("golem.!exp.gap-35.v1.inf.gpu.model", "N/A")
+
+        # Add data to our list
+        nodes_data.append([provider_id, name, gpu_model, subnet])
+
+    # Print elegant table if we found any nodes
+    if nodes_data:
+        headers = ["Provider ID", "Node Name", "GPU Model", "Subnet"]
+        print("\nFound GPU Nodes:")
+        print(tabulate(nodes_data, headers=headers, tablefmt="pretty"))
+        print(f"\nTotal nodes found: {len(nodes_data)}")
+    else:
+        print("\nNo GPU nodes found.")
+
+    print("\nScan completed.")
+    await golem.aclose()
+
+
+if __name__ == "__main__":
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(do_scan())

golem/node/node.py

@@ -1,10 +1,13 @@
 import asyncio
+import json
 from collections import defaultdict
 from datetime import datetime, timezone
 from decimal import Decimal
-from typing import Any, DefaultDict, Dict, Iterable, List, Optional, Set, Type, Union
+from typing import Any, AsyncIterator, DefaultDict, Dict, Iterable, List, Optional, Set, Type, Union
 from uuid import uuid4
 
+from ya_market import ApiClient
+
 from golem.event_bus import EventBus
 from golem.event_bus.in_memory import InMemoryEventBus
 from golem.node.events import SessionStarted, ShutdownFinished, ShutdownStarted
@@ -26,6 +29,7 @@
     Resource,
     TResource,
 )
+from golem.resources.scan import ScanOfferEvent
 from golem.utils.logging import get_trace_id_name, set_trace_id
 from golem.utils.low import ApiConfig, ApiFactory
 
@@ -411,6 +415,69 @@ def all_resources(self, cls: Type[TResource]) -> List[TResource]:
         """Return all known resources of a given type."""
         return list(self._resources[cls].values())  # type: ignore
 
+    async def scan(
+        self, *, timeout: int = 300, constraints: Optional[str] = None, quick_scan=False
+    ) -> AsyncIterator[ScanOfferEvent]:
+        """Scan the market for proposals (offers).
+
+        Args:
+            timeout: Time in seconds for how long to keep the subscription alive.
+                Defaults to 300 seconds (5 minutes).
+            constraints: Optional constraints expression to filter offers.
+            quick_scan: If True, returns only the initial batch of proposals without
+                continuous polling. Defaults to False.
+
+        Yields:
+            Proposal objects for each offer found in the market.
+        """
+        scan_id = None
+        try:
+            # Create scan subscription
+            api: ApiClient = self._ya_market_api
+            body = {"timeout": timeout, "type": "offer"}
+            if constraints is not None:
+                body["constraints"] = constraints
+
+            (scan_id, rc, headers) = await api.call_api(
+                "/scan", "POST", body=body, response_type=str
+            )
+
+            while True:
+                # Fetch events in batches
+                (events_bytes, rc, headers) = await api.call_api(
+                    "/scan/{scan_id}/events",
+                    "GET",
+                    path_params={"scan_id": scan_id},
+                    query_params=[("timeout", 5), ("maxEvents", 100)],
+                    response_type=bytes,
+                )
+                events = []
+                if rc == 200 and events_bytes:
+                    events = (
+                        json.loads(events_bytes) if type(events_bytes) is bytes else events_bytes
+                    )
+                    # print('events=', events)
+
+                if not events:
+                    if quick_scan:
+                        break
+                    await asyncio.sleep(0.5)
+                    continue
+
+                for event in events:
+                    yield ScanOfferEvent.model_validate(event)
+
+        finally:
+            # Ensure we clean up the subscription
+            if scan_id:
+                try:
+                    await api.call_api(
+                        "/scan/{scan_id}", "DELETE", path_params={"scan_id": scan_id}
+                    )
+                except Exception:
+                    # Best effort cleanup
+                    pass
+
     def __str__(self) -> str:
         lines = [
             f"{type(self).__name__}(",

golem/payload/mixins.py

@@ -2,7 +2,7 @@
 import enum
 import inspect
 from dataclasses import Field
-from typing import Any
+from typing import Any, Type, cast
 
 from golem.utils.typing import match_type_union_aware
 
@@ -30,14 +30,20 @@ def _deserialize_value(cls, value: Any, field: Field) -> Any:
 
         Intended to be overriden with additional type serialisation methods.
         """
-        if matched_type := match_type_union_aware(
-            field.type, lambda t: inspect.isclass(t) and issubclass(t, datetime.datetime)
-        ):
-            return matched_type.fromtimestamp(int(float(value) * 0.001), datetime.timezone.utc)
-
-        if matched_type := match_type_union_aware(
-            field.type, lambda t: inspect.isclass(t) and issubclass(t, enum.Enum)
-        ):
-            return matched_type(value)
+        field_type = field.type
+        # Only attempt match_type_union_aware if field_type is a proper Type
+        if isinstance(field_type, type) or hasattr(field_type, "__origin__"):
+            # Use cast to help mypy understand the type
+            type_obj = cast(Type[Any], field_type)
+
+            if matched_type := match_type_union_aware(
+                type_obj, lambda t: inspect.isclass(t) and issubclass(t, datetime.datetime)
+            ):
+                return matched_type.fromtimestamp(int(float(value) * 0.001), datetime.timezone.utc)
+
+            if matched_type := match_type_union_aware(
+                type_obj, lambda t: inspect.isclass(t) and issubclass(t, enum.Enum)
+            ):
+                return matched_type(value)
 
         return value

golem/resources/scan.py

@@ -0,0 +1,14 @@
+from datetime import datetime
+from typing import Any, Dict
+
+from pydantic import BaseModel, Field
+
+
+class ScanOfferEvent(BaseModel):
+    """Represents a proposal (offer) from the market."""
+
+    properties: Dict[str, Any] = Field(description="Provider capabilities and specifications")
+    constraints: str = Field(description="Constraints expression")
+    offerId: str = Field(description="Unique identifier for the offer")
+    providerId: str = Field(description="provider node id")
+    timestamp: datetime = Field(description="When event is received")

pyproject.toml

@@ -18,7 +18,7 @@ requires = ["poetry_core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry.dependencies]
-python = "^3.8.1"
+python = "^3.10"
 
 # CLI
 prettytable = "^3.4.1"
@@ -32,6 +32,7 @@ async-exit-stack = "1.0.1"
 textx = "^3.1.1"
 
 setuptools = "*"  # textx external dependency
+pydantic = "^2.11.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7"
@@ -55,6 +56,7 @@ sphinx-autodoc-typehints = "^1"
 sphinx-rtd-theme = "^1"
 
 poethepoet = "^0.8"
+tabulate = "^0.9.0"
 
 [tool.poe.tasks]
 checks = {sequence = ["checks_codestyle", "checks_typing", "checks_license"], help = "Run all available code checks"}
@@ -83,6 +85,8 @@ sphinx = {cmd = "sphinx-build docs/sphinx/ build -E", help = "Build Sphinx docs"
 ya-client-activity = "0.1.0"
 ya-client-payment = "0.1.0"
 ya-market = "0.1.0"
+attrs = "25.3.0"
+typing-extensions = "4.13.0"
 
 [tool.liccheck]
 authorized_licenses = [
@@ -109,6 +113,8 @@ unauthorized_licenses = [
     "gpl v3"
 ]
 
+
+
 [tool.flake8]
 max-line-length = 100
 extend-ignore = [