juham.base

Description

Base classes for Juham - Juha’s Ultimate Home Automation Masterpiece

This package represents the most low level layer in the framework. Most notably, it defines two essential abstractions on which communcation between various IoT nodes and the data tracking is based on:

  1. jmqtt - publish-subscriber model data transmission

  2. jdatabase - interface to time series database used for data recording

3. jobject - base class of everything, from which all the objects in the framework are derived.

class juham.base.Base(name='')[source]

Bases: Object

An automation object with MQTT networking, time series database and logging features.

This class is typically not instantiated directly, but used as a base class for automation object implementations. The object includes MQTT networking, logging, time series database recording, and event processing capabilities. It can be instantiated and run either via cron, a thread, or by any other means.

Example:

obj = Base("foo")
obj.mqtt_host = "myhost.com"
obj.subscribe('foo/bar')
obj.run()

To allow any MQTT or time series implementation to be used all objects of this kind support factory method pattern. All classes has unique class identifier class attribute through which they can be instantiated.

To configure Base class to use a specific MQTT and time series implementations set the class attributes to refer to desired MQTT and time series database implementations. When instantiated the object will instantiate the given MQTT and database objects with it.

A class that writes measurements to a database with retry functionality.

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 | None = 0
debug(msg, details='')[source]

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)[source]

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='')[source]

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)[source]

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='')[source]

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)[source]

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)[source]

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()

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.

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)

Given class identifier create the corresponding object. This method implements so called factory method pattern.

Parameters:

class_id (int) – class identifier

Returns:

instance of the given class.

Return type:

obj (obj)

classmethod instantiate_with_param(class_id, param)

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

Parameters:
  • class_id (str | None) – 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)[source]

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='')[source]

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 | None = None
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

Return type:

None

on_disconnect(client, userdata, rc=0)[source]

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

classmethod parse_args()

Parse the startup arguments defined by this class.

Return type:

None

pubsub_class_id = 0
read(point)[source]

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

run()[source]

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()[source]

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()[source]

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

subscribe(topic)[source]

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()[source]

Returns the current date-time in UTC.

Returns:

datetime in UTC.

Return type:

rc (datetime)

timestampstr(ts)[source]

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()[source]

Convert instance attributes to a dictionary.

warning(msg, details='')[source]

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)[source]

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.base.Group(name='group')[source]

Bases: Base

Group base class that can consist of Base and Group objects as children.

This class can be used for grouping home automation objects into larger logical entities.

Example:

motion_sensors = Group("motionsensors")
motion_sensors.add(ShellyMotionSensor("downstairs"))
motion_sensors.add(ShellyMotionSensor("upstairs"))
add(h)[source]

Add new automation object as children. The object to be inserted must be derived from Object base class.

Parameters:

h (Object) – object to be inserted.

Return type:

None

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 | None = 0
database_client: object | None
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)[source]

Recursively deserialize the group from a dictionary, including it children.

Parameters:

data (dict) – data to deserialize from.

Example:

from juham.base import Group, Base

# create a group with a child
group = Group("foo")
base = Base("bar")
group.add(base)

# group to dictionary
data = group.to_dict()

copy_of_group = Group()
copy_of_group.from_dict(data)
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()

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.

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)

Given class identifier create the corresponding object. This method implements so called factory method pattern.

Parameters:

class_id (int) – class identifier

Returns:

instance of the given class.

Return type:

obj (obj)

classmethod instantiate_with_param(class_id, param)

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

Parameters:
  • class_id (str | None) – 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 | None = None
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

classmethod parse_args()

Parse the startup arguments defined by this class.

Return type:

None

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

run()[source]

Start up all automation objects in the group and enter the network event loop to process MQTT messages.

The method returns only when the process is terminated by some means e.g. Ctrl+C or kill, or by sending ‘quit’ event to MQTT message broker.

Return type:

None

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()[source]

Convert instance attributes to a dictionary.

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.base.JDatabase(name)[source]

Bases: Object

The base class for data storage classes.Serves as an abstract interface for managing interactions with various types of databases. Designed to support multiple backend databases, this class provides a unified API for writing sensor data and other parameters, ensuring that the system can seamlessly integrate with different storage solutions.

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 = 'home'
debug(msg, details='')

Logs the given debug message to the application log.

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

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

Return type:

None

deserialize_from_json(f)

Load attributes from a JSON file.

error(msg, details='')

Logs the given error message to the application log.

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

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

Return type:

None

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_dict)[source]

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.

host: str = 'https://eu-central-1-1.aws.cloud2.influxdata.com'
info(msg, details='')

