juham.automation

Description

Classes implementing automated tasks. Objects in this package typically control some physical device, such as heating radiator, or water cirulator bump. In order to do they job they subscribe to appropropate Juham MQTT topics to acquire telemetry from sensors and web resources.

class juham.automation.EnergyCostCalculator(name='ecc')[source]

Bases: Base

Energy revenue/cost calculator for calculating energy balance between produced and consumed energy.

Subscribes to spot and power MQTT topics, calculates per hour, per day, per month and per year costs and writes them the it to the time series database.

app_name = 'juham'
calculate_net_energy_cost(ts_prev, ts_now, energy)[source]

Given time interval as start and stop Calculate the cost over the given time period. Positive values indicate revenue, negative cost.

Parameters:
  • ts_prev (timestamp) – beginning time stamp of the interval

  • ts_now (timestamp) – end of the interval

  • energy (float) – energy consumed during the time interval

Returns:

Cost or revenue

classmethod classattrs_from_dict(attributes)

Set class attributes from a dictionary.

classmethod classattrs_to_dict()

Convert class attributes to a dictionary.

config_folder = 'config'
copy()

Create and return a copy of the current object.

This method serializes the current object to a dictionary using the to_dict method, creates a new instance of the object’s class, and populates it with the serialized data using the from_dict method.

This method uses class identifier based instantiation (see factory method pattern) to create a new instance of the object, and ‘to_dict’ and ‘from_dict’ methods to initialize object’s state.

Returns:

A new instance of the object’s class with the same state as the original object.

Example:

john = Human("john")
john.height = 1.8
john.age = 62
clone_of_john = john.copy()
database_class_id: str = 0
debug(msg, details='')

Logs the given debug message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

deserialize_from_json(f)

Load attributes from a JSON file.

epoc2utc(epoch)

Converts the given epoch time to UTC time string. All time coordinates are represented in UTC time. This allows the time coordinate to be mapped to any local time representation without ambiguity.

Parameters:
  • epoch (float) – timestamp in UTC time

  • rc (str) – time string describing date, time and time zone e.g 2024-07-08T12:10:22Z

Returns:

UTC time

error(msg, details='')

Logs the given error message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

classmethod find_class(class_id)

Given class identifier find the registered class. If no class with the give identifier exists return None.

Parameters:

class_id (int) – class identifier

Returns:

class or null if not registered

Return type:

obj (obj)

from_dict(data)

Update instance attributes from a dictionary.

classmethod get_class_id()

Return the class id of the class. Each class has an unique identifier that can be used for instantiating the class via Object.instantiate() method.

Parameters:

cls (class) – class

Return type:

str

Returns:

id (int) unique class identifier through which the class can be instantiated by factory method pattern.

classmethod get_json_file()

Generate the JSON file name based on the class name.

The file is created into users home folder.

get_prices(ts_prev, ts_now)[source]

Fetch the electricity prices for the given two subsequent time stamps.

Parameters:
  • ts_prev (timestamp) – previous time

  • ts_now (timestamp) – current time

Returns:

Electricity prices for the given interval

info(msg, details='')

Logs the given information message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

Example:

obj = new Base('test')
obj.info('Message arrived', str(msg))
init_database(name)

Instantiates the configured time series database object.

Issues a warning if the database_class_id has not been configured, in which case the object will not have the time series recording feature.

This method is called internally and typically there is no need to call it from the application code.

Return type:

None

init_mqtt(name)

Instantiates the configured MQTT object for networking.

This method is called internally and typically there is no need to call it from the application code.

Issues a warning if the pubsub_class_id has not been configured, even though objects without a capability to communicate are rather crippled.

Return type:

None

classmethod initialize_class(load_class_attrs=True)

Initialize the class for instantiation, if not initialized already. This method initializes the class identifier and deserializes the public attributes from the specified configuration folder.

Parameters:
  • load_class_attrs (bool, optional) – If true then attempts to initialize class attributes

  • True. (from the class specific configuration files. Defaults to)

Returns:

returns true if the class was initialized, false implies the class is already initialized in which case the method call has no effect.

Return type:

bool

classmethod instantiate(class_id)

Create an instance of the class corresponding to the given class identifier. This method implements the factory method pattern, which is essential for a plugin architecture.

Parameters:

class_id (int) – Identifier of the class to instantiate.

Returns:

An instance of the class corresponding to the given class identifier.

Return type:

obj

classmethod instantiate_with_param(class_id, param)

Given class identifier and one constructor argument create the corresponding object.

Parameters:
  • class_id (str) – class identifier

  • param (Any) – class specific constructor parameter

Returns:

instance of the given class.

Return type:

obj

classmethod is_abstract()

Check whether the class is abstract or real. Override in the derived sub-classes. The default is False.

Return type:

bool

Returns:

True (bool) if abstract

is_time_between(begin_time, end_time, check_time=None)

Check if the given time is within the given time line. All timestamps must be in UTC time.

Parameters:
  • begin_time (timestamp) – beginning of the timeline

  • end_time (timestamp) – end of the timeline

  • check_time (timestamp) – time to be checked

Returns:

True if within the timeline

Return type:

rc (bool)

classmethod load_from_json()

Load class attributes from a JSON file.

log_message(type, msg, details='')

Publish the given log message to the MQTT ‘log’ topic.

