# -*- coding: utf-8 -*-
#
# Author: Krishna Sunkara (kpsunkara)
#
# Re-entrant, reusable context manager to store execution context. Introduced
# in pmdarima 1.5.0 (see #221), redesigned not to use thread locals in #273
# (see #275 for context).
from abc import ABC, abstractmethod
from enum import Enum
import collections
__all__ = ['AbstractContext', 'ContextStore', 'ContextType']
class _CtxSingleton:
"""Singleton class to store context information"""
store = {}
_ctx = _CtxSingleton()
class ContextType(Enum):
"""Context Type Enumeration
An enumeration of Context Types known to :class:`ContextStore`
"""
EMPTY = 0
STEPWISE = 1
class AbstractContext(ABC):
"""An abstract context manager to store execution context.
A generic, re-entrant, reusable context manager to store
execution context. Has helper methods to iterate over the context info
and provide a string representation of the context info.
"""
def __init__(self, **kwargs):
# remove None valued entries,
# since __getattr__ returns None if an attr is not present
self.props = {k: v for k, v in kwargs.items() if v is not None} \
if kwargs else {}
def __enter__(self):
ContextStore._add_context(self)
def __exit__(self, exc_type, exc_val, exc_tb):
ContextStore._remove_context(self)
def __getattr__(self, item):
"""Lets us access, e.g., ``ctx.max_steps`` even if not in a context"""
return self.props[item] if item in self.props else None
def __contains__(self, item):
return item in self.props
def __getitem__(self, item):
return self.props[item] if item in self.props else None
def __iter__(self):
return iter(self.props)
def keys(self):
return self.props.keys()
def values(self):
return self.props.values()
def items(self):
return self.props.items()
def update(self, other):
parent_props = dict(other)
parent_props.update(self.props)
self.props = parent_props
def __repr__(self):
return self.props.__repr__()
@abstractmethod
def get_type(self):
"""Get the ContextType"""
class _emptyContext(AbstractContext):
"""An empty context for convenience use"""
def __init__(self):
super(_emptyContext, self).__init__()
def get_type(self):
"""Indicates we are not in a context manager"""
return ContextType.EMPTY
class ContextStore:
"""A class to wrap access to the global context store
This class hosts static methods to wrap access to and encapsulate the
singleton content store instance
"""
@staticmethod
def get_context(context_type):
"""Returns most recently added instance of given Context Type
Parameters
----------
context_type : ContextType
Context type to retrieve from the store
Returns
-------
res : AbstractContext
An instance of AbstractContext subclass or None
"""
if not isinstance(context_type, ContextType):
raise ValueError('context_type must be an instance of ContextType')
if context_type in _ctx.store and len(_ctx.store[context_type]) > 0:
return _ctx.store[context_type][-1]
# If not present
return None
@staticmethod
def get_or_default(context_type, default):
"""Returns most recent instance of given Context Type or default
Parameters
----------
context_type : ContextType
Context type to retrieve from the store
default : AbstractContext
Value to return in case given context does not exist
Returns
-------
ctx : AbstractContext
An instance of AbstractContext subclass or default
"""
ctx = ContextStore.get_context(context_type)
return ctx if ctx else default
@staticmethod
def get_or_empty(context_type):
"""Returns recent instance of given Context Type or an empty context
Parameters
----------
context_type : ContextType
Context type to retrieve from the store
Returns
-------
res : AbstractContext
An instance of AbstractContext subclass
"""
return ContextStore.get_or_default(context_type, _emptyContext())
@staticmethod
def _add_context(ctx):
"""Add given instance of AbstractContext subclass to context store
This private member is only called by ``AbstractContext.__init__()``
if the given ctx is nested, merge parent context, to support
following usage:
Examples
--------
>>> from pmdarima.arima import StepwiseContext, auto_arima
>>> with StepwiseContext(max_steps=10):
... with StepwiseContext(max_dur=30):
... auto_arima(samp,...)
This is identical to:
>>> from contextlib import ExitStack
... stack = ExitStack()
... outer_ctx = StepwiseContext(max_steps=10)
... inner_ctx = StepwiseContext(max_dur=30)
... stack.enter_context(outer_ctx)
... stack.enter_context(inner_ctx)
... with stack:
... auto_arima(samp, ...)
However, the nested context can override parent context. In the
example below, the effective context for inner most call to
``auto_arima(...)`` is: ``max_steps=15, max_dur=30``. The effective
context for the second call to ``auto_arima(..)`` is: ``max_steps=10``
>>> with StepwiseContext(max_steps=10):
... with StepwiseContext(max_steps=15, max_dur=30):
... auto_arima(samp,...)
...
... auto_arima(samp,...)
"""
if not isinstance(ctx, AbstractContext):
raise ValueError('ctx must be be an instance of AbstractContext')
# if given Context Type is not present into store, make an entry
context_type = ctx.get_type()
if context_type not in _ctx.store:
_ctx.store[context_type] = collections.deque()
# if the context is nested, merge with parent's context
if len(_ctx.store[context_type]) > 0:
parent = _ctx.store[context_type][-1]
ctx.update(parent)
_ctx.store[context_type].append(ctx)
@staticmethod
def _remove_context(ctx):
"""Removes the most recently added context of given Context Type
This private member is only used by ``AbstractContext``
:param ctx:
:return: None
"""
if not isinstance(ctx, AbstractContext):
raise ValueError('ctx must be be an instance of AbstractContext')
context_type = ctx.get_type()
if context_type not in _ctx.store or \
len(_ctx.store[context_type]) == 0:
return
_ctx.store[context_type].pop()