Proxies

class proxies.HealthStatuses(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Enumerated types representing the health status of the sensor in accordance with the OpenAPI specification detailed in AppMAIS/api/openapi.yaml

CRITICAL = 2

NA = -1

OK = 0

UNKNOWN = -2

WARNING = 1

class proxies.SettingSectionProxy(setting_section_name: str, section_type: ConfigurationSections, beemon_config: Config, logger: Logger | Logger)

_abc_impl = <_abc._abc_data object>

abstract _section_specific_configuration_values() → Dict[str, str]

Todo

Docstrings.

_set_global_settings_with_section_overrides() → Tuple[timezone, str, str, datetime, datetime, int, int, bool, List[str]]

Retrieves the global settings from the [global] section of the beemon-config.ini file and then overrides any settings that are present in the section that this class is representing.

Returns:

• pytz_time_zone (datetime.tzinfo): The timezone to which capture_window_start_time and capture_window_end_time should be converted to when populating the recording start time queues.

• root_upload_dir (str): The absolute path to the upload directory on the specified 'host' machine in the ['ftp'] setting of the beemon-config.ini. This parameter is not expected to be overridden on a per-sensor basis.

• local_output_dir (str): The absolute path to the temporary data storage location on the Raspberry Pi to which data will be written prior to being uploaded to the specified 'host'. This parameter is not expected to be overridden on a per-sensor basis.

• capture_window_start_datetime (:class:datetime.datetime`): A datetime aware object which specifies what time to start recording data from all sensors. Note that this parameter may be overridden on a per-sensor basis.

• capture_window_end_datetime (datetime.datetime): A datetime aware object which specifies what time to start recording data from all sensors. Note that this parameter may be overridden on a per-sensor basis.

• capture_duration_seconds (int): How long each sensor should be instructed to record for (if applicable) in seconds. Note that this parameter may be overridden on a per-sensor basis.

• capture_interval_seconds (int): How long to wait (in seconds) before instructing all sensors to record again for the duration specified by capture_duration_seconds. Note that this parameter may be overridden on a per-sensor basis.

• auto_start (bool): Whether all sensors should be instructed to automatically record at the specified capture_window_start_time. If this flag is disabled, then recording commands are expected to be issued manually. Note that this parameter may be overridden on a per-sensor basis.

• overridden_settings (List[str]): A list of all global settings that are overridden in the specific settings section.

Return type:

Tuple[datetime.tzinfo, str, str, datetime.datetime, datetime.datetime, int, int, bool]

Raises:

NotImplementedError – This method will raise a NotImplementedError in the event that the specified pytz_time_zone in the beemon-config.ini does not yet have a corresponding datetime conversion method in datetimeutils.

_update_setting_on_proxy(setting_name: str, value: str | int | float | bool)

Updates a setting on the proxy object, to keep the settings on the object and in the configuration file in sync. If updating the pytz_time_zone, then the capture_window_start_time and capture_window_end_time will be updated to reflect the new timezone as well.

Parameters:

• setting_name (str) – The name of the setting to update.

• value (Union[str, int, float, bool]) – The value of the setting to update. Some settings may require conversion to the appropriate type before updating the setting.

property auto_start: bool

property capture_duration_seconds: int

property capture_interval_seconds: int

property capture_window_end_time: datetime

property capture_window_start_time: datetime

configuration_values() → Dict[str, str]

Retrieves the configuration values for the section. This method retrieves only the values that are present in the configuration file, not including any default values or values that are retrieved from the [global] section of the beemon-config.ini file. Should add any values that are present in the configuration file, as well as unpacking all of self._overridden_settings for all settings that override the global values.

Returns:

The key-value pairs of ‘setting_name’: ‘setting_value’. All values should be strings to match the configuration file as closely as possible. The objects associated with the values should be returned by the :meth:settings method.

Return type:

Dict[str, str]

get_settings_section() → Dict[str, str | int | float | bool]

Retrieves a particular setting section from the beemon-config.ini file.

Returns:

A dictionary of the settings in the section.

Return type:

Dict[str, Union[str, int, float, bool]]

property local_output_directory: str

property logger: Logger | Logger

property pytz_time_zone: timezone

property root_upload_directory: str

property setting_section_name: str

abstract settings() → Dict[str, Any]

Retrieves the settings for the section. This method retrieves all the settings for the section, including any default values or values that are retrieved from the [global] section of the beemon-config.ini file.

Returns:

The key-value pairs of ‘setting_name’: ‘setting_value’.

Return type:

Dict[str, any]

update_setting(setting: str, value: str | int | float | bool)

Updates a setting in the beemon-config.ini under a particular section.

Parameters:

• setting (str) – The setting to update.

• value (Union[str, int, float, bool]) – The value for the setting to update.

update_settings_section(settings: Dict[str, str | int | float | bool])

Updates multiple settings in the beemon-config.ini under a particular section.

Parameters:

settings (Dict[str, Union[str, int, float, bool]]) – A dictionary of setting names and setting values to update.

update_settings_section_in_memory(settings: Dict[str, str | int | float | bool])

Updates the settings in memory for a particular section. This method is useful for updating the settings in memory without writing to the configuration file. Be Careful when using this method, as any crash could cause the settings changes to be lost. This should only be used when attempting a bulk write, with the Config write method called as soon as all changes are staged.

Parameters:

settings (Dict[str, Union[str, int, float, bool]]) – A dictionary of setting names and setting values to update.

class proxies.SensorProxy(setting_section_name: str, section_type: ConfigurationSections, beemon_config: Config, logger: Logger | Logger)

property auto_start: bool

property capture_duration_seconds: int

property capture_interval_seconds: int

property capture_window_end_time: datetime

property capture_window_start_time: datetime

property dashboard_access_token: str

property dashboard_host: str

get_all_recording_start_times_within_window(capture_window_start_time: datetime, capture_window_end_time: datetime, capture_duration_seconds: float, capture_interval_seconds: float) → List[datetime]

Computes the maximum number of recordings which can occur between the the range of datetime objects (inclusive), given that each recording has the specified interval of: capture_interval_seconds. This method then returns a list of datetime objects (non-naive and EDT timezone aware) which constitute the times at which recording should begin in order to successfully record a maximum number of times between the supplied range of times.

Parameters:

• capture_window_start_time (datetime) – The starting datetime (non-naive and inclusive) at which the first recording session (of capture_duration_seconds length in seconds) is to begin at.

• capture_window_end_time (datetime) – The ending datetime (non-naive and inclusive) at which the last recording session (of capture_duration_seconds length in seconds) is to begin at.

• capture_duration_seconds (float) – The length/duration that each recording session between capture_window_start_time and capture_window_end_time should be. This method will produce the times which recording commands should be issued to all registered Sensor concrete subclass objects.

• capture_interval_seconds (float) – How frequently to initialize recording sessions. For instance, a value of 120.0 seconds indicates that recording should be performed every two minutes from capture_window_start_time to capture_window_end_time (inclusive) for a duration of capture_duration_seconds.

Returns:

Todo

Docstring.

Return type:

list[datetime]

get_closest_previous_recording_start_time(a_time: datetime) → datetime | None

Returns the closest previous recording start time (inclusive) to the specified time. This can be useful for determining if a sensor should be currently recording.

Parameters:

a_time (datetime) – The time to compare against the previous recording start times (inclusive).

Returns:

The closest previous recording time to the specified time, or None if there were no previous recording times to the time provided.

Return type:

Optional[datetime]

get_closest_recording_start_time(a_time: datetime)

Returns the closest recording start time to the current time (inclusive).

Returns:

The closest recording start time to the current time.

Return type:

datetime

get_closest_recording_time_with_tolerance(time: datetime, tolerance_seconds: int)

Returns the closest recording time to the specified time within a tolerance of tolerance_seconds. This can be useful for determining the closest recording time if the sensor is already recording.

Todo

Docstrings.

get_elapsed_recording_start_times(a_time: datetime, inclusive: bool) → List[datetime | None]

Returns all recording start times for the sensor that were to have occurred before the specified time. This method is optionally exclusive, in which case it will not return a recording start time that occurred at the specified time.

Parameters:

• a_time (datetime) – The time to compare against the recording start times (exclusive).

• inclusive (bool) – If True, the recording start time that occurred at the specified time will be included in the returned list. If False, the recording start time that occurred at the specified time will not be included in the returned list.

Returns:

A list of recording start times that have occurred before the provided time. The returned list will be empty if there are no recording start times that have occurred before the provided time.

Return type:

List[Optional[datetime]]

get_health_status() → HealthStatuses

Returns the health status of the sensor by reading the CSV in the output directory of the sensor. The health status of the sensor is defined in the OpenAPI specification detailed in AppMAIS/api/openapi.yaml.

Returns:

The health status of the sensor.

Return type:

HealthStatuses

get_last_n_successful_recording_values(num_recording_values: int, sensor_data_dict: Dict[str, float | Dict[str, float]] | None = None) → List[float | Dict[str, float]]

Retrieves the last num_recording_values successful recording values from the sensor’s CSV file.

Parameters:

• num_recording_values (int) – The number of recording values to retrieve from the sensor’s CSV file.

• sensor_data_dict (Optional[Dict[str, Union[float, Dict[str, float]]]]) – The dictionary of the sensor’s CSV file data. If None, then the CSV file will be read. Provided as an argument such that this method can be called multiple times without having to re-read the CSV file.

Returns:

Ordered list of last num_recording_values recordings. This is based on the number of anticipated recordings, so any missed or non-existent recordings will mean less than num_recording_values values will be returned. Ordered by most recent first.

Return type:

List[Union[float, Dict[str, float]]

get_number_of_successful_recordings_on_date(date: ~.datetime.date) → int

Retrieves the number of successful (non None) recordings present in the sensor’s CSV file for the specified date.

Parameters:

date (date) – The date of recording for the sensor to read the CSV from.

Returns:

The number of successful (non None) recordings present in the sensor’s CSV file for the specified date.

Return type:

int

property local_output_directory: str

abstract read_csv(date: ~.datetime.date) → Dict[str, Any] | None

Reads the CSV file for the sensor on the specified date and returns the rows of the CSV in a dictionary, indexed by the recording time.

Parameters:

date (date) – The date of recording for the sensor to read the CSV from.

Returns:

A dictionary of the rows in the CSV file, indexed by the recording time. If the CSV file does not exist, None will be returned.

Return type:

Optional[Dict[str, Any]]

should_be_recording() → bool

Determines if the sensor should be currently recording.

Returns:

True if the sensor should be currently recording, False otherwise.

Return type:

bool

should_have_recorded() → bool

Determines if the sensor should have recorded by the current time.

Returns:

True if the sensor should have recorded by the current time, False otherwise.

Return type:

bool

class proxies.GlobalProxy(beemon_config: Config, logger: Logger | Logger)

settings() → Dict[str, str | int | float | bool]

Retrieves the settings for the section. This method retrieves all the settings for the section, including any default values or values that are retrieved from the [global] section of the beemon-config.ini file.

Returns:

The key-value pairs of ‘setting_name’: ‘setting_value’.

Return type:

Dict[str, any]

class proxies.SFTPProxy(beemon_config: Config, logger: Logger | Logger)

property host: str

property port: int

property private_key_file_path: str

settings() → Dict[str, Any]

Retrieves the settings for the section. This method retrieves all the settings for the section, including any default values or values that are retrieved from the [global] section of the beemon-config.ini file.

Returns:

The key-value pairs of ‘setting_name’: ‘setting_value’.

Return type:

Dict[str, any]

property upload_delay: int

property username: str

class proxies.DashboardProxy(beemon_config: Config, logger: Logger | Logger)

property access_token: str

property host: str

settings() → Dict[str, Any]

Retrieves the settings for the section. This method retrieves all the settings for the section, including any default values or values that are retrieved from the [global] section of the beemon-config.ini file.

Returns:

The key-value pairs of ‘setting_name’: ‘setting_value’.

Return type:

Dict[str, any]

class proxies.LogProxy(beemon_config: Config, logger: Logger | Logger)

property log_cycle_days: int

property logging_level: str

property max_archived_files: int

property max_file_size_megabytes: int

settings() → Dict[str, Any]

Retrieves the settings for the section. This method retrieves all the settings for the section, including any default values or values that are retrieved from the [global] section of the beemon-config.ini file.

Returns:

The key-value pairs of ‘setting_name’: ‘setting_value’.

Return type:

Dict[str, any]

class proxies.AudioProxy(beemon_config: Config, logger: Logger | Logger)

property channel_count: int

property gain: int

read_csv(date: ~.datetime.date) → Dict[str, bool | None] | None

Reads the CSV file for the sensor on the specified date and returns the rows of the CSV in a dictionary, indexed by the recording time.

Parameters:

date (date) – The date of recording for the sensor to read the CSV from.

Returns:

A dictionary of the rows in the CSV file, indexed by the recording time. If the CSV file does not exist, None will be returned.

Return type:

Optional[Dict[str, Any]]

property sample_format: int

property sampling_frequency: int

settings() → Dict[str, int | bool]

Retrieves the settings for the section. This method retrieves all the settings for the section, including any default values or values that are retrieved from the [global] section of the beemon-config.ini file.

Returns:

The key-value pairs of ‘setting_name’: ‘setting_value’.

Return type:

Dict[str, any]

class proxies.VideoProxy(beemon_config: Config, logger: Logger | Logger)

property flip_video: bool

property frames_per_second: int

read_csv(date: ~.datetime.date) → Dict[str, float | None] | None

Reads the CSV file for the sensor on the specified date and returns the rows of the CSV in a dictionary, indexed by the recording time.

Parameters:

date (date) – The date of recording for the sensor to read the CSV from.

Returns:

A dictionary of the rows in the CSV file, indexed by the recording time. If the CSV file does not exist, None will be returned.

Return type:

Optional[Dict[str, Any]]

property resolution_x: int

property resolution_y: int

settings() → Dict[str, int | bool]

Retrieves the settings for the section. This method retrieves all the settings for the section, including any default values or values that are retrieved from the [global] section of the beemon-config.ini file.

Returns:

The key-value pairs of ‘setting_name’: ‘setting_value’.

Return type:

Dict[str, any]

class proxies.AirQualityProxy(beemon_config: Config, logger: Logger | Logger)

read_csv(date: ~.datetime.date) → Dict[str, Any] | None

Reads the CSV file for the sensor on the specified date and returns the rows of the CSV in a dictionary, indexed by the recording time.

Parameters:

date (date) – The date of recording for the sensor to read the CSV from.

Returns:

A dictionary of the rows in the CSV file, indexed by the recording time. If the CSV file does not exist, None will be returned.

Return type:

Optional[Dict[str, Any]]

settings() → Dict[str, Any]

Retrieves the settings for the section. This method retrieves all the settings for the section, including any default values or values that are retrieved from the [global] section of the beemon-config.ini file.

Returns:

The key-value pairs of ‘setting_name’: ‘setting_value’.

Return type:

Dict[str, any]

class proxies.TempProxy(beemon_config: Config, logger: Logger | Logger)

read_csv(date: ~.datetime.date) → Dict[str, Dict[str, float | None]] | None

Reads the CSV file for the sensor on the specified date and returns the rows of the CSV in a dictionary, indexed by the recording time.

Parameters:

date (date) – The date of recording for the sensor to read the CSV from.

Returns:

A dictionary of the rows in the CSV file, indexed by the recording time. If the CSV file does not exist, None will be returned.

Return type:

Optional[Dict[str, Any]]

settings() → Dict[str, Any]

Retrieves the settings for the section. This method retrieves all the settings for the section, including any default values or values that are retrieved from the [global] section of the beemon-config.ini file.

Returns:

The key-value pairs of ‘setting_name’: ‘setting_value’.

Return type:

Dict[str, any]

class proxies.ScaleProxy(beemon_config: Config, logger: Logger | Logger)

read_csv(date: ~.datetime.date) → Dict[str, float | None] | None

Reads the CSV file for the sensor on the specified date and returns the rows of the CSV in a dictionary, indexed by the recording time.

Parameters:

date (date) – The date of recording for the sensor to read the CSV from.

Returns:

A dictionary of the rows in the CSV file, indexed by the recording time. If the CSV file does not exist, None will be returned.

Return type:

Optional[Dict[str, Any]]

settings() → Dict[str, float]

Retrieves the settings for the section. This method retrieves all the settings for the section, including any default values or values that are retrieved from the [global] section of the beemon-config.ini file.

Returns:

The key-value pairs of ‘setting_name’: ‘setting_value’.

Return type:

Dict[str, any]

property weight_multiple: float

property weight_offset: float

class proxies.CPUProxy(beemon_config: Config, logger: Logger | Logger)

read_csv(date: ~.datetime.date) → Dict[str, float | None] | None

Reads the CSV file for the sensor on the specified date and returns the rows of the CSV in a dictionary, indexed by the recording time.

Parameters:

date (date) – The date of recording for the sensor to read the CSV from.

Returns:

A dictionary of the rows in the CSV file, indexed by the recording time. If the CSV file does not exist, None will be returned.

Return type:

Optional[Dict[str, Any]]

settings() → Dict[str, Any]

Retrieves the settings for the section. This method retrieves all the settings for the section, including any default values or values that are retrieved from the [global] section of the beemon-config.ini file.

Returns:

The key-value pairs of ‘setting_name’: ‘setting_value’.

Return type:

Dict[str, any]