This method constructs a log message with a timestamp, class type, source name, message, and optional details. It then publishes this message to the ‘log’ topic using the MQTT protocol.

Parameters:
  • type – str The classification or type of the log message (e.g., ‘Error’, ‘Info’).

  • msg – str The main log message to be published.

  • details – str, optional Additional details about the log message (default is an empty string).

Returns:

None

Raises:

Exception – If there is an issue with the MQTT client while publishing the message.

Example:

# publish info message to the Juham's 'log' topic
self.log_message("Info", f"Some cool message {some_stuff}", str(dict))
map_prices_to_joules(price)[source]

Convert the given electricity price in kWh to Watt seconds (J) :type price: :param price: electricity price given as kWh :type price: float

Returns:

Electricity price per watt second (J)

mqtt_class_id: str = ''
mqtt_host: str = 'localhost'
mqtt_port: int = 1883
mqtt_root_topic: str = 'juham'
on_connect(client, userdata, flags, rc)[source]

Notification on connect.

This method is called whenever the MQTT broker is connected. For more information on this method consult MQTT documentation available in many public sources.

Parameters:
  • client (obj) – MQTT client

  • userdata (Any) – application specific data

  • flags (int) – Consult MQTT

  • rc (int) – See MQTT docs

on_disconnect(client, userdata, rc=0)

Notification on disconnect.

This method is called whenever the MQTT broker is disconnected. For more information on this method consult MQTT documentation available in many public sources.

Parameters:
  • client (obj) – MQTT client

  • userdata (Any) – application specific data

  • rc (int) – See MQTT docs

on_message(client, userdata, msg)[source]

Handle MQTT message.

Parameters:
  • client (object) – client

  • userdata (any) – user data

  • msg (object) – mqtt message

on_powerconsumption(ts_now, m)[source]

Update energy consumption.

Parameters:
  • ts_now (timestamp) – time stamp of the energy consumed

  • m (dictionary) – Juham MQTT message holding energy reading

on_spot(spot)[source]

Stores the received per hour electricity prices to spots list.

Parameters:

spot (list) – list of hourly spot prices

classmethod parse_args()

Parse the startup arguments defined by this class.

Return type:

None

publish(topic, msg, qos=1, retain=True)

Publish the given message to the given MQTT topic. For more information consult MQTT.

Parameters:
  • topic (str) – topic

  • msg (str) – message to be published

  • qos (int, optional) – quality of service. Defaults to 1.

  • retain (bool, optional) – retain. Defaults to True.

pubsub_class_id = 0
read(point)

Reads the given measurement from the database.

Parameters:

point – point with initialized time stamp.

… note: NOT IMPLEMENTED YET

record_energycost(ts_now, site, cost)[source]

Record energy cost/revenue to data storage. Positive values represent revenue whereas negative cost. :type ts_now: :param ts_now: timestamp :type ts_now: float :type site: :param site: site :type site: [type] :type cost: :param cost: cost or revenue. :type cost: [type]

classmethod register()[source]

Register the class to the class database.

Once registered the class can be instantiated by its class identifier. Note that this method is called automatically by the system when the python loads the class. In this method sub classes should prepare themselves for instantiation by initializing their class attributes, for example.

classmethod register_class(class_id, ctor)

Register the given class identifier for identifier based instantiation . This, factory method pattern, as it is called, decouples the actual implementation from the interface. For more information see instantiate() method.

Parameters:
  • class_id (str) – class identifier

  • ctor (function) – constructor

classmethod register_topic(topic)

Register the given topic as known valid topic. Publishing and subscriptions are allowed only to registered topics. Each topic must have an owner, who is responsible for registering the topic.

Parameters:

topic (str) – valid topic

run()

Start a new thread to runs the network loop in the background.

Allows the main program to continue executing while the MQTT client handles incoming and outgoing messages in the background.

run_forever()

Starts the network loop and blocks the main thread, continuously running the loop to process MQTT messages.

The loop will run indefinitely unless the connection is lost or the program is terminated.

classmethod save_to_json()

Create class configuration file, if the file does not exist yet.

serialize_to_json(f)

Serialize.

classmethod set_log(l)

Set logger.

Parameters:

l (logger) – logger object

Return type:

None

shutdown()

Shut down all services, free resources, stop threads, disconnect from mqtt, in general, prepare for shutdown.

subscribe(topic)

Subscribe to the given MQTT topic.

This method sets up the subscription to the specified MQTT topic and registers the on_message() method as the callback for incoming messages.

Parameters:

topic (str) – The MQTT topic to subscribe to.

Return type:

None

Example:

# configure
obj.subscribe('foo/bar')
timestamp()

Returns the current date-time in UTC.

Returns:

datetime in UTC.

Return type:

rc (datetime)

timestampstr(ts)

Converts the given timestamp to human readable string of form ‘Y-m-d H:M:S’.

Parameters:

ts (timestamp) – time stamp to be converted

Returns:

human readable date-time string

Return type:

rc (string)

to_dict()

Convert instance attributes to a dictionary.

to_joule_coeff = 2.7777777777777776e-07
topic_in_powerconsumption = 'juham/powerconsumption'
topic_in_spot = 'juham/spot'
valid_topic(topic)

Check if the given topic has been registered as valid topic.

Parameters:

topic (str) – topic to be validated

