epicsarchiver.statistics.services.channelfinder

Minimal Channel Finder interface for calculating archiver statistics.

Attributes

LOG

Exceptions

ChannelFinderRequestError

Exception raised when error running requests against the channelfinder.

Classes

`Channel`	Outline class of a channel finder channel data.
`ScrollChannels`	Outline class of a channel finder channel data.
`ChannelFinder`	Minimal Channel Finder client.

Functions

_channel_list_to_dict(→ dict[str, Channel])

Module Contents

epicsarchiver.statistics.services.channelfinder.LOG: logging.Logger[source]

class epicsarchiver.statistics.services.channelfinder.Channel[source]

Outline class of a channel finder channel data.

Returns:: includes name, properties and tags of a channel.
Return type:: Channel

name: str[source]

properties: dict[str, str][source]

tags: list[str][source]

classmethod from_json(json: dict[str, Any]) → Channel[source]

Convert from json direct from channel finder to a “Channel”.

Parameters:: json (dict) – input json dictionary
Returns:: corresponding channel
Return type:: Channel

__hash__() → int[source]

Calculates a has of a “Channel”.

Returns:: hash of channel
Return type:: int

class epicsarchiver.statistics.services.channelfinder.ScrollChannels[source]

Outline class of a channel finder channel data.

Returns:: includes name, properties and tags of a channel.
Return type:: Channel

scroll_id: str | None[source]

channels: list[Channel][source]

classmethod from_json(json: dict[str, Any]) → ScrollChannels[source]

Convert from json direct from channel finder to a “Channel”.

Parameters:: json (dict) – input json dictionary
Returns:: corresponding channel
Return type:: Channel

exception epicsarchiver.statistics.services.channelfinder.ChannelFinderRequestError[source]

Bases: BaseException

Exception raised when error running requests against the channelfinder.

url: str[source]

params: dict[str, str][source]

session_info: str[source]

epicsarchiver.statistics.services.channelfinder._channel_list_to_dict(channels: list[Channel], pvs_set: set[str] | None) → dict[str, Channel][source]

class epicsarchiver.statistics.services.channelfinder.ChannelFinder(hostname: str = 'localhost')[source]

Bases: epicsarchiver.common.async_service.ServiceClient

Minimal Channel Finder client.

Hold a session to the Channel Finder web application.

Parameters:: hostname – Channel Finder url [default: localhost]

Examples:

from epicsarchiver.channelfinder import ChannelFinder

channelfinder = ChannelFinder("channelfinder.tn.esss.lu.se")
channel = channelfinder.get_channels(["AccPSS::FBIS-BP_A"])

hostname = 'localhost'[source]

async _fetch_channels_scroll(params: dict[str, str]) → list[Channel][source]

async _fetch_channels(params: dict[str, str]) → list[Channel][source]

__repr__() → str[source]

String representation of Channel Finder.

Returns:: details including hostname of Channel Finder.
Return type:: str

async get_channels(pvs: set[str] | None, properties: dict[str, str] | None = None) → list[Channel][source]

Get the list of channels matching the pv name from channelfinder.

Parameters:

pvs (list[str]) – pv names
properties (dict[str, str]) – Properties to filter by

Returns:

list of matching channels

Return type:

list[Channel]

async get_channels_chunked(pvs: list[str] | None, properties: dict[str, str] | None = None, chunk_size: int = 10) → dict[str, Channel][source]

Get the list of channels matching the pv names from channelfinder.

Parameters:

pvs (list[str]) – list of pv names
properties (dict[str, str]) – Properties to filter by
chunk_size (int) – Number of pvs to search at once to submit to channelfinder

Returns:

dict of matching channels

Return type:

dict[str, Channel]

async get_ioc_channels(ioc_name: str) → list[Channel][source]

Get the list of channels with the specified ioc_name.

Parameters:: ioc_name – name of the ioc
Returns:: dict of matching channels
Return type:: dict[str, Channel]

async get_all_alias_channels(pvs: list[str], ioc_name: str | None = None) → dict[str, list[Channel]][source]

Get the list of channels aliases of pvs from channelfinder.

Parameters:

pvs (list[str]) – list of pv names
ioc_name (str) – ioc to filter by

Returns:

dict of matching channels to pv names

Return type:

dict[str, list[Channel]]