Logs the given information message to the application log.

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

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

Return type:

None

classmethod initialize_class()

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.

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)

Given class identifier create the corresponding object. This method implements so called factory method pattern.

Parameters:

class_id (int) – class identifier

Returns:

instance of the given class.

Return type:

obj (obj)

classmethod instantiate_with_param(class_id, param)

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

Parameters:
  • class_id (str | None) – 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

classmethod load_from_json()

Load class attributes from a JSON file.

org: str = 'juham'
classmethod parse_args()

Parse the startup arguments defined by this class.

Return type:

None

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 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

to_dict()[source]

Convert instance attributes to a dictionary.

token: str = 'Oxj1TY_3rvpPhhpNtOjhAHgJpgDcg-uR3b3__kwC2N7PNhWvpczQoOjnkx67t-XkK7HdPR7Vh2eePHsbKx-tGA=='
warning(msg, details='')

Logs the given warning message to the application log.

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

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

Return type:

None

write(point)[source]

Write record to database table.

@param point point to be written

class juham.base.JLog(name, level=10)[source]

Bases: Logger

Default logger for logging events to application log and/or to console.

addFilter(filter)

Add the specified filter to this handler.

addHandler(hdlr)

Add the specified handler to this logger.

callHandlers(record)

Pass a record to all relevant handlers.

Loop through all handlers for this logger and its parents in the logger hierarchy. If no handler was found, output a one-off error message to sys.stderr. Stop searching up the hierarchy whenever a logger with the “propagate” attribute set to zero is found - that will be the last logger whose handlers are called.