Returns:

true if known topic.

Return type:

bool

warning(msg, details='')

Logs the given warning message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

write(point)

Writes the given measurement to the database. In case of an error, it tries again until the maximum number of attempts is reached. If it is still unsuccessful, it gives up and passes the first encountered exception to the caller.

Parameters:

point – a measurement describing a time stamp and related attributes for one measurement.

write_attempts = 3
class juham.automation.RBoiler(name='rboiler')[source]

Bases: JShelly

Automation class for controlling Shelly Wifi relay. Subscribes to ‘power’ topic and controls the Shelly relay accordingly.

Parameters:

JShelly (class) – Super class

app_name = 'juham'
classmethod classattrs_from_dict(attributes)

Set class attributes from a dictionary.

classmethod classattrs_to_dict()

Convert class attributes to a dictionary.

config_folder = 'config'
copy()

Create and return a copy of the current object.

This method serializes the current object to a dictionary using the to_dict method, creates a new instance of the object’s class, and populates it with the serialized data using the from_dict method.

This method uses class identifier based instantiation (see factory method pattern) to create a new instance of the object, and ‘to_dict’ and ‘from_dict’ methods to initialize object’s state.

Returns:

A new instance of the object’s class with the same state as the original object.

Example:

john = Human("john")
john.height = 1.8
john.age = 62
clone_of_john = john.copy()
database_class_id: str = 0
debug(msg, details='')

Logs the given debug message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

deserialize_from_json(f)

Load attributes from a JSON file.

elapsed(secs)

Check if a certain time interval has elapsed and update the start timestamp attribute to count elapsed seconds for future calls.

Parameters:

secs (float) – The expected elapsed time in seconds since the previous call

Returns:

True if the given number of seconds has elapsed, False otherwise.

Return type:

bool

epoc2utc(epoch)

Converts the given epoch time to UTC time string. All time coordinates are represented in UTC time. This allows the time coordinate to be mapped to any local time representation without ambiguity.

Parameters:
  • epoch (float) – timestamp in UTC time

  • rc (str) – time string describing date, time and time zone e.g 2024-07-08T12:10:22Z

Returns:

UTC time

error(msg, details='')

Logs the given error message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

classmethod find_class(class_id)

Given class identifier find the registered class. If no class with the give identifier exists return None.

Parameters:

class_id (int) – class identifier

Returns:

class or null if not registered

Return type:

obj (obj)

from_dict(data)

Update instance attributes from a dictionary.

classmethod get_class_id()

Return the class id of the class. Each class has an unique identifier that can be used for instantiating the class via Object.instantiate() method.

Parameters:

cls (class) – class

Return type:

str

Returns:

id (int) unique class identifier through which the class can be instantiated by factory method pattern.

classmethod get_json_file()

Generate the JSON file name based on the class name.

The file is created into users home folder.

info(msg, details='')

Logs the given information message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

Example:

obj = new Base('test')
obj.info('Message arrived', str(msg))
init_database(name)

Instantiates the configured time series database object.

Issues a warning if the database_class_id has not been configured, in which case the object will not have the time series recording feature.

This method is called internally and typically there is no need to call it from the application code.

Return type:

None

init_mqtt(name)

Instantiates the configured MQTT object for networking.

This method is called internally and typically there is no need to call it from the application code.

Issues a warning if the pubsub_class_id has not been configured, even though objects without a capability to communicate are rather crippled.

Return type:

None

classmethod initialize_class(load_class_attrs=True)

Initialize the class for instantiation, if not initialized already. This method initializes the class identifier and deserializes the public attributes from the specified configuration folder.

Parameters:
  • load_class_attrs (bool, optional) – If true then attempts to initialize class attributes

  • True. (from the class specific configuration files. Defaults to)

Returns:

returns true if the class was initialized, false implies the class is already initialized in which case the method call has no effect.

Return type:

bool

classmethod instantiate(class_id)

Create an instance of the class corresponding to the given class identifier. This method implements the factory method pattern, which is essential for a plugin architecture.

Parameters:

class_id (int) – Identifier of the class to instantiate.

Returns:

An instance of the class corresponding to the given class identifier.

Return type:

obj

classmethod instantiate_with_param(class_id, param)

Given class identifier and one constructor argument create the corresponding object.

Parameters:
  • class_id (str) – class identifier

  • param (Any) – class specific constructor parameter

Returns:

instance of the given class.

Return type:

obj

classmethod is_abstract()

Check whether the class is abstract or real. Override in the derived sub-classes. The default is False.

Return type:

bool

Returns:

True (bool) if abstract

is_time_between(begin_time, end_time, check_time=None)

Check if the given time is within the given time line. All timestamps must be in UTC time.

Parameters:
  • begin_time (timestamp) – beginning of the timeline

  • end_time (timestamp) – end of the timeline

  • check_time (timestamp) – time to be checked

Returns:

True if within the timeline

Return type:

rc (bool)

classmethod load_from_json()

Load class attributes from a JSON file.

log_message(type, msg, details='')

Publish the given log message to the MQTT ‘log’ topic.

This method constructs a log message with a timestamp, class type, source name, message, and optional details. It then publishes this message to the ‘log’ topic using the MQTT protocol.

