Channel#

class nominal.core.Channel(name: str, data_source: str, data_type: ChannelDataType | None, unit: str | None, description: str | None)#

Bases: RefreshableGrpcMixin[ChannelMetadata]

Metadata for working with channels.

name: str#
data_source: str#
data_type: ChannelDataType | None#
unit: str | None#
description: str | None#
get_available_tags(start_time: str | datetime | int | None = None, end_time: str | datetime | int | None = None, initial_tags: Mapping[str, str] | None = None) → Mapping[str, set[str]]#

Return available tag key-values for the channel over the given time period

Parameters:

  • start_time – Start time to start scanning for tags from (defaults to the beginning of time)

  • end_time – End time to stop scanning for tags until (defaults to the end of time)

  • initial_tags – Initial tags to filter results by

Returns:

Mapping of all tag names to all tag values present

Example

Given data with the following tags per-point within the specified time range:

{"tag_a": "123", "tag_b": "xyz"}
{"tag_a": "123", "tag_b": "abc"}
{"tag_a": "234", "tag_b": "qqq"}

With an initial filter of {"tag_a": "123"}, we would return:

{"tag_a": set(["123"]), "tag_b": set(["abc", "xyz"])}

Because we filtered data to only include data where “tag_a” is “123”

Note

It is not accurate to say that the cartesian product of all returned tag key-value pairs is present in the data, only that for each unique key-value pair present, that specific tag key/value pair is present on at least one point in the data.

Note

This may be used to determine if the given set of initial tags fully constrains data in a channel over a given timespan by checking to see if the length of all of the tag-value sets is 1.

refresh() → Self#
search_logs(*, tags: Mapping[str, str] | None = None, regex_match: str, start: str | datetime | int | None = None, end: str | datetime | int | None = None) → Iterable[LogPoint]#
search_logs(*, tags: Mapping[str, str] | None = None, insensitive_match: str, start: str | datetime | int | None = None, end: str | datetime | int | None = None) → Iterable[LogPoint]
search_logs(*, tags: Mapping[str, str] | None = None, start: str | datetime | int | None = None, end: str | datetime | int | None = None) → Iterable[LogPoint]

Yields logpoints from the current channel that match the provided arguments

Parameters:

  • regex_match – If provided, a regex match to filter potential log messages by.

    Note: Must not be present with insensitive_match.

  • insensitive_match – If provided, a case insensitive string that yielded logs match exactly.

    Note: Must not be present with regex_match.

  • tags – Tags to filter logs from the channel with

  • start – Timestamp to start yielding results from. If not present, searches starting from unix epoch

  • end – Timestamp after which to stop yielding results from. If not present, searches until end of time.

update(*, description: str | None = None, unit: UnitLike | _NotProvided = _NotProvided()) → Self#

Replace channel metadata within Nominal, and updates / returns the local instance.

Only the metadata passed in will be replaced, the rest will remain untouched.

Parameters:

  • description – Human-readable description of data within the channel

  • unit – Unit symbol to apply to the channel. If unit is a string or a Unit, this will update the unit symbol for the channel. If unit is None, this will clear the unit symbol for the channel. If not provided (or _NotProvided), this will leave the unit unaffected.

    Note: This is in contrast to other fields in other update() calls where None is treated as a “no-op”.