From 74a8d746732332deb6deb3597ceb7c1182fb18f2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Przemys=C5=82aw=20Krzysztof=20Rekucki?= Date: Thu, 27 Mar 2025 20:41:11 +0100 Subject: [PATCH] feat: Add scanning functionality and update dependencies Introduce a `scan` method in `GolemNode` to query market offers based on constraints, - Added a sample script (`examples/find-gpu-nodes.py`) for scanning and displaying - python >= 3.10 - pydantic --- examples/find-gpu-nodes.py | 41 ++++++++++++++++++++++ golem/node/node.py | 69 +++++++++++++++++++++++++++++++++++++- golem/payload/mixins.py | 26 ++++++++------ golem/resources/scan.py | 14 ++++++++ pyproject.toml | 8 ++++- 5 files changed, 146 insertions(+), 12 deletions(-) create mode 100644 examples/find-gpu-nodes.py create mode 100644 golem/resources/scan.py diff --git a/examples/find-gpu-nodes.py b/examples/find-gpu-nodes.py new file mode 100644 index 00000000..8841b14c --- /dev/null +++ b/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()) diff --git a/golem/node/node.py b/golem/node/node.py index 55cdd40a..0dad1f4d 100644 --- a/golem/node/node.py +++ b/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__}(", diff --git a/golem/payload/mixins.py b/golem/payload/mixins.py index 04b73f3d..b074503d 100644 --- a/golem/payload/mixins.py +++ b/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 diff --git a/golem/resources/scan.py b/golem/resources/scan.py new file mode 100644 index 00000000..83b49121 --- /dev/null +++ b/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") diff --git a/pyproject.toml b/pyproject.toml index 644e1423..f34108c2 100644 --- a/pyproject.toml +++ b/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 = [