Parameters:
  • type – str The classification or type of the log message (e.g., ‘Error’, ‘Info’).

  • msg – str The main log message to be published.

  • details – str, optional Additional details about the log message (default is an empty string).

Returns:

None

Raises:

Exception – If there is an issue with the MQTT client while publishing the message.

Example:

# publish info message to the Juham's 'log' topic
self.log_message("Info", f"Some cool message {some_stuff}", str(dict))
mqtt_class_id: str = ''
mqtt_host: str = 'localhost'
mqtt_port: int = 1883
mqtt_root_topic: str = 'juham'
on_connect(client, userdata, flags, rc)[source]

Notification on connect.

This method is called whenever the MQTT broker is connected. For more information on this method consult MQTT documentation available in many public sources.

Parameters:
  • client (obj) – MQTT client

  • userdata (Any) – application specific data

  • flags (int) – Consult MQTT

  • rc (int) – See MQTT docs

on_disconnect(client, userdata, rc=0)

Notification on disconnect.

This method is called whenever the MQTT broker is disconnected. For more information on this method consult MQTT documentation available in many public sources.

Parameters:
  • client (obj) – MQTT client

  • userdata (Any) – application specific data

  • rc (int) – See MQTT docs

on_message(client, userdata, msg)[source]

Handle incoming MQTT message :type client: :param client: mqtt client :type client: object :type userdata: :param userdata: user context :type userdata: any :type msg: :param msg: mqtt message :type msg: object

on_power(m)[source]

Process power_topic message.

Parameters:

m (dict) – holding data from the power sensor

classmethod parse_args()

Parse the startup arguments defined by this class.

Return type:

None

power_topic = 'power'
publish(topic, msg, qos=1, retain=True)

Publish the given message to the given MQTT topic. For more information consult MQTT.

Parameters:
  • topic (str) – topic

  • msg (str) – message to be published

  • qos (int, optional) – quality of service. Defaults to 1.

  • retain (bool, optional) – retain. Defaults to True.

pubsub_class_id = 0
read(point)

Reads the given measurement from the database.

Parameters:

point – point with initialized time stamp.

… note: NOT IMPLEMENTED YET

classmethod register()[source]

Register the class to the class database.

Once registered the class can be instantiated by its class identifier. Note that this method is called automatically by the system when the python loads the class. In this method sub classes should prepare themselves for instantiation by initializing their class attributes, for example.

classmethod register_class(class_id, ctor)

Register the given class identifier for identifier based instantiation . This, factory method pattern, as it is called, decouples the actual implementation from the interface. For more information see instantiate() method.

Parameters:
  • class_id (str) – class identifier

  • ctor (function) – constructor

classmethod register_topic(topic)

Register the given topic as known valid topic. Publishing and subscriptions are allowed only to registered topics. Each topic must have an owner, who is responsible for registering the topic.

Parameters:

topic (str) – valid topic

relay_url = 'shellyplus1-alakerta/command/switch:0'
run()

Start a new thread to runs the network loop in the background.

Allows the main program to continue executing while the MQTT client handles incoming and outgoing messages in the background.

run_forever()

Starts the network loop and blocks the main thread, continuously running the loop to process MQTT messages.

The loop will run indefinitely unless the connection is lost or the program is terminated.

classmethod save_to_json()

Create class configuration file, if the file does not exist yet.

serialize_to_json(f)

Serialize.

classmethod set_log(l)

Set logger.

Parameters:

l (logger) – logger object

Return type:

None

shutdown()

Shut down all services, free resources, stop threads, disconnect from mqtt, in general, prepare for shutdown.

subscribe(topic)

Subscribe to the given MQTT topic.

This method sets up the subscription to the specified MQTT topic and registers the on_message() method as the callback for incoming messages.

Parameters:

topic (str) – The MQTT topic to subscribe to.

Return type:

None

Example:

# configure
obj.subscribe('foo/bar')
timestamp()

Returns the current date-time in UTC.

Returns:

datetime in UTC.

Return type:

rc (datetime)

timestampstr(ts)

Converts the given timestamp to human readable string of form ‘Y-m-d H:M:S’.

Parameters:

ts (timestamp) – time stamp to be converted

Returns:

human readable date-time string

Return type:

rc (string)

to_dict()

Convert instance attributes to a dictionary.

valid_topic(topic)

Check if the given topic has been registered as valid topic.

Parameters:

topic (str) – topic to be validated

Returns:

true if known topic.

Return type:

bool

warning(msg, details='')

Logs the given warning message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

write(point)

Writes the given measurement to the database. In case of an error, it tries again until the maximum number of attempts is reached. If it is still unsuccessful, it gives up and passes the first encountered exception to the caller.

Parameters:

point – a measurement describing a time stamp and related attributes for one measurement.

write_attempts = 3
class juham.automation.RPowerPlan(name='rpowerplan')[source]

Bases: Base

Automation class for optimized control of home energy consumers e.g hot water boilers. Reads spot prices, boiler water temperatures and controls heating radiators.

app_name = 'juham'
classmethod classattrs_from_dict(attributes)

Set class attributes from a dictionary.

classmethod classattrs_to_dict()

Convert class attributes to a dictionary.

compute_fom(solpower, spot)[source]

Compute UOI - utilization optimization index.

Parameters:
  • solpower (float) – current solar power forecast

  • spot (float) – spot price

Returns:

utilization optimization index

