poly_lithic.src.interfaces package🔗

Submodules🔗

Module contents🔗

Interfaces module.

class poly_lithic.src.interfaces.BaseInterface(config)[source]🔗

Bases: ABC

abstractmethod monitor(name, handler, **kwargs)[source]🔗

abstractmethod get(name, **kwargs)[source]🔗

abstractmethod put(name, value, **kwargs)[source]🔗

abstractmethod put_many(data, **kwargs)[source]🔗

abstractmethod get_many(data, **kwargs)[source]🔗

abstractmethod get_outputs()[source]🔗

Get a list of outputs that we should publish to

Return type:: list[str]

abstractmethod get_inputs()[source]🔗

Get a list of inputs that we should publish to

Return type:: list[str]

class poly_lithic.src.interfaces.BaseDataInterface(config, **kwargs)[source]🔗

Bases: ABC

abstractmethod load(**kwargs)[source]🔗

abstractmethod save(data, **kwargs)[source]🔗

class poly_lithic.src.interfaces.h5dfInterface(config)[source]🔗

Bases: BaseDataInterface

load(**kwargs)[source]🔗

save(data, **kwargs)[source]🔗

monitor(handler, **kwargs)[source]🔗

class poly_lithic.src.interfaces.SimplePVAInterface(config)[source]🔗

Bases: BaseInterface

monitor(handler, **kwargs)[source]🔗

get(name, **kwargs)[source]🔗

put(name, value, **kwargs)[source]🔗

put_many(data, **kwargs)[source]🔗

get_many(data, **kwargs)[source]🔗

close()[source]🔗

get_outputs()[source]🔗: Get a list of outputs that we should publish to

get_inputs()[source]🔗: Get a list of inputs that we should publish to

class poly_lithic.src.interfaces.SimplePVAInterfaceServer(config)[source]🔗

Bases: SimplePVAInterface

Simple PVA integfcae with a server rather than just a client, this will host the PVs provided in the config

close()[source]🔗

put(name, value, **kwargs)[source]🔗

get(name, **kwargs)[source]🔗

put_many(data, **kwargs)[source]🔗

get_many(data, **kwargs)[source]🔗

class poly_lithic.src.interfaces.K2EGInterface(config)[source]🔗

Bases: BaseInterface

monitor(handler, **kwargs)[source]🔗

get(name, **kwargs)[source]🔗

put(name, value, **kwargs)[source]🔗

put_many(data, **kwargs)[source]🔗

get_many(data, **kwargs)[source]🔗

close()[source]🔗

get_inputs()[source]🔗: Get a list of inputs that we should publish to

get_outputs()[source]🔗: Get a list of outputs that we should publish to

class poly_lithic.src.interfaces.SimpleFastAPIInterfaceServer(config)[source]🔗

Bases: BaseInterface

HTTP interface backed by FastAPI with an in-memory job queue.

Registered as "fastapi_server" so it can be referenced in the YAML config as type: "interface.fastapi_server".

Parameters:: config (dict)

get(name, **kwargs)[source]🔗

Return type:: tuple[str, dict]
Parameters:: name (str)

put(name, value, **kwargs)[source]🔗

Return type:

None

Parameters:

name (str)
value (Any)

get_many(data, **kwargs)[source]🔗

Dual-return method.

consume_jobs=True with queued jobs → dequeue all, mark running, return a list of dicts (one per job with input snapshots).
Otherwise → return a single dict mapping requested names to values.

Return type:: dict | list[dict]

put_many(data, **kwargs)[source]🔗

Write multiple variables.

If any struct’s metadata contains a trace.job_id matching a running job, that job is transitioned to completed.

Return type:: None
Parameters:: data (dict)

get_inputs()[source]🔗

Get a list of inputs that we should publish to

Return type:: list[str]

get_outputs()[source]🔗

Get a list of outputs that we should publish to

Return type:: list[str]

monitor(handler, **kwargs)[source]🔗

Return type:: bool

close()[source]🔗: Shutdown the embedded uvicorn server (if running).