#!/usr/bin/env python
##############################################################################
##
# This file is part of Sardana
##
# http://www.sardana-controls.org/
##
# Copyright 2011 CELLS / ALBA Synchrotron, Bellaterra, Spain
##
# Sardana is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
##
# Sardana is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
##
# You should have received a copy of the GNU Lesser General Public License
# along with Sardana. If not, see <http://www.gnu.org/licenses/>.
##
##############################################################################
"""
Environment manager supporting Shelve, Redis & TangoDB backends.
This module provides the EnvironmentManager class for use with Sardana's
MacroServer. It abstracts persistence of environment variables across three
backends: Python shelve, Redis, and TangoDB (via PyTango).
"""
__all__ = ["EnvironmentManager"]
__docformat__ = "restructuredtext"
import os
from typing import Iterable
from sardana.macroserver.environment.common import (
dict_from_pairs_or_dict,
iter_env_scope,
parse_fq_key,
parse_from_str,
)
from sardana.macroserver.environment.redis import RedisEnvironmentBackend
from sardana.macroserver.environment.shelve import ShelveEnvironmentBackend
from sardana.macroserver.environment.tangodb import TangoDBEnvironmentBackend
from sardana.macroserver.msexception import UnknownEnv
from sardana.macroserver.msmanager import MacroServerManager
[docs]
class EnvironmentManager(MacroServerManager):
"""
Unified environment manager supporting Shelve, Redis & TangoDB backends.
"""
def __init__(self, macro_server, environment_db=None):
super(EnvironmentManager, self).__init__(macro_server)
self._backend = None
self._env_uri = None
self._macro_server = macro_server
if environment_db is not None:
self.setEnvironmentDb(environment_db)
def reInit(self):
"""Reinitialize caches (backend stays open)."""
super(EnvironmentManager, self).reInit()
def cleanUp(self):
"""Close backend and then parent cleanup."""
if self._backend:
self._backend.close_backend()
super(EnvironmentManager, self).cleanUp()
def setEnvironmentDb(self, environment_db):
"""
Configure the storage backend based on URI or path.
"""
self._env_uri = environment_db
# TangoDB
if environment_db == "property":
uri = self._macro_server.get_name()
self._backend = TangoDBEnvironmentBackend()
self._backend.init_backend(uri)
self.info("Environment is being stored in %s", uri)
elif environment_db.startswith("property:"):
uri = environment_db.split("property:", 1)[1]
self._backend = TangoDBEnvironmentBackend()
self._backend.init_backend(uri)
self.info("Environment is being stored in %s", environment_db)
# Redis
elif environment_db.startswith("redis://") or environment_db.startswith(
"rediss://"
):
prefix = self._macro_server.get_name()
self._backend = RedisEnvironmentBackend()
self._backend.init_backend(environment_db, prefix)
self.info("Environment is being stored in %s", environment_db)
# Shelve (default)
else:
path = os.path.abspath(environment_db)
parent = os.path.dirname(path)
if parent and not os.path.isdir(parent):
self.info("Creating environment directory: %s", parent)
os.makedirs(parent)
self._backend = ShelveEnvironmentBackend()
self._backend.init_backend(path)
self.info("Environment is being stored in %s", path)
def hasEnv(self, key, macro_name=None, door_name=None):
# cascade: door+macro → macro → door → global
for prop, dn, mn in iter_env_scope(key, door_name, macro_name):
if self._backend.has_env(prop, dn, mn):
return True
return False
def setEnv(self, key, value):
dn, mn, prop = parse_fq_key(key)
return self._backend.set_env(prop, parse_from_str(value), dn, mn)
def setEnvObj(self, obj):
obj = dict_from_pairs_or_dict(obj)
result = {}
for k, v in obj.items():
dn, mn, prop = parse_fq_key(k)
nk, nv = self._backend.set_env(prop, parse_from_str(v), dn, mn)
result[nk] = nv
return result
def getEnv(self, key=None, door_name=None, macro_name=None):
if key is None:
flat = self._backend.list_all()
if door_name and macro_name:
return self.getAllDoorMacroEnv(door_name, macro_name, flat=flat)
if door_name:
return self.getAllDoorEnv(door_name, flat=flat)
return flat # global + everything, unchanged
for prop, dn, mn in iter_env_scope(key, door_name, macro_name):
if self._backend.has_env(prop, dn, mn):
return self._backend.get_env(prop, dn, mn)
raise UnknownEnv("Unknown environment variable %s" % key)
def unsetEnv(self, key):
if isinstance(key, str):
keys = (key,)
elif isinstance(key, Iterable):
keys = tuple(key)
else:
raise TypeError("unsetEnv expects a string or an iterable of strings")
removed = []
for k in keys:
dn, mn, prop = parse_fq_key(k)
self._backend.unset_env(prop, dn, mn)
removed.append(k)
return tuple(removed)
def getAllDoorEnv(self, door_name, flat=None):
"""
global + door-only scope
"""
if flat is None:
flat = self._backend.list_all()
result = {}
for full, val in flat.items():
if "." not in full:
result[full] = val
prefix = door_name.lower() + "."
for full, val in flat.items():
if full.lower().startswith(prefix) and full.count(".") == 1:
name = full.split(".", 1)[1]
result[name] = val
return result
def getAllDoorMacroEnv(self, door_name, macro_name, flat=None):
"""
global + door-only + macro-only + door+macro
"""
if flat is None:
flat = self._backend.list_all()
result = self.getAllDoorEnv(door_name, flat=flat)
if isinstance(macro_name, str):
macro_names = [macro_name]
elif isinstance(macro_name, Iterable):
macro_names = list(macro_name)
else:
macro_names = []
for macro in reversed(macro_names):
macro_prefix = macro.lower() + "."
for full, val in flat.items():
if full.lower().startswith(macro_prefix) and full.count(".") == 1:
result[full.split(".", 1)[1]] = val
door_macro_prefix = door_name.lower() + "." + macro.lower() + "."
for full, val in flat.items():
if full.lower().startswith(door_macro_prefix):
result[full[len(door_macro_prefix) :]] = val
return result
def getDoorMacroEnv(self, door_name, macro_name, keys=None):
scope = self.getAllDoorMacroEnv(door_name, macro_name)
if keys is None:
return scope
if isinstance(keys, str):
keys = (keys,)
filtered = {}
for k in keys:
if k in scope:
filtered[k] = scope[k]
return filtered