Return type:

float

config_folder = 'config'
consider_heating(ts)[source]

Consider whether the target boiler needs heating.

Parameters:

ts (float) – current UTC time

Returns:

1 if heating is needed, 0 if not

Return type:

int

copy()

Create and return a copy of the current object.

This method serializes the current object to a dictionary using the to_dict method, creates a new instance of the object’s class, and populates it with the serialized data using the from_dict method.

This method uses class identifier based instantiation (see factory method pattern) to create a new instance of the object, and ‘to_dict’ and ‘from_dict’ methods to initialize object’s state.

Returns:

A new instance of the object’s class with the same state as the original object.

Example:

john = Human("john")
john.height = 1.8
john.age = 62
clone_of_john = john.copy()
create_heating_plan()[source]

Create heating plan.

Returns:

list of heating entries

Return type:

int

create_power_plan()[source]

Create power plan.

Returns:

list of utilization entries

Return type:

list

database_class_id: str = 0
debug(msg, details='')

Logs the given debug message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

deserialize_from_json(f)

Load attributes from a JSON file.

epoc2utc(epoch)

Converts the given epoch time to UTC time string. All time coordinates are represented in UTC time. This allows the time coordinate to be mapped to any local time representation without ambiguity.

Parameters:
  • epoch (float) – timestamp in UTC time

  • rc (str) – time string describing date, time and time zone e.g 2024-07-08T12:10:22Z

Returns:

UTC time

error(msg, details='')

Logs the given error message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

classmethod find_class(class_id)

Given class identifier find the registered class. If no class with the give identifier exists return None.

Parameters:

class_id (int) – class identifier

Returns:

class or null if not registered

Return type:

obj (obj)

from_dict(data)

Update instance attributes from a dictionary.

classmethod get_class_id()

Return the class id of the class. Each class has an unique identifier that can be used for instantiating the class via Object.instantiate() method.

Parameters:

cls (class) – class

Return type:

str

Returns:

id (int) unique class identifier through which the class can be instantiated by factory method pattern.

classmethod get_json_file()

Generate the JSON file name based on the class name.

The file is created into users home folder.

info(msg, details='')

Logs the given information message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

Example:

obj = new Base('test')
obj.info('Message arrived', str(msg))
init_database(name)

Instantiates the configured time series database object.

Issues a warning if the database_class_id has not been configured, in which case the object will not have the time series recording feature.

This method is called internally and typically there is no need to call it from the application code.

Return type:

None

init_mqtt(name)

Instantiates the configured MQTT object for networking.

This method is called internally and typically there is no need to call it from the application code.

Issues a warning if the pubsub_class_id has not been configured, even though objects without a capability to communicate are rather crippled.

Return type:

None

classmethod initialize_class(load_class_attrs=True)

Initialize the class for instantiation, if not initialized already. This method initializes the class identifier and deserializes the public attributes from the specified configuration folder.

Parameters:
  • load_class_attrs (bool, optional) – If true then attempts to initialize class attributes

  • True. (from the class specific configuration files. Defaults to)

Returns:

returns true if the class was initialized, false implies the class is already initialized in which case the method call has no effect.

Return type:

bool

classmethod instantiate(class_id)

Create an instance of the class corresponding to the given class identifier. This method implements the factory method pattern, which is essential for a plugin architecture.

Parameters:

class_id (int) – Identifier of the class to instantiate.

Returns:

An instance of the class corresponding to the given class identifier.

Return type:

obj

classmethod instantiate_with_param(class_id, param)

Given class identifier and one constructor argument create the corresponding object.

Parameters:
  • class_id (str) – class identifier

  • param (Any) – class specific constructor parameter

Returns:

instance of the given class.

Return type:

obj

classmethod is_abstract()

Check whether the class is abstract or real. Override in the derived sub-classes. The default is False.

Return type:

bool

Returns:

True (bool) if abstract

is_time_between(begin_time, end_time, check_time=None)

Check if the given time is within the given time line. All timestamps must be in UTC time.

Parameters:
  • begin_time (timestamp) – beginning of the timeline

  • end_time (timestamp) – end of the timeline

  • check_time (timestamp) – time to be checked

Returns:

True if within the timeline

Return type:

rc (bool)

classmethod load_from_json()

Load class attributes from a JSON file.

log_message(type, msg, details='')

Publish the given log message to the MQTT ‘log’ topic.

This method constructs a log message with a timestamp, class type, source name, message, and optional details. It then publishes this message to the ‘log’ topic using the MQTT protocol.

Parameters:
  • type – str The classification or type of the log message (e.g., ‘Error’, ‘Info’).

  • msg – str The main log message to be published.

  • details – str, optional Additional details about the log message (default is an empty string).

Returns:

None

Raises:

Exception – If there is an issue with the MQTT client while publishing the message.

Example:

# publish info message to the Juham's 'log' topic
self.log_message("Info", f"Some cool message {some_stuff}", str(dict))
maximum_boiler_temperature = 70
minimum_boiler_temperature = 45
mqtt_class_id: str = ''
mqtt_host: str = 'localhost'
mqtt_port: int = 1883
mqtt_root_topic: str = 'juham'
on_connect(client, userdata, flags, rc)[source]

Notification on connect.

This method is called whenever the MQTT broker is connected. For more information on this method consult MQTT documentation available in many public sources.

