`plaid.utils.process_client`¶

plaid.utils.process_client ¶

Class to use the /process capability of a server.

plaid.utils.process_client.PlaidClient ¶

PlaidClient(host, port)

Client for making requests to a PLAID process server.

Initialize a process server client.

Parameters:

host (str) –

Hostname or IP address of the process server.
port (int) –

Port number used by the process server.

Source code in plaid/utils/process_client.py

def __init__(self, host: str, port: int):
    """Initialize a process server client.

    Args:
        host: Hostname or IP address of the process server.
        port: Port number used by the process server.

    """
    self.host = host
    self.port = port
    self.endpoints = {
        "health": "/health",
        "process": "/process",
        "problem_definition": "/problem_definition",
        "infos": "/infos",
        "samples": "/samples",
    }
    self.protocol = "http"
    self.timeout = 100  # timeout for the response

plaid.utils.process_client.PlaidClient.check_connection ¶

check_connection()

Check if the server is healthy by querying the health endpoint.

Source code in plaid/utils/process_client.py

def check_connection(self) -> bool:
    """Check if the server is healthy by querying the health endpoint."""
    try:
        data = self._request_json("health", {}).get("status", "payload missing")
        if data != "ok":
            print("Server health check failed: status not ok")
            print(f"Received data: {data}")
            return False
        return True
    except Exception as e:
        print(f"Connection check failed: {e}")
        return False

plaid.utils.process_client.PlaidClient.process ¶

process(sample=None, **fields)

Send a single-sample request to the process endpoint.

The optional sample and any Sample passed as a keyword field are JSON-encoded automatically. Other keyword fields are encoded with the generic JSON codec, so NumPy arrays (and nested arrays in lists or dicts) are sent as portable base64 payloads. The server operation contract stays opaque to this client, which operates on a single sample at a time.

Parameters:

sample (Sample | None, default: None ) –

Optional inline sample, sent as the sample field.
**fields (Any, default: {} ) –

Additional request fields forwarded to the server. Any Sample value is JSON-encoded, and any NumPy array is encoded with the portable base64 schema.

Returns:

Sample –

The single Sample returned by the server.

Source code in plaid/utils/process_client.py

def process(self, sample: Sample | None = None, **fields: Any) -> Sample:
    """Send a single-sample request to the process endpoint.

    The optional ``sample`` and any ``Sample`` passed as a keyword field are
    JSON-encoded automatically. Other keyword fields are encoded with the
    generic JSON codec, so NumPy arrays (and nested arrays in lists or
    dicts) are sent as portable base64 payloads. The server operation
    contract stays opaque to this client, which operates on a single sample
    at a time.

    Args:
        sample: Optional inline sample, sent as the ``sample`` field.
        **fields: Additional request fields forwarded to the server. Any
            ``Sample`` value is JSON-encoded, and any NumPy array is encoded
            with the portable base64 schema.

    Returns:
        The single ``Sample`` returned by the server.

    """
    payload: dict[str, Any] = {
        key: sample_to_json_payload(value)
        if isinstance(value, Sample)
        else encode_json_value(value)
        for key, value in fields.items()
    }
    if sample is not None:
        payload["sample"] = sample_to_json_payload(sample)

    response = self._request_json("process", payload)
    return sample_from_json_payload(response["samples"][0])

plaid.utils.process_client.PlaidClient.problem_definition ¶

problem_definition()

Get the problem definition from the server.

Returns:

dict[str, Any] –

A dictionary containing the problem definition as provided by the server.

Source code in plaid/utils/process_client.py

def problem_definition(self) -> dict[str, Any]:
    """Get the problem definition from the server.

    Returns:
        A dictionary containing the problem definition as provided by the server.
    """
    return self._request_json("problem_definition", {})

plaid.utils.process_client.PlaidClient.infos ¶

infos()

Get the infos from the server.

Returns:

dict[str, Any] –

A dictionary containing the infos as provided by the server.

Source code in plaid/utils/process_client.py

def infos(self) -> dict[str, Any]:
    """Get the infos from the server.

    Returns:
        A dictionary containing the infos as provided by the server.
    """
    return self._request_json("infos", {})

plaid.utils.process_client.PlaidClient.samples ¶

samples(sample_ids, split)

Request samples from the server by sample IDs and split.

Parameters:

sample_ids (list[int]) –

A list of integers representing the IDs of the samples to request.
split (str) –

A string indicating the data split (e.g., "training", "validation", "test") from which to request the samples.

Returns:

list[Sample] –

A list of Sample objects corresponding to the requested sample IDs and split.

Source code in plaid/utils/process_client.py

def samples(self, sample_ids: list[int], split: str) -> list[Sample]:
    """Request samples from the server by sample IDs and split.

    Args:
        sample_ids: A list of integers representing the IDs of the samples to request.
        split: A string indicating the data split (e.g., "training", "validation", "test") from which to request the samples.

    Returns:
        A list of Sample objects corresponding to the requested sample IDs and split.
    """
    payload: dict[str, Any] = {"sample_ids": sample_ids, "split": split}
    response = self._request_json("samples", payload)
    return [
        sample_from_json_payload(sample_payload)
        for sample_payload in response["samples"]
    ]

plaid.utils.process_client¶