critical(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘CRITICAL’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.critical(“Houston, we have a %s”, “major disaster”, exc_info=True)

debug(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘DEBUG’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.debug(“Houston, we have a %s”, “thorny problem”, exc_info=True)

error(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘ERROR’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.error(“Houston, we have a %s”, “major problem”, exc_info=True)

exception(msg, *args, exc_info=True, **kwargs)

Convenience method for logging an ERROR with exception information.

fatal(msg, *args, **kwargs)

Don’t use this method, use critical() instead.

filter(record)

Determine if a record is loggable by consulting all the filters.

The default is to allow the record to be logged; any filter can veto this by returning a false value. If a filter attached to a handler returns a log record instance, then that instance is used in place of the original log record in any further processing of the event by that handler. If a filter returns any other true value, the original log record is used in any further processing of the event by that handler.

If none of the filters return false values, this method returns a log record. If any of the filters return a false value, this method returns a false value.

Changed in version 3.2: Allow filters to be just callables.

Changed in version 3.12: Allow filters to return a LogRecord instead of modifying it in place.

findCaller(stack_info=False, stacklevel=1)

Find the stack frame of the caller so that we can note the source file name, line number and function name.

getChild(suffix)

Get a logger which is a descendant to this one.

This is a convenience method, such that

logging.getLogger(‘abc’).getChild(‘def.ghi’)

is the same as

logging.getLogger(‘abc.def.ghi’)

It’s useful, for example, when the parent logger is named using __name__ rather than a literal string.

getChildren()
getEffectiveLevel()

Get the effective level for this logger.

Loop through this logger and its parents in the logger hierarchy, looking for a non-zero logging level. Return the first one found.

handle(record)

Call the handlers for the specified record.

This method is used for unpickled records received from a socket, as well as those created locally. Logger-level filtering is applied.

hasHandlers()

See if this logger has any handlers configured.

Loop through all handlers for this logger and its parents in the logger hierarchy. Return True if a handler was found, else False. Stop searching up the hierarchy whenever a logger with the “propagate” attribute set to zero is found - that will be the last logger which is checked for the existence of handlers.

info(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘INFO’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.info(“Houston, we have a %s”, “notable problem”, exc_info=True)

isEnabledFor(level)

Is this logger enabled for level ‘level’?

log(level, msg, *args, **kwargs)

Log ‘msg % args’ with the integer severity ‘level’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.log(level, “We have a %s”, “mysterious problem”, exc_info=True)

makeRecord(name, level, fn, lno, msg, args, exc_info, func=None, extra=None, sinfo=None)

A factory method which can be overridden in subclasses to create specialized LogRecords.

manager = <logging.Manager object>
removeFilter(filter)

Remove the specified filter from this handler.

removeHandler(hdlr)

Remove the specified handler from this logger.

root = <RootLogger root (WARNING)>
setLevel(level)

Set the logging level of this logger. level must be an int or a str.

warn(msg, *args, **kwargs)
warning(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘WARNING’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.warning(“Houston, we have a %s”, “bit of a problem”, exc_info=True)

class juham.base.JMqtt(name)[source]

Bases: Object

Abstract base class for MQTT brokers.

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'
connect_to_server(host='localhost', port=1883, keepalive=60)[source]

Connect to MQTT server

Parameters:
  • host (str, optional) – host. Defaults to “localhost”.

  • port (int, optional) – port. Defaults to 1883.

  • keepalive (int, optional) – keep alive, in seconds. Defaults to 60.

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()
debug(msg, details='')

Logs the given debug message to the application log.

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

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

Return type:

None

deserialize_from_json(f)

Load attributes from a JSON file.

disconnect_from_server()[source]

Disconnect from the MQTT broker.

It is up to the sub classes to implement the method.

error(msg, details='')

Logs the given error message to the application log.

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

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

Return type:

None

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 application log.

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

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

Return type:

None

classmethod initialize_class()

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.

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)

Given class identifier create the corresponding object. This method implements so called factory method pattern.

Parameters:

class_id (int) – class identifier

Returns:

instance of the given class.

Return type:

obj (obj)

classmethod instantiate_with_param(class_id, param)

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

Parameters:
  • class_id (str | None) – 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

classmethod load_from_json()

Load class attributes from a JSON file.

loop_stop()[source]

Stop the network loop.

No further messages shall be dispatched.

classmethod parse_args()

Parse the startup arguments defined by this class.

Return type:

None

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 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

to_dict()

Convert instance attributes to a dictionary.

warning(msg, details='')

Logs the given warning message to the application log.

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

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

Return type:

None

class juham.base.Object(name='noname')[source]

Bases: object

Base class of everything. Serves as the foundational class offering key features needed by any robust object-oriented software.

Logging:

All objects have logging methods e.g. info() and error() at their fingertips, for centralized logging.

if not self.do_good():
    self.error(f"Damn, did bad")

Factory Method Support:

Instantiation via class identifiers, adhering to the factory method pattern. This allows for the dynamic creation of instances based on identifier strings, promoting decoupled and extensible design required by plugin architecture.

# instead of fixed implementation car = Ferrari()
car = Object.instantiate(car_class_id)

Serialization:

Serialization of both class and instance attributes, serving as by means of configuration.

# serialize to json file
with open("ferrari.json", "w") as f:
    ferrari.serialize(f)

# deserialize
ferrari = Ferrari()
with open("ferrari.json", "r") as f:
    ferrari.deserialize(f)

Cloning:

Any object can be copied.

sphere = Sphere("mycar")
sphere.radius = 1.8
sphere.center = [0.1, 0.2, 0.3]

sphere2 = sphere.copy()

Class attributes should follow consistent naming convention where underscore prefix implies non-serializable attribute.

app_name = 'juham'
classmethod classattrs_from_dict(attributes)[source]

Set class attributes from a dictionary.

classmethod classattrs_to_dict()[source]

Convert class attributes to a dictionary.

config_folder = 'config'
copy()[source]

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()
debug(msg, details='')[source]

Logs the given debug message to the application log.

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

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

Return type:

None

deserialize_from_json(f)[source]

Load attributes from a JSON file.

error(msg, details='')[source]

Logs the given error message to the application log.

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

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

Return type:

None

classmethod find_class(class_id)[source]

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)[source]

Update instance attributes from a dictionary.

classmethod get_class_id()[source]

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()[source]

Generate the JSON file name based on the class name.

The file is created into users home folder.

info(msg, details='')[source]

Logs the given information message to the application log.

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

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

Return type:

None

classmethod initialize_class()[source]

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.

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)[source]

Given class identifier create the corresponding object. This method implements so called factory method pattern.

Parameters:

class_id (int) – class identifier

Returns:

instance of the given class.

Return type:

obj (obj)

classmethod instantiate_with_param(class_id, param)[source]

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

Parameters:
  • class_id (str | None) – class identifier

  • param (Any) – class specific constructor parameter

Returns:

instance of the given class.

Return type:

obj

classmethod is_abstract()[source]

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

classmethod load_from_json()[source]

Load class attributes from a JSON file.

classmethod parse_args()[source]

Parse the startup arguments defined by this class.

Return type:

None

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)[source]

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 save_to_json()[source]

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

serialize_to_json(f)[source]

Serialize.

classmethod set_log(l)[source]

Set logger.

Parameters:

l (logger) – logger object

Return type:

None

to_dict()[source]

Convert instance attributes to a dictionary.

warning(msg, details='')[source]

Logs the given warning message to the application log.

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

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

Return type:

None