Parameters:
  • client (obj) – MQTT client

  • userdata (Any) – application specific data

  • flags (int) – Consult MQTT

  • rc (int) – See MQTT docs

on_disconnect(client, userdata, rc=0)

Notification on disconnect.

This method is called whenever the MQTT broker is disconnected. For more information on this method consult MQTT documentation available in many public sources.

Parameters:
  • client (obj) – MQTT client

  • userdata (Any) – application specific data

  • rc (int) – See MQTT docs

on_message(client, userdata, msg)[source]

MQTT message notification on arrived message.

Called whenever a new message is posted on one of the topics the object has subscribed to via subscribe() method. This method is the heart of automation: here, derived subclasses should automate whatever they were designed to automate. For example, they could switch a relay when a boiler temperature sensor signals that the temperature is too low for a comforting shower for say one’s lovely wife.

For more information on this method consult MQTT documentation available in many public sources.

Parameters:
  • client (obj) – MQTT client

  • userdata (Any) – application specific data

  • msg (object) – The MQTT message

Return type:

None

on_powerplan(ts_utc_now)[source]

Apply power plan.

Parameters:

ts_utc_now (float) – utc time

Return type:

None

classmethod parse_args()

Parse the startup arguments defined by this class.

Return type:

None

publish(topic, msg, qos=1, retain=True)

Publish the given message to the given MQTT topic. For more information consult MQTT.

Parameters:
  • topic (str) – topic

  • msg (str) – message to be published

  • qos (int, optional) – quality of service. Defaults to 1.

  • retain (bool, optional) – retain. Defaults to True.

pubsub_class_id = 0
read(point)

Reads the given measurement from the database.

Parameters:

point – point with initialized time stamp.

… note: NOT IMPLEMENTED YET

classmethod register()[source]

Register the class to the class database.

Once registered the class can be instantiated by its class identifier. Note that this method is called automatically by the system when the python loads the class. In this method sub classes should prepare themselves for instantiation by initializing their class attributes, for example.

Return type:

None

classmethod register_class(class_id, ctor)

Register the given class identifier for identifier based instantiation . This, factory method pattern, as it is called, decouples the actual implementation from the interface. For more information see instantiate() method.

Parameters:
  • class_id (str) – class identifier

  • ctor (function) – constructor

classmethod register_topic(topic)

Register the given topic as known valid topic. Publishing and subscriptions are allowed only to registered topics. Each topic must have an owner, who is responsible for registering the topic.

Parameters:

topic (str) – valid topic

run()

Start a new thread to runs the network loop in the background.

Allows the main program to continue executing while the MQTT client handles incoming and outgoing messages in the background.

run_forever()

Starts the network loop and blocks the main thread, continuously running the loop to process MQTT messages.

The loop will run indefinitely unless the connection is lost or the program is terminated.

classmethod save_to_json()

Create class configuration file, if the file does not exist yet.

serialize_to_json(f)

Serialize.

classmethod set_log(l)

Set logger.

Parameters:

l (logger) – logger object

Return type:

None

shutdown()

Shut down all services, free resources, stop threads, disconnect from mqtt, in general, prepare for shutdown.

sort_by_power(solarpower, ts_utc_now)[source]

Sort forecast of solarpower to decreasing order.

Parameters:
  • solarpower (list) – list of entries describing hourly solar energy forecast

  • ts_utc_now (float) – curren time, for exluding entries that are in the past

Returns:

list from the highest solarenergy to lowest.

Return type:

list

sort_by_rank(hours, ts_utc_now)[source]

Sort the given electricity prices by their rank value. Given a list of electricity prices, return a sorted list from the cheapest to the most expensive hours. Entries that represent electricity prices in the past are excluded.

Parameters:
  • hours (list) – list of hourly electricity prices

  • ts_utc_now (float) – current time

Returns:

sorted list of electricity prices

Return type:

list

subscribe(topic)

Subscribe to the given MQTT topic.

This method sets up the subscription to the specified MQTT topic and registers the on_message() method as the callback for incoming messages.

Parameters:

topic (str) – The MQTT topic to subscribe to.

Return type:

None

Example:

# configure
obj.subscribe('foo/bar')
timestamp()

Returns the current date-time in UTC.

Returns:

datetime in UTC.

Return type:

rc (datetime)

timestampstr(ts)

Converts the given timestamp to human readable string of form ‘Y-m-d H:M:S’.

Parameters:

ts (timestamp) – time stamp to be converted

Returns:

human readable date-time string

Return type:

rc (string)

to_dict()

Convert instance attributes to a dictionary.

topic_forecast = 'juham/forecast'
topic_in_powerconsumption = 'juham/powerconsumption'
topic_power = 'juham/power'
topic_powerplan = 'juham/powerplan'
topic_spot = 'juham/spot'
topic_temperature = 'juham/temperature/102'
uoi_limit = 0.75
valid_topic(topic)

Check if the given topic has been registered as valid topic.

Parameters:

topic (str) – topic to be validated

Returns:

true if known topic.

Return type:

bool

warning(msg, details='')

Logs the given warning message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

write(point)

Writes the given measurement to the database. In case of an error, it tries again until the maximum number of attempts is reached. If it is still unsuccessful, it gives up and passes the first encountered exception to the caller.

