"""
Contains functions that get called on server start and on creation respectively destruction of a client side session.
"""
import os
import sys
import shutil
import RPi.GPIO as GPIO
import yaml
import json
from attrd import AttrDict
from colorama import Fore
# used to set absolut path for definition of sensors module
sys.path.insert(0, os.path.abspath('.'))
from sensors.Utilities.initdevices import InitDevices
[docs]def load_config_file(output_terminal):
"""Function gets called on server load respectively session creation, and checks for possible custom
configuration `yaml-files`, loads and returns the corresponding dictionary.
:param output_terminal: flag for print output in terminal
:type output_terminal: boolean
:return: dictionary containing various parameters for sensors
:rtype: YAML dictionary
"""
if not os.path.exists('sensors/custom_config_files'):
os.makedirs('sensors/custom_config_files')
custom_config_filename = 'custom_config.yaml'
dict_filename = {"custom_config_filename": custom_config_filename}
path_to_config = os.path.join('sensors/custom_config_files/', custom_config_filename)
with open('sensors/custom_config_files/load_custom_config.json', 'w') as outfile:
json.dump(dict_filename, outfile, indent=4, ensure_ascii=False)
shutil.copyfile('sensors/templates/config.yaml', path_to_config)
with open(path_to_config) as f:
config = AttrDict(yaml.safe_load(f))
f.close()
print(Fore.RED + 'A new customizable configuration file ' + Fore.GREEN + custom_config_filename +
Fore.RED + ' has been created in /sensors/custom_config_files! Please provide additional information'
' about this multisensor system in this configuration file!' + Fore.WHITE)
elif os.path.exists('sensors/custom_config_files/load_custom_config.json'):
with open('sensors/custom_config_files/load_custom_config.json', encoding='UTF-8') as f:
filename_dict = json.load(f)
f.close()
custom_config_filename = filename_dict["custom_config_filename"]
path_to_config = os.path.join('sensors/custom_config_files/', custom_config_filename)
if os.path.exists(path_to_config):
if output_terminal:
print('{} {}!'.format('Loaded custom configuration file' + Fore.GREEN,
custom_config_filename + Fore.WHITE))
else:
print(Fore.RED + 'The configuration file specified in load_custom_config.json does not exist,'
' loaded default configuration file!\n'
'A new customizable configuration file ' + Fore.GREEN + custom_config_filename + Fore.RED +
' has been created in /sensors/custom_config_files! Please provide additional information'
' about this multisensor system in this configuration file!' + Fore.WHITE)
shutil.copyfile('sensors/templates/config.yaml', path_to_config)
with open(path_to_config) as f:
config = AttrDict(yaml.safe_load(f))
f.close()
else:
custom_config_filename = 'custom_config.yaml'
dict_filename = {"custom_config_filename": custom_config_filename}
path_to_config = os.path.join('sensors/custom_config_files/', custom_config_filename)
with open('sensors/custom_config_files/load_custom_config.json', 'w') as outfile:
json.dump(dict_filename, outfile, indent=4, ensure_ascii=False)
shutil.copyfile('sensors/templates/config.yaml', path_to_config)
with open(path_to_config) as f:
config = AttrDict(yaml.safe_load(f))
f.close()
print(Fore.RED + 'A new customizable configuration file ' + Fore.GREEN + custom_config_filename + Fore.RED +
' has been created in /sensors/custom_config_files! Please provide additional information'
' about this multisensor system in this configuration file!' + Fore.WHITE)
config.custom_config_filename = custom_config_filename
return config
[docs]def on_server_loaded(server_context):
"""Function gets called when a Bokeh server is instantiated. It loads the `config.yaml` information \
as dictionary and add it to Bokeh's server_context attribute. If a custom config filename is specified and exists\
in `custom_config_path` its information is used instead of the `config.yaml` information. Initializes the \
devices and corresponding attributes are added to the config object to pass it to the Bokeh server instance.
"""
config = load_config_file(output_terminal=False)
# pass config to server_context object, initialize and add device objects to config object
server_context.config = config
server_context.all_devices = InitDevices(config)
GPIO.setmode(GPIO.BCM)
GPIO.setup(config.GPIO_Switch, GPIO.OUT)
GPIO.output(config.GPIO_Switch, GPIO.LOW)
pass
[docs]def on_server_unloaded(server_context):
"""Function gets called when a Bokeh server cleanly exits. Sets GPIO switch for LED to OFF.
"""
GPIO.output(server_context.config.GPIO_Switch, GPIO.LOW)
pass
[docs]def on_session_created(session_context):
"""Function gets called when a new session is created. It gets the number of the currently opened sessions. The
first opened session is flagged as master session and gains controls over the sensor devices. A flag for the
master session and a counter for added ViewerTabs is added to session_context to pass it to the session
instance. It loads the `config.yaml` information and overwrites the config in the server_context attribute.
If a custom config filename is specified and exists in `custom_config_path` its information is used instead of \
the `config.yaml` information.
"""
num_sessions = len(session_context.server_context.sessions)
if num_sessions == 0:
print('First active session opened! Sensor control available!')
master_session = True
else:
print(num_sessions, 'other active session(s) opened! No sensor control available!')
master_session = False
# True for first session, passed to main.py
session_context.request.master_session = master_session
# Reset counter on ViewerTab when new session is opened, passed to main.py
session_context.request.viewer_counter = 0
# overwrite keys of configuration file to take effect in new session
session_config = load_config_file(output_terminal=True)
for key in session_config:
session_context.server_context.config[key] = session_config[key]
pass
[docs]def on_session_destroyed(session_context):
"""Function gets called when a session is closed. If the master session gets closed the PyAudiohandler thread
will be killed and the Picamera recording stopped, if running.
"""
num_sessions = len(session_context.server_context.sessions)
try:
session_context.server_context.config.camera.stop_recording()
print('Stopped PiCamera recording!')
except:
pass
try:
session_context.audio.kill()
print('Master Session closed! Old PyAudiohandler thread killed!')
except:
if num_sessions == 0:
print('Master Session closed!')
else:
print('Viewer Session closed!')
print(num_sessions, 'active session(s) opened!')
pass