Skip to content

plaid.cli.serve

plaid.cli.serve

Serve PLAID datasets over a small HTTP API.

The plaid-serve command exposes local PLAID datasets to client tools and the ParaView plugin. It supports lightweight discovery routes (GET /health and GET /entry_points) plus JSON POST routes for dataset metadata, problem definitions, and samples.

plaid.cli.serve.ServeContext dataclass 

ServeContext(default_dataset_uri=None, stores=dict())

Serving context shared by all HTTP handlers.

plaid.cli.serve.ServeContext.resolve_dataset_uri 

resolve_dataset_uri(request)

Resolve dataset URI from a request or the server default.

Parameters:

  • request (dict[str, object]) –

    Parsed JSON payload.

Returns:

  • str –

    Dataset path/URI string.

Source code in plaid/cli/serve.py 
180
181
182
183
184
185
186
187
188
189
def resolve_dataset_uri(self, request: dict[str, object]) -> str:
    """Resolve dataset URI from a request or the server default.

    Args:
        request: Parsed JSON payload.

    Returns:
        Dataset path/URI string.
    """
    return _resolve_dataset_uri(request, self.default_dataset_uri)

plaid.cli.serve.ServeContext.get_sample_objects 

get_sample_objects(dataset_uri, split, sample_ids)

Return source samples from a PLAID dataset.

Parameters:

  • dataset_uri (str) –

    Dataset path/URI.

  • split (str | None) –

    Requested split.

  • sample_ids (list[int]) –

    Existing sample IDs.

Returns:

  • list[Sample] –

    PLAID sample objects.

Source code in plaid/cli/serve.py 
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def get_sample_objects(
    self,
    dataset_uri: str,
    split: str | None,
    sample_ids: list[int],
) -> list[Sample]:
    """Return source samples from a PLAID dataset.

    Args:
        dataset_uri: Dataset path/URI.
        split: Requested split.
        sample_ids: Existing sample IDs.

    Returns:
        PLAID sample objects.
    """
    store = self._get_store(dataset_uri)
    split_key = self._resolve_split(store, split)
    dataset = store.datasetdict[split_key]
    converter = store.converterdict[split_key]

    return [converter.to_plaid(dataset, sample_id) for sample_id in sample_ids]

plaid.cli.serve.ServeContext.get_infos staticmethod 

get_infos(dataset_uri)

Return dataset infos as a JSON-serializable dictionary.

Parameters:

  • dataset_uri (str) –

    Dataset path/URI.

Returns:

  • dict[str, object] –

    Serialized dataset infos.

Source code in plaid/cli/serve.py 
278
279
280
281
282
283
284
285
286
287
288
@staticmethod
def get_infos(dataset_uri: str) -> dict[str, object]:
    """Return dataset infos as a JSON-serializable dictionary.

    Args:
        dataset_uri: Dataset path/URI.

    Returns:
        Serialized dataset infos.
    """
    return cast(dict[str, object], load_infos_from_disk(dataset_uri).model_dump())

plaid.cli.serve.ServeContext.get_problem_definition staticmethod 

get_problem_definition(dataset_uri, name=None)

Return one problem definition JSON.

Parameters:

  • dataset_uri (str) –

    Dataset path/URI.

  • name (str | None, default: None ) –

    Optional problem definition name to select.

Returns:

  • dict[str, object] –

    Serialized problem definition.

Raises:

  • ValueError –

    If the requested definition is unavailable.

Source code in plaid/cli/serve.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
@staticmethod
def get_problem_definition(
    dataset_uri: str,
    name: str | None = None,
) -> dict[str, object]:
    """Return one problem definition JSON.

    Args:
        dataset_uri: Dataset path/URI.
        name: Optional problem definition name to select.

    Returns:
        Serialized problem definition.

    Raises:
        ValueError: If the requested definition is unavailable.
    """
    problem_definitions = load_problem_definitions_from_disk(dataset_uri)
    if name is not None:
        if name not in problem_definitions:
            raise ValueError(
                f"Problem definition {name!r} not found; available definitions: "
                f"{sorted(problem_definitions)}"
            )
        return cast(dict[str, object], problem_definitions[name].model_dump())

    if "PLAID_benchmark" in problem_definitions:
        return cast(
            dict[str, object],
            problem_definitions["PLAID_benchmark"].model_dump(),
        )
    if len(problem_definitions) == 1:
        problem_definition = next(iter(problem_definitions.values()))
        return cast(dict[str, object], problem_definition.model_dump())

    first_name = sorted(problem_definitions)[0]
    return cast(dict[str, object], problem_definitions[first_name].model_dump())

plaid.cli.serve.main 

main(argv=None)

Run HTTP server for PLAID dataset access.

Parameters:

  • argv (list[str] | None, default: None ) –

    Optional override of sys.argv[1:] for tests.

Returns:

  • int –

    Process exit code.

Source code in plaid/cli/serve.py 
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
def main(argv: list[str] | None = None) -> int:
    """Run HTTP server for PLAID dataset access.

    Args:
        argv: Optional override of ``sys.argv[1:]`` for tests.

    Returns:
        Process exit code.
    """
    args = _build_parser().parse_args(argv)
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )
    if args.ParaViewRun:
        os.environ["PLAID_PORT"] = str(args.port)
        from .paraview_plugin import run_paraview_with_plugin

        process = run_paraview_with_plugin()
    else:
        process = None

    context = ServeContext(default_dataset_uri=args.dataset)
    server = _ServeHTTPServer((str(args.host), int(args.port)), context)
    log.info("PLAID data API listening on http://%s:%s", args.host, args.port)

    if process is None:
        log.info("Running PLAID data API until interrupted")
        server.serve_forever()
    else:
        log.info("Running PLAID data API for ParaView")
        _run_server_until_process_exits(server, process)

    return 0