Parameters:

point – a measurement describing a time stamp and related attributes for one measurement.

write_attempts = 3
class juham.automation.WaterCirculator(name='rwatercirculator')[source]

Bases: Base

Hot Water Circulation Controller.

Uses motion sensor data to determine if someone is at home. If so, it runs the water circulator pump to ensure that hot water is instantly available when the tap is turned on.

app_name = 'juham'
classmethod classattrs_from_dict(attributes)

Set class attributes from a dictionary.

classmethod classattrs_to_dict()

Convert class attributes to a dictionary.

config_folder = 'config'
copy()

Create and return a copy of the current object.

This method serializes the current object to a dictionary using the to_dict method, creates a new instance of the object’s class, and populates it with the serialized data using the from_dict method.

This method uses class identifier based instantiation (see factory method pattern) to create a new instance of the object, and ‘to_dict’ and ‘from_dict’ methods to initialize object’s state.

Returns:

A new instance of the object’s class with the same state as the original object.

Example:

john = Human("john")
john.height = 1.8
john.age = 62
clone_of_john = john.copy()
database_class_id: str = 0
debug(msg, details='')

Logs the given debug message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

deserialize_from_json(f)

Load attributes from a JSON file.

epoc2utc(epoch)

Converts the given epoch time to UTC time string. All time coordinates are represented in UTC time. This allows the time coordinate to be mapped to any local time representation without ambiguity.

Parameters:
  • epoch (float) – timestamp in UTC time

  • rc (str) – time string describing date, time and time zone e.g 2024-07-08T12:10:22Z

Returns:

UTC time

error(msg, details='')

Logs the given error message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

classmethod find_class(class_id)

Given class identifier find the registered class. If no class with the give identifier exists return None.

Parameters:

class_id (int) – class identifier

Returns:

class or null if not registered

Return type:

obj (obj)

from_dict(data)

Update instance attributes from a dictionary.

classmethod get_class_id()

Return the class id of the class. Each class has an unique identifier that can be used for instantiating the class via Object.instantiate() method.

Parameters:

cls (class) – class

Return type:

str

Returns:

id (int) unique class identifier through which the class can be instantiated by factory method pattern.

classmethod get_json_file()

Generate the JSON file name based on the class name.

The file is created into users home folder.

info(msg, details='')

Logs the given information message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

Example:

obj = new Base('test')
obj.info('Message arrived', str(msg))
init_database(name)

Instantiates the configured time series database object.

Issues a warning if the database_class_id has not been configured, in which case the object will not have the time series recording feature.

This method is called internally and typically there is no need to call it from the application code.

Return type:

None

init_mqtt(name)

Instantiates the configured MQTT object for networking.

This method is called internally and typically there is no need to call it from the application code.

Issues a warning if the pubsub_class_id has not been configured, even though objects without a capability to communicate are rather crippled.

Return type:

None

classmethod initialize_class(load_class_attrs=True)

Initialize the class for instantiation, if not initialized already. This method initializes the class identifier and deserializes the public attributes from the specified configuration folder.

Parameters:
  • load_class_attrs (bool, optional) – If true then attempts to initialize class attributes

  • True. (from the class specific configuration files. Defaults to)

Returns:

returns true if the class was initialized, false implies the class is already initialized in which case the method call has no effect.

Return type:

bool

classmethod instantiate(class_id)

Create an instance of the class corresponding to the given class identifier. This method implements the factory method pattern, which is essential for a plugin architecture.

Parameters:

class_id (int) – Identifier of the class to instantiate.

Returns:

An instance of the class corresponding to the given class identifier.

Return type:

obj

classmethod instantiate_with_param(class_id, param)

Given class identifier and one constructor argument create the corresponding object.

Parameters:
  • class_id (str) – class identifier

  • param (Any) – class specific constructor parameter

Returns:

instance of the given class.

Return type:

obj

classmethod is_abstract()

Check whether the class is abstract or real. Override in the derived sub-classes. The default is False.

Return type:

bool

Returns:

True (bool) if abstract

is_time_between(begin_time, end_time, check_time=None)

Check if the given time is within the given time line. All timestamps must be in UTC time.

Parameters:
  • begin_time (timestamp) – beginning of the timeline

  • end_time (timestamp) – end of the timeline

  • check_time (timestamp) – time to be checked

Returns:

True if within the timeline

Return type:

rc (bool)

classmethod load_from_json()

Load class attributes from a JSON file.

log_message(type, msg, details='')

Publish the given log message to the MQTT ‘log’ topic.

This method constructs a log message with a timestamp, class type, source name, message, and optional details. It then publishes this message to the ‘log’ topic using the MQTT protocol.

Parameters:
  • type – str The classification or type of the log message (e.g., ‘Error’, ‘Info’).

  • msg – str The main log message to be published.

  • details – str, optional Additional details about the log message (default is an empty string).

Returns:

None

Raises:

Exception – If there is an issue with the MQTT client while publishing the message.

Example:

# publish info message to the Juham's 'log' topic
self.log_message("Info", f"Some cool message {some_stuff}", str(dict))
min_temperature = 37
motion_sensor_topic = 'shellies/shellymotion2/info'
motion_topics = 'shellies/shellymotion2/#'
mqtt_class_id: str = ''
mqtt_host: str = 'localhost'
mqtt_port: int = 1883
mqtt_root_topic: str = 'juham'
on_connect(client, userdata, flags, rc)[source]

