epicsarchiver.statistics.services.channelfinder
===============================================

.. py:module:: epicsarchiver.statistics.services.channelfinder

.. autoapi-nested-parse::

   Minimal Channel Finder interface for calculating archiver statistics.



Attributes
----------

.. autoapisummary::

   epicsarchiver.statistics.services.channelfinder.LOG


Exceptions
----------

.. autoapisummary::

   epicsarchiver.statistics.services.channelfinder.ChannelFinderRequestError


Classes
-------

.. autoapisummary::

   epicsarchiver.statistics.services.channelfinder.Channel
   epicsarchiver.statistics.services.channelfinder.ScrollChannels
   epicsarchiver.statistics.services.channelfinder.ChannelFinder


Functions
---------

.. autoapisummary::

   epicsarchiver.statistics.services.channelfinder._channel_list_to_dict


Module Contents
---------------

.. py:data:: LOG
   :type:  logging.Logger

.. py:class:: Channel

   Outline class of a channel finder channel data.

   :returns: includes name, properties and tags of a channel.
   :rtype: Channel


   .. py:attribute:: name
      :type:  str


   .. py:attribute:: properties
      :type:  dict[str, str]


   .. py:attribute:: tags
      :type:  list[str]


   .. py:method:: from_json(json: dict[str, Any]) -> Channel
      :classmethod:


      Convert from json direct from channel finder to a "Channel".

      :param json: input json dictionary
      :type json: dict

      :returns: corresponding channel
      :rtype: Channel



   .. py:method:: __hash__() -> int

      Calculates a has of a "Channel".

      :returns: hash of channel
      :rtype: int



.. py:class:: ScrollChannels

   Outline class of a channel finder channel data.

   :returns: includes name, properties and tags of a channel.
   :rtype: Channel


   .. py:attribute:: scroll_id
      :type:  str | None


   .. py:attribute:: channels
      :type:  list[Channel]


   .. py:method:: from_json(json: dict[str, Any]) -> ScrollChannels
      :classmethod:


      Convert from json direct from channel finder to a "Channel".

      :param json: input json dictionary
      :type json: dict

      :returns: corresponding channel
      :rtype: Channel



.. py:exception:: ChannelFinderRequestError

   Bases: :py:obj:`BaseException`


   Exception raised when error running requests against the channelfinder.


   .. py:attribute:: url
      :type:  str


   .. py:attribute:: params
      :type:  dict[str, str]


   .. py:attribute:: session_info
      :type:  str


.. py:function:: _channel_list_to_dict(channels: list[Channel], pvs_set: set[str] | None) -> dict[str, Channel]

.. py:class:: ChannelFinder(hostname: str = 'localhost')

   Bases: :py:obj:`epicsarchiver.common.async_service.ServiceClient`


   Minimal Channel Finder client.

   Hold a session to the Channel Finder web application.

   :param hostname: Channel Finder url [default: localhost]

   Examples:

   .. code-block:: python

       from epicsarchiver.channelfinder import ChannelFinder

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


   .. py:attribute:: hostname
      :value: 'localhost'



   .. py:method:: _fetch_channels_scroll(params: dict[str, str]) -> list[Channel]
      :async:



   .. py:method:: _fetch_channels(params: dict[str, str]) -> list[Channel]
      :async:



   .. py:method:: __repr__() -> str

      String representation of Channel Finder.

      :returns: details including hostname of Channel Finder.
      :rtype: str



   .. py:method:: get_channels(pvs: set[str] | None, properties: dict[str, str] | None = None) -> list[Channel]
      :async:


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

      :param pvs: pv names
      :type pvs: list[str]
      :param properties: Properties to filter by
      :type properties: dict[str, str]

      :returns: list of matching channels
      :rtype: list[Channel]



   .. py:method:: get_channels_chunked(pvs: list[str] | None, properties: dict[str, str] | None = None, chunk_size: int = 10) -> dict[str, Channel]
      :async:


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

      :param pvs: list of pv names
      :type pvs: list[str]
      :param properties: Properties to filter by
      :type properties: dict[str, str]
      :param chunk_size: Number of pvs to search at once to submit to channelfinder
      :type chunk_size: int

      :returns: dict of matching channels
      :rtype: dict[str, Channel]



   .. py:method:: get_ioc_channels(ioc_name: str) -> list[Channel]
      :async:


      Get the list of channels with the specified ioc_name.

      :param ioc_name: name of the ioc

      :returns: dict of matching channels
      :rtype: dict[str, Channel]



   .. py:method:: get_all_alias_channels(pvs: list[str], ioc_name: str | None = None) -> dict[str, list[Channel]]
      :async:


      Get the list of channels aliases of pvs from channelfinder.

      :param pvs: list of pv names
      :type pvs: list[str]
      :param ioc_name: ioc to filter by
      :type ioc_name: str

      :returns: dict of matching channels to pv names
      :rtype: dict[str, list[Channel]]