sensor.envirophat
#
Description#
Plugin to interact with a Pimoroni enviropHAT device.
You can use an enviropHAT device to read e.g. temperature, pressure, altitude, accelerometer, magnetometer and luminosity data.
Configuration#
sensor.envirophat:
# [Optional]
# A number, numeric pair or mapping of ``str`` to
# number/numeric pair representing the thresholds for the sensor.
#
# Examples:
#
# .. code-block:: yaml
#
# # Any value below 25 from any sensor will trigger a
# # SensorDataBelowThresholdEvent, if the previous value was
# # equal or above, and any value above 25 will trigger a
# # SensorDataAboveThresholdEvent, if the previous value was
# # equal or below
# thresholds: 25.0
#
# # Same as above, but the threshold is only applied to
# # ``temperature`` readings
# thresholds:
# temperature: 25.0
#
# # Any value below 20 from any sensor will trigger a
# # SensorDataBelowThresholdEvent, if the previous value was
# # equal or above, and any value above 25 will trigger a
# # SensorDataAboveThresholdEvent, if the previous value was
# # equal or below (hysteresis configuration with double
# # threshold)
# thresholds:
# - 20.0
# - 25.0
#
# # Same as above, but the threshold is only applied to
# # ``temperature`` readings
# thresholds:
# temperature:
# - 20.0
# - 25.0
# thresholds: # type=Union[float, int, Tuple[Union[float, int], Union[float, int]], Mapping[str, Union[float, int, Tuple[Union[float, int], Union[float, int]]]], NoneType]
# [Optional]
# If set, then the sensor change events will be
# triggered only if the difference between the new value and the
# previous value is higher than the specified tolerance. For example,
# if the sensor data is mapped to a dictionary::
#
# {
# "temperature": 0.01, # Tolerance on the 2nd decimal digit
# "humidity": 0.1 # Tolerance on the 1st decimal digit
# }
#
# Or, if it's a raw scalar number::
#
# 0.1 # Tolerance on the 1st decimal digit
#
# Or, if it's a list of values::
#
# [
# 0.01, # Tolerance on the 2nd decimal digit for the first value
# 0.1 # Tolerance on the 1st decimal digit for the second value
# ]
# tolerance: 0 # type=Union[float, int, Mapping[str, Union[float, int]], Iterable[Union[float, int]]]
# [Optional]
# If `SensorPlugin.get_measurement <https://docs.platypush.tech/platypush/plugins/sensor.html#platypush.plugins.sensor.SensorPlugin.get_measurement>`_ returns a key-value
# mapping, and ``enabled_sensors`` is set, then only the reported
# sensor keys will be returned.
# enabled_sensors: # type=Optional[Iterable[str]]
# [Optional]
# How often the `RunnablePlugin.loop <https://docs.platypush.tech/platypush/plugins/.html#platypush.plugins.RunnablePlugin.loop>`_ function should be
# executed (default: 15 seconds). *NOTE*: For back-compatibility
# reasons, the `poll_seconds` argument is also supported, but it's
# deprecated.
# poll_interval: 15 # type=Optional[float]
# [Optional]
# How long we should wait for any running
# threads/processes to stop before exiting (default: 5 seconds).
# stop_timeout: 5 # type=Optional[float]
# [Optional]
# If set to True then the plugin will not monitor
# for new events. This is useful if you want to run a plugin in
# stateless mode and only leverage its actions, without triggering any
# events. Defaults to False.
# disable_monitor: False # type=bool
Dependencies#
pip
pip install envirophat
Triggered events#
Actions#
Module reference#
- class platypush.plugins.sensor.envirophat.SensorEntityMapping(name: str, entity_type: ~typing.Type[~platypush.entities.sensors.Sensor], unit: str | None = None, kwargs: ~typing.Dict[str, ~typing.Any] = <factory>)[source]#
Bases:
object
Maps sensor attributes returned by
SensorEnvirophatPlugin.get_measurement()
to native sensor entities.- __init__(name: str, entity_type: ~typing.Type[~platypush.entities.sensors.Sensor], unit: str | None = None, kwargs: ~typing.Dict[str, ~typing.Any] = <factory>) None #
- property id#
Standard format for an Envirophat sensor external ID.
- class platypush.plugins.sensor.envirophat.SensorEnvirophatPlugin(*_, **__)[source]#
Bases:
SensorPlugin
Plugin to interact with a Pimoroni enviropHAT device.
You can use an enviropHAT device to read e.g. temperature, pressure, altitude, accelerometer, magnetometer and luminosity data.
- __init__(thresholds: float | int | Tuple[float | int, float | int] | Mapping[str, float | int | Tuple[float | int, float | int]] | None = None, tolerance: float | int | Mapping[str, float | int] | Iterable[float | int] = 0, enabled_sensors: Iterable[str] | None = None, **kwargs)#
- Parameters:
thresholds –
A number, numeric pair or mapping of
str
to number/numeric pair representing the thresholds for the sensor.Examples:
# Any value below 25 from any sensor will trigger a # SensorDataBelowThresholdEvent, if the previous value was # equal or above, and any value above 25 will trigger a # SensorDataAboveThresholdEvent, if the previous value was # equal or below thresholds: 25.0 # Same as above, but the threshold is only applied to # ``temperature`` readings thresholds: temperature: 25.0 # Any value below 20 from any sensor will trigger a # SensorDataBelowThresholdEvent, if the previous value was # equal or above, and any value above 25 will trigger a # SensorDataAboveThresholdEvent, if the previous value was # equal or below (hysteresis configuration with double # threshold) thresholds: - 20.0 - 25.0 # Same as above, but the threshold is only applied to # ``temperature`` readings thresholds: temperature: - 20.0 - 25.0
tolerance –
If set, then the sensor change events will be triggered only if the difference between the new value and the previous value is higher than the specified tolerance. For example, if the sensor data is mapped to a dictionary:
{ "temperature": 0.01, # Tolerance on the 2nd decimal digit "humidity": 0.1 # Tolerance on the 1st decimal digit }
Or, if it’s a raw scalar number:
0.1 # Tolerance on the 1st decimal digit
Or, if it’s a list of values:
[ 0.01, # Tolerance on the 2nd decimal digit for the first value 0.1 # Tolerance on the 1st decimal digit for the second value ]
enabled_sensors – If
get_measurement()
returns a key-value mapping, andenabled_sensors
is set, then only the reported sensor keys will be returned.
- get_data(*args, **kwargs)#
(Deprecated) alias for
get_measurement()
- get_measurement(*_, qnh: float = 1020.0, **__)[source]#
- Param:
qnh: Local value for atmospheric pressure adjusted to sea level (default: 1020)
- Returns:
dict. Example:
output = { "temperature": 21.0, # Celsius "pressure": 101555.08, # pascals "altitude": 10, # meters "luminosity": 426, # lumens # Measurements from the custom analog channels "analog": [0.513, 0.519, 0.531, 0.528], "accelerometer": { "x": -0.000915, "y": 0.0760, "z": 1.026733 }, "magnetometer": { "x": -2297, "y": 1226, "z": -7023 }, }
- main()#
Implementation of the main loop of the plugin.
- publish_entities(entities: float | int | Mapping[str, float | int] | Iterable[float | int], *args, **kwargs) Collection[Entity] #
Publishes a list of entities. The downstream consumers include:
The entity persistence manager
The web server
- Any consumer subscribed to
platypush.message.event.entities.EntityUpdateEvent
events (e.g. web clients)
It also accepts an optional callback that will be called when each of the entities in the set is flushed to the database.
You usually don’t need to override this class (but you may want to extend
transform_entities()
instead if your extension doesn’t natively handle Entity objects).
- start()#
Start the plugin.
- status(*_, **__) float | int | Mapping[str, float | int] | Iterable[float | int] | None #
Returns the latest read values and publishes the
platypush.message.event.entities.EntityUpdateEvent
events if required.
- stop()#
Stop the plugin.
- transform_entities(entities: Dict[str, float | int]) List[Device] [source]#
This method takes a list of entities in any (plugin-specific) format and converts them into a standardized collection of Entity objects. Since this method is called by
publish_entities()
before entity updates are published, you may usually want to extend it to pre-process the entities managed by your extension into the standard format before they are stored and published to all the consumers.
- wait_stop(timeout=None)#
Wait until a stop event is received.