Functions' definitions

In order to support the creation of sensors, triggers, charts or reports, Layrz provides a standardized structure to invoke the functions, and the return will change according to the function invoked.

Disclaimer

Our systems only watch for a single function, the name depends on the implementation, but you cannot define another function outside of that. If you need to declare other functions implement them inside of the main function!

DON'T do this:

python

""" DONT DO THIS! """
def square(a: float) -> float: # Function defined outside of the main function
    return a * a

def calculate(asset: Asset, message: Message, previous_message: Message | None) -> float:
  """ Return the current squared speed of the asset """
  return square(message.position.speed) or 0

""" DO THIS INSTEAD! """
def calculate(asset: Asset, message: Message, previous_message: Message | None) -> float:
  """ Return the current squared speed of the asset """
  def square(a: float) -> float: # Function defined inside the main function
    return a * a
  return square(message.position.speed) or 0

Let's dive into the structure of the functions.

For basic sensors

Python in sensors is very straightforward, you just need to create a function called calculate with three arguments:

Argument	Type	Description
`asset`	`Asset`	The asset that the sensor is attached to
`message`	`Message`	The data that the sensor will use to calculate the value
`previous_message`	`Message`	The previous message that the sensor used to calculate the value

The return of the function can be anything you want to, feel free to return a str, a int, a float, a list, a dict, or any other type of data.

Let's look an example:

Equivalent in LCL

This example returns the current speed of the asset, if the speed is not available, it returns 0

lcl

GET_PARAM(
  CONCAT(
    PRIMARY_DEVICE(),
    ".position.speed"
  ),
  CONSTANT(0)
)

python

def calculate(asset: Asset, message: Message, previous_message: Message | None) -> float:
  """ Return the current speed of the asset """
  return message.position.speed or 0

Be careful!

The `previous_message` argument can be None if the current message follows this rules:

The previous_message is after message. Aka, a blackbox message
If is the first time sending data to Layrz

Otherwise, it will always be a valid message.

For dynamic sensors

What is a dynamic sensor? Well, in some cases, you need to filter or standardize the value using median or average, or maybe you need to predict some things using statistical models. In those cases, you need to use a dynamic sensor.

Different from the basic sensor, the dynamic sensor, the function name is calculate_dynamic with two arguments:

Argument	Type	Description
`asset`	`Asset`	The asset that the sensor is attached to
`messages`	`list[Message]`	The last 20 messages to analyze, sorted by received_at ascending

Sensor setup

This 20 messages depends of the configuration of the sensor and the messages available in the system. By default we only scan for the last hour to get the last 20 messages, but you can change this search time up to 6 hours.

The return of the function can be anything you want to, feel free to return a str, a int, a float, a list, a dict, or any other type of data.

Let's look an example:

python

def calculate_dynamic(asset: Asset, messages: list[Message]) -> float:
  """ Return the average speed of the last 20 messages """
  return sum([message.position.speed for message in messages]) / len(messages)

For triggers

For Triggers, is similar to a basic sensor, but you need to create a function called validate with three arguments:

Argument	Type	Description
`asset`	`Asset`	The asset that the sensor is attached to
`message`	`Message`	The data that the sensor will use to calculate the value
`previous_message`	`Message`	The previous message that the sensor used to calculate the value

The return of the function must be a bool, True if the trigger is activated, False if not.

Let's look an example:

python

def validate(asset, message, previous_message):
  """ Return True if the speed is greater than 100 """
  return message.position.speed > 100

Be careful!

If the return is not a bool, the trigger will not work in any scenario.

For fixed execution triggers

For fixed execution Triggers, instead you need to create a function called validate_fixed with only one argument:

Argument	Type	Description
`last_messages`	list[`LastMessage`]	The last messages that the sensor used to calculate the value

The return of the function must be a list of IDs (list[int]) of the assets that were positive.

Let's look an example:

python

def validate_fixed(last_messages: list[LastMessage]) -> list[int]:
  """ Return the IDs of the assets that have the engine ignition status as True """
  # Do a positive only for assets that engine.ignition.status is true  positives = []
  for message in last_messages:
    if message.sensors is None:
      continue
    
    if message.sensors.get('engine.ignition.status', False):
      positives.append(message.asset_id)
      
  return positives

For charts

Charts are a little bit different, the function name is calculate_series with two common arguments:

Argument	Type	Description
`assets`	`list[Asset]`	The list of assets submitted in the render request
`configuration`	`ChartConfiguration`	The configuration of the chart

And, also, depending of the chart data source, whe third argument will change:

Data Source	Argument	Type
For messages as a source	`messages`	`list[Message]`
For events as a source	`events`	`list[Event]`
For cases as a source	`cases`	`list[Case]`
For checkpoints as a source	`checkpoints`	`list[Checkpoint]`