Notification on connect.

This method is called whenever the MQTT broker is connected. For more information on this method consult MQTT documentation available in many public sources.

Parameters:
  • client (obj) – MQTT client

  • userdata (Any) – application specific data

  • flags (int) – Consult MQTT

  • rc (int) – See MQTT docs

on_disconnect(client, userdata, rc=0)

Notification on disconnect.

This method is called whenever the MQTT broker is disconnected. For more information on this method consult MQTT documentation available in many public sources.

Parameters:
  • client (obj) – MQTT client

  • userdata (Any) – application specific data

  • rc (int) – See MQTT docs

on_message(client, userdata, msg)[source]

MQTT message notification on arrived message.

Called whenever a new message is posted on one of the topics the object has subscribed to via subscribe() method. This method is the heart of automation: here, derived subclasses should automate whatever they were designed to automate. For example, they could switch a relay when a boiler temperature sensor signals that the temperature is too low for a comforting shower for say one’s lovely wife.

For more information on this method consult MQTT documentation available in many public sources.

Parameters:
  • client (obj) – MQTT client

  • userdata (Any) – application specific data

  • msg (object) – The MQTT message

on_motion_sensor(m, ts_utc_now)[source]

Control the water cirulator bump.

Given message from the motion sensor consider switching the circulator bump on.

Parameters:
  • msg (dict) – directionary holding motion sensor data

  • ts_utc_now (float) – current time stamp

Return type:

None

on_temperature_sensor(m, ts_utc_now)[source]

Handle message from the hot water pipe temperature sensor.

Records the temperature and updates the water_temperature_updated attribute.

Parameters:
  • m (dict) – temperature reading from the hot water blump sensor

  • ts_utc_now (float) – _current utc time

Return type:

None

classmethod parse_args()

Parse the startup arguments defined by this class.

Return type:

None

publish(topic, msg, qos=1, retain=True)

Publish the given message to the given MQTT topic. For more information consult MQTT.

Parameters:
  • topic (str) – topic

  • msg (str) – message to be published

  • qos (int, optional) – quality of service. Defaults to 1.

  • retain (bool, optional) – retain. Defaults to True.

pubsub_class_id = 0
read(point)

Reads the given measurement from the database.

Parameters:

point – point with initialized time stamp.

… note: NOT IMPLEMENTED YET

classmethod register()[source]

Register the class to the class database.

Once registered the class can be instantiated by its class identifier. Note that this method is called automatically by the system when the python loads the class. In this method sub classes should prepare themselves for instantiation by initializing their class attributes, for example.

classmethod register_class(class_id, ctor)

Register the given class identifier for identifier based instantiation . This, factory method pattern, as it is called, decouples the actual implementation from the interface. For more information see instantiate() method.

Parameters:
  • class_id (str) – class identifier

  • ctor (function) – constructor

classmethod register_topic(topic)

Register the given topic as known valid topic. Publishing and subscriptions are allowed only to registered topics. Each topic must have an owner, who is responsible for registering the topic.

Parameters:

topic (str) – valid topic

relay_topic = 'shellyplus1-a0a3b3c309c4/command/switch:0'
run()

Start a new thread to runs the network loop in the background.

Allows the main program to continue executing while the MQTT client handles incoming and outgoing messages in the background.

run_forever()

Starts the network loop and blocks the main thread, continuously running the loop to process MQTT messages.

The loop will run indefinitely unless the connection is lost or the program is terminated.

classmethod save_to_json()

Create class configuration file, if the file does not exist yet.

serialize_to_json(f)

Serialize.

classmethod set_log(l)

Set logger.

Parameters:

l (logger) – logger object

Return type:

None

shutdown()

Shut down all services, free resources, stop threads, disconnect from mqtt, in general, prepare for shutdown.

subscribe(topic)

Subscribe to the given MQTT topic.

This method sets up the subscription to the specified MQTT topic and registers the on_message() method as the callback for incoming messages.

Parameters:

topic (str) – The MQTT topic to subscribe to.

Return type:

None

Example:

# configure
obj.subscribe('foo/bar')
temperature_topic = 'juham/temperature/103'
timestamp()

Returns the current date-time in UTC.

Returns:

datetime in UTC.

Return type:

rc (datetime)

timestampstr(ts)

Converts the given timestamp to human readable string of form ‘Y-m-d H:M:S’.

Parameters:

ts (timestamp) – time stamp to be converted

Returns:

human readable date-time string

Return type:

rc (string)

to_dict()

Convert instance attributes to a dictionary.

uptime = 300
valid_topic(topic)

Check if the given topic has been registered as valid topic.

Parameters:

topic (str) – topic to be validated

Returns:

true if known topic.

Return type:

bool

warning(msg, details='')

Logs the given warning message to the database after logging it using the BaseClass’s info() method.

Parameters:
  • msg (str) – The information message to be logged.

  • details (str) – Additional detailed information for the message to be logged

write(point)

Writes the given measurement to the database. In case of an error, it tries again until the maximum number of attempts is reached. If it is still unsuccessful, it gives up and passes the first encountered exception to the caller.

Parameters:

point – a measurement describing a time stamp and related attributes for one measurement.

write_attempts = 3