import os
import pytz
from loguru import logger
from typing import Dict, Type, Optional, Tuple, Union
import importlib
import inspect
from datetime import datetime, tzinfo
from distutils.util import strtobool
from src.beemon.configuration import Config
from src.utils import datetimeutils
[docs]
def parse_global_sensor_settings_from_config_file(
beemon_config: Config, sensor_name: Optional[str] = None, sensor_override: Optional[bool] = True) -> \
Tuple[tzinfo, str, str, datetime, datetime, int, int, bool]:
"""
A helper method which parses sensor-specific settings from the ``beemon-config.ini`` file, while defaulting to the
``[global]`` settings specified in the ``beemon-config.ini``. Additionally, this method allows for the Sensor's
settings to take precedence over the global settings if the ``sensor_override`` flag is set to ``True``.
Notes:
This method will only return the configuration file arguments which are intended to be overridden by individual
sensors. For instance, although ``pytz_time_zone`` is an argument in the ``[global]`` section of the
``beemon-config.ini`` this method will not return it; as individual sensors should not have the ability to
override this value. The same is true of ``root_upload_directory``, which is ignored by this method for the same
reason.
See Also:
The utility/helper method invoked by this method :meth:`~configuration.Config.parse_global_config`.
Args:
beemon_config (Config): An instance of the :class:`~configuration.Config` class.
sensor_name (Optional[str]): The name of the :class:`~sensor.Sensor` whose configuration file arguments should be
parsed. If no sensor name is provided, this method will simply return the parsed ``[global]`` section
of the ``beemon-config.ini`` configuration file. If a sensor name is provided and the ``sensor_override`` flag
is set to ``True`` then this method will override any settings specified under the ``[global]`` section, with
the settings specified under the sensor's section in the ``beemon-config.ini`` file. If the
``sensor_override`` flag is set to ``False`` then this method will simply return the parsed ``[global]``
section of the ``beemon-config.ini``.
sensor_override (Optional[bool]): Specifies if attributes declared in the ``beemon-config.ini`` file's
``[global]`` section should be overridden by the concrete sensor subclass's section in the same
``beemon-config.ini`` file. This value is ``True`` by default, indicating that the sensor's configuration
settings should take precedence over the global configuration. Please note that if the ``sensor_name`` is not
provided, then this parameter will be set to ``False`` automatically, as there is no way to override global
settings for an unspecified sensor.
Returns:
Tuple[str, str, str, :class:`~datetime.datetime`, :class:`~datetime.datetime`, int, int, bool]:
- **pytz_time_zone** (*:class:`datetime.tzinfo`*): The `pytz` timezone in which the RPI is based.
- **root_upload_directory** (*str*): The root directory on the remote machine to which data should be uploaded
to.
- **sensor_local_output_directory** (*str*): The local directory on the Pi to which recorded data should be
written to prior to uploading. By default this value is: ``/home/bee/bee_tmp``.
- **capture_window_start_time** (:class:`~datetime.datetime`): A tz-aware :class:`datetime.datetime` object which
specifies when the first recording for the day should occur for the provided ``sensor_name``. Note that this
datetime will be tz-aware in accordance with the timezone/locale specified by the OS of the RPI itself (e.g.
that which is given by the command: ``sudo raspi-config``).
- **capture_window_end_time** (:class:`datetime.datetime`): A tz-aware :class:`datetime.datetime` object which
specifies when the last recording for the day should occur for the provided ``sensor_name``. Note that this
datetime will be tz-aware in accordance with the timezone/locale specified by the OS of the RPI itself (e.g.
that which is given by the command: ``sudo raspi-config``).
- **sensor_capture_duration_seconds** (*int*): The number of seconds applicable sensors are to record for. For
sensors which read instantaneously, this variable should have no discernible effect. For sensors which record
in a loop (e.g. a video camera) this variable dictates how long the sensor is to consecutively record for.
- **sensor_capture_interval_seconds** (*int*): The number of seconds which will be permitted to elapse before
the next recording. Please note that it does not make sense for the ``sensor_capture_duration_seconds`` to
exceed the ``sensor_capture_interval_seconds``. Such a configuration (if specified) will be parsed without
error, but downstream logic will raise an exception.
- **sensor_auto_start** (*bool*): A boolean value indicating if the sensor should start recording automatically,
or if it should wait to receive a manual ``start`` command via the ``bmon`` client.
Raises:
AttributeError: Raises an ``AttributeError`` in the event that the sensor is not formally declared in the
``beemon-config.ini``.
"""
if sensor_name is None:
sensor_override = False
(pytz_time_zone, global_root_upload_dir, global_sensor_local_output_dir,
global_sensor_capture_window_start_datetime, global_sensor_capture_window_end_datetime,
global_sensor_capture_duration_seconds, global_sensor_capture_interval_seconds, global_sensor_auto_start) = beemon_config.parse_global_config()
if sensor_override:
# .. todo: MAKE SURE THIS IS MULTIPROCESSING SAFE:
if sensor_name not in beemon_config.instance().configuration:
error_message = f"Cannot override the [global] settings in \"beemon-config.ini\" with settings for " \
f"Sensor [{sensor_name.lower()}], as the Sensor named: {sensor_name.lower()} is not a " \
f"section of the configuration file: {beemon_config}."
logger.critical(error_message)
raise AttributeError(error_message)
# There is a custom config section in :class:`configuration.Config` for this sensor.
# Attempt to parse the sensor's attributes first:
lower_case_sensor_name: str = sensor_name.lower()
# See if the 'sensor_local_output_directory' is overridden by this sensor:
sensor_local_output_dir: str = beemon_config.instance().get(
section=lower_case_sensor_name,
key='local_output_directory',
default=global_sensor_local_output_dir
)
# See if the 'sensor_capture_duration_seconds' is overridden by this sensor:
sensor_capture_duration_seconds: Union[str, int] = beemon_config.instance().get(
section=lower_case_sensor_name,
key='capture_duration_seconds',
default=global_sensor_capture_duration_seconds
)
if type(sensor_capture_duration_seconds) is str:
sensor_capture_duration_seconds = int(sensor_capture_duration_seconds)
# See if the 'sensor_capture_interval_seconds' is overridden by this sensor:
sensor_capture_interval_seconds: Union[str, int] = beemon_config.instance().get(
section=lower_case_sensor_name,
key='capture_interval_seconds',
default=global_sensor_capture_interval_seconds
)
if type(sensor_capture_interval_seconds) is str:
sensor_capture_interval_seconds = int(sensor_capture_interval_seconds)
# See if the 'capture_window_start_time' is overridden by this sensor:
sensor_capture_window_start_datetime: Union[str, datetime] = beemon_config.instance().get(
section=lower_case_sensor_name,
key='capture_window_start_time',
default=global_sensor_capture_window_start_datetime
)
# Check for sensor-level override, and assign to global default if necessary:
if type(sensor_capture_window_start_datetime) is str:
# Parse raw sensor overridden start time into tz-aware datetime object.
if pytz_time_zone.zone == 'US/Eastern':
sensor_capture_window_start_datetime: datetime = datetimeutils.military_time_to_edt_datetime_aware(
military_time=sensor_capture_window_start_datetime
)
elif pytz_time_zone.zone == 'Europe/Brussels':
sensor_capture_window_start_datetime: datetime = datetimeutils.military_time_to_cest_datetime_aware(
military_time=sensor_capture_window_start_datetime
)
else:
not_implemented_error_message = f'The timezone: \'{pytz_time_zone}\' specified in the ' \
f'\'beemon-config.ini\' [global] section was ' \
f'recognized as a valid timezone, but we have not implemented support ' \
f'for it yet.'
raise NotImplementedError(not_implemented_error_message)
# See if the 'capture_window_end_time' is overridden by this sensor:
sensor_capture_window_end_datetime: Union[str, datetime] = beemon_config.instance().get(
section=lower_case_sensor_name,
key='capture_window_end_time',
default=global_sensor_capture_window_end_datetime
)
# Check for sensor-level override, and assign to global default if necessary:
if type(sensor_capture_window_end_datetime) is str:
if pytz_time_zone.zone == 'US/Eastern':
sensor_capture_window_end_datetime: datetime = datetimeutils.military_time_to_edt_datetime_aware(
military_time=sensor_capture_window_end_datetime
)
elif pytz_time_zone.zone == 'Europe/Brussels':
sensor_capture_window_end_datetime: datetime = datetimeutils.military_time_to_cest_datetime_aware(
military_time=sensor_capture_window_end_datetime
)
else:
not_implemented_error_message = f'The timezone: \'{pytz_time_zone}\' specified in the ' \
f'\'beemon-config.ini\' [global] section was ' \
f'recognized as a valid timezone, but we have not implemented ' \
f'support for it yet.'
raise NotImplementedError(not_implemented_error_message)
# See if the 'sensor_auto_start' flag is overridden by this sensor:
sensor_auto_start: Union[str, bool] = beemon_config.instance().get(
section=lower_case_sensor_name,
key='auto_start',
default=global_sensor_auto_start
)
if type(sensor_auto_start) is str:
logger.debug(
f'Sensor \'{lower_case_sensor_name}\' inheriting global sensor_auto_start flag as: '
f'{global_sensor_auto_start}'
)
sensor_auto_start: bool = bool(strtobool(sensor_auto_start))
return (
pytz_time_zone, global_root_upload_dir, sensor_local_output_dir,
sensor_capture_window_start_datetime, sensor_capture_window_end_datetime,
sensor_capture_duration_seconds, sensor_capture_interval_seconds, sensor_auto_start)
else:
return (
pytz_time_zone, global_root_upload_dir, global_sensor_local_output_dir,
global_sensor_capture_window_start_datetime, global_sensor_capture_window_end_datetime,
global_sensor_capture_duration_seconds, global_sensor_capture_interval_seconds, global_sensor_auto_start
)
def get_class_in_module(module_name: str, class_name: str) -> Tuple[Optional[Type], Optional[str]]:
cls: Optional[Type] = None
error_message: Optional[str] = None
try:
module = importlib.import_module(module_name)
except ModuleNotFoundError as err:
error_message = f"Failed to identify the module named: {module_name}. Error: {err}"
logger.error(error_message)
return cls, error_message
try:
# Retrieve the class dynamically using the provided name:
cls = getattr(module, class_name)
except AttributeError as err:
error_message = f"Failed to retrieve class: {class_name} from the specified module: {module_name}. " \
f"Received: {err}"
logger.error(error_message)
return cls, error_message
[docs]
def get_classes_in_module(module_name: str, package_name: str, concrete_only: bool) -> Dict[str, Type]:
"""
Dynamically imports and identifies Python classes in the specified module and package.
Args:
module_name (str): The name of the python module (e.g. 'sensor') whose classes should be imported.
package_name (str): The name of the python package containing the specified module (e.g. 'src.beemon').
concrete_only (bool): If True, only concrete subclasses of the python class which matches the specified module
name will be imported. For instance, if this flag is set to ``True`` then :class:`sensor.Sensor` will be
ignored during the dynamic import, but :class:`sensor.Audio` and class:`sensor.Video` will not be ignored
since they are concrete subclasses of :class:`sensor.Sensor`.
Returns:
Dict[str, Type]: A dictionary of class names and class types.
Notes:
This method returns a dictionary of **class** types (**NOT** instance types). If you wish to use the object, you
must instantiate the object by calling the class's constructor yourself.
"""
logger.debug(f"Importing classes from parent \'{module_name}\' module...")
# Get a list of all known sensor classes in the sensor.py parent module:
parent_module = importlib.import_module(name=module_name, package=package_name)
classes: Dict[str, Type] = {}
base_class: Optional[Type] = None
# First identify the parent/base class:
for name, obj in inspect.getmembers(parent_module):
# logger.debug(f"member name: {name}")
if inspect.isclass(obj):
# The inspected object is a python class.
parent_module_name = parent_module.__name__.split('.')[-1]
parent_class_name = parent_module_name.title()
# logger.debug(f"parent_module: {parent_module} parent_module name: {parent_module.__name__}")
if obj.__name__ == parent_class_name:
# This is the parent class (e.g. 'Sensor' in module 'sensor.py'
base_class = obj
break
if base_class is None:
# Failed to identify the parent/base class in the specified module.
logger.warning(f"Failed to identify the parent/base class: {parent_module.__name__.title()} "
f"in provided module: {module_name}")
else:
if not concrete_only:
classes[base_class.__name__] = base_class
# Identify all concrete subclasses of the parent/base class:
for name, obj in inspect.getmembers(parent_module):
if inspect.isclass(obj):
# This is a class object.
if issubclass(obj, base_class):
# This is a subclass of the identified base class.
if not concrete_only:
classes[obj.__name__] = obj
else:
if obj.__name__.title() != base_class.__name__ and obj.__name__.title() != 'Process':
classes[obj.__name__] = obj
return classes
def parse_information_from_file_path(file_path: str) -> Tuple[str, str, str]:
path, filename = os.path.split(file_path)
path, date = os.path.split(path)
if date == 'temp':
date = filename.split('.')[0]
sensor_name = 'temp'
else:
path, sensor_name = os.path.split(path)
return filename, date, sensor_name
if __name__ == '__main__':
pass
# parse_global_settings_from_config_file(beemon_config=configuration.Config(), sensor_name='Audio', sensor_override=True)
# logger.debug(f"syspath: {sys.path}")
# sensor_classes = get_classes_in_module(module_name='src.beemon.sensor', package_name='src.beemon', concrete_only=True)
# print(f"sensor_classes: {sensor_classes}")