The return statement must be the chart type:

Chart Type	Class
Line	`LineChart`
Column	`ColumnChart`
Area	`AreaChart`
RadialBar	`RadialBarChart`
Pie	`PieChart`
Bar	`BarChart`
Radar	`RadarChart`
Scatter	`ScatterChart`
Timeline	`TimelineChart`
HTML	`HTMLChart`
Map	`MapChart`
Table	`TableChart`
Number	`NumberChart`

Let's look an example of each data source returning a LineChart:

For messages as a source

python

def calculate_series(assets: list[Asset], configuration: ChartConfiguration, messages: list[Message]) -> LineChart:
  return LineChart(
    title='Example chart',
    x_axis=[],
    y_axis=[]
  )

For events as a source

python

def calculate_series(assets: list[Asset], configuration: ChartConfiguration, events: list[Event]) -> LineChart:
  return LineChart(
    title='Example chart',
    x_axis=[],
    y_axis=[]
  )

For cases as a source

python

def calculate_series(assets: list[Asset], configuration: ChartConfiguration, cases: list[Case]) -> LineChart:
  return LineChart(
    title='Example chart',
    x_axis=[],
    y_axis=[]
  )

For checkpoints as a source

python

def calculate_series(assets: list[Asset], configuration: ChartConfiguration, checkpoints: list[Checkpoint]) -> LineChart:
  return LineChart(
    title='Example chart',
    x_axis=[],
    y_axis=[]
  )

Be careful!

This examples are just to show the structure of the function, checkout the definition of each chart type to understand how to use it.

Checkout our examples!

We have a lot of examples of charts in our Python examples

For reports

For reports, is similar to the chart function, the main difference is the function name, that is process_report with two common arguments:

Argument	Type	Description
`assets`	`list[Asset]`	The list of assets submitted in the render request
`configuration`	`ReportConfiguration`	The configuration of the report

And, also, depending of the report data source, whe third argument will change:

Data Source	Argument	Type
For messages as a source	`messages`	`list[Message]`
For events as a source	`events`	`list[Event]`
For cases as a source	`cases`	`list[Case]`
For checkpoints as a source	`checkpoints`	`list[Checkpoint]`
For broadcasts' results as a source	`broadcasts`	`list[BroadcastResult]`

The return statement must be the ReportPage class, this class will be appended to the report.

Let's look an example of each data source returning a ReportPage:

For messages as a source

python

def process_report(assets: list[Asset], configuration: ReportConfiguration, messages: list[Message]) -> ReportPage:
  return ReportPage(
    name='Example page',
    headers=[],
    rows=[],
    freeze_header=False
  )

For events as a source

python

def process_report(assets: list[Asset], configuration: ReportConfiguration, events: list[Event]) -> ReportPage:
  return ReportPage(
    name='Example page',
    headers=[],
    rows=[],
    freeze_header=False
  )

For cases as a source

python

def process_report(assets: list[Asset], configuration: ReportConfiguration, cases: list[Case]) -> ReportPage:
  return ReportPage(
    name='Example page',
    headers=[],
    rows=[],
    freeze_header=False
  )

For checkpoints as a source

python

def process_report(assets: list[Asset], configuration: ReportConfiguration, checkpoints: list[Checkpoint]) -> ReportPage:
  return ReportPage(
    name='Example page',
    headers=[],
    rows=[],
    freeze_header=False
  )

For broadcasts' results as a source

python

def process_report(assets: list[Asset], configuration: ReportConfiguration, broadcasts: list[BroadcastResult]) -> ReportPage:
  return ReportPage(
    name='Example page',
    headers=[],
    rows=[],
    freeze_header=False
  )

Be careful!

This examples are just to show the structure of the function, checkout the definition of the report class to understand how to use it.

Checkout our examples!

We have examples of reports in our Python examples

Functions' definitions ​

Disclaimer

For basic sensors ​

Equivalent in LCL

Be careful!

For dynamic sensors ​

Sensor setup

For triggers ​

Be careful!

For fixed execution triggers ​

For charts ​

For messages as a source ​

For events as a source ​

For cases as a source ​

For checkpoints as a source ​

Be careful!

Checkout our examples!

For reports ​

For messages as a source ​

For events as a source ​

For cases as a source ​

For checkpoints as a source ​

For broadcasts' results as a source ​

Be careful!

Checkout our examples!

Functions' definitions

For basic sensors

For dynamic sensors

For triggers

For fixed execution triggers

For charts

For messages as a source

For events as a source

For cases as a source

For checkpoints as a source

For reports

For messages as a source

For events as a source

For cases as a source

For checkpoints as a source

For broadcasts' results as a source