Sensor Tag#

Sensor is a time series sequence from a data source. The purpose of the sensor tag is to identify the unique time series within this data source.

class gordo_core.sensor_tag.SensorTag(name: str, **kwargs: str | None)[source]#

Bases: object

Representation of the sensor tag. It contains the sensor tag name and additional fields/metadata.

Example

>>> SensorTag("Test tag", field1="value1", field2="value2")
SensorTag("Test tag",field1="value1",field2="value2")
Parameters:

  • name – Sensor tag name. Required field and should be unique among all the tags.

  • kwargs – Additional fields.

mutate_fields(**kwargs: str | None) → SensorTag[source]#

Update SensorTag fields from kwargs. Instantiate a new object if needed.

Examples

>>> sensor_tag = SensorTag("Test tag", field1="value1", field2="value2")
>>> sensor_tag.mutate_fields(field1="value", field3="value3")
SensorTag("Test tag",field1="value",field2="value2",field3="value3")
Parameters:

kwargs – Fields to update.

to_json() → dict[str, Optional[str]][source]#

JSON representation of the sensor tag.

Example

>>> sensor_tag = SensorTag("Test tag", field1="value1", field2="value2")
>>> sensor_tag.to_json()
{'name': 'Test tag', 'field1': 'value1', 'field2': 'value2'}
Return type:

Sensor tag name along with additional fields in one dict.

exception gordo_core.sensor_tag.SensorTagNormalizationError[source]#

Bases: ValueError

Error indicating that something went wrong normalizing a sensor tag

gordo_core.sensor_tag.extract_tag_name(tag: str | SensorTag) → str[source]#

Get tag name.

Parameters:

tag – Sensor tag to extract.

gordo_core.sensor_tag.load_sensor_tag(sensor: dict[str, Optional[str]], required_fields: Iterable[str] | None = None) → SensorTag[source]#

Load sensor tag from the dict.

Example

>>> load_sensor_tag({"name": "tag", "field1": "value1"})
SensorTag("tag",field1="value1")
Parameters:

  • sensor – It should at least contain name field.

  • required_fields – Required additional fields.

gordo_core.sensor_tag.normalize_sensor_tag(sensor: dict[str, Optional[str]] | str | SensorTag, required_fields: Iterable[str] | None = None, **kwargs: str | None) → str | SensorTag[source]#

Take sensor tag information and tries to convert it to SensorTag.

Parameters:

  • sensor – Sensor tag information. Could be either dict suitable for load_sensor_tag() function or str

  • required_fields – Required additional fields.

  • kwargs – Additional fields.

gordo_core.sensor_tag.tag_to_json(tag: str | SensorTag) → str | dict[str, Optional[str]][source]#

Convert ::.Tag to JSON representation.

Parameters:

tag – Tag to convert.

gordo_core.sensor_tag.to_list_of_strings(sensor_tag_list: list[gordo_core.sensor_tag.SensorTag]) → list[str][source]#

List of sensor tags names.

Parameters:

sensor_tag_list – Sensor tags list

gordo_core.sensor_tag.unique_tag_names(*tags: Iterable[str | SensorTag]) → dict[str, Union[str, gordo_core.sensor_tag.SensorTag]][source]#

Check the uniqueness of the tags.

Parameters:

tags – Iterables of tags.

Returns:

  • Keys here are the unique tag names.

  • Tags will be in lower case if case_insensitive True.

gordo_core.sensor_tag.validate_tag_equality(tag1: str | SensorTag, tag2: str | SensorTag)[source]#

SensorTags with the same name should not have different fields. str and SensorTag should not have the same name.

Parameters:

  • tag1 – First tag to compare.

  • tag2 – Second tag to compare.