src.fairreckitlib.core.config.config_factories

  1"""This module contains the (base) factories that are used in other packages.
  2
  3Classes:
  4
  5    BaseFactory: base class for all factories.
  6    Factory: class that instantiates new objects (a leaf).
  7    GroupFactory: class that groups other factories (a branch).
  8
  9Functions:
 10
 11    create_factory_from_list: create object factory from list of object (param) creation tuples.
 12    resolve_factory: resolve the object factory from the name of the object.
 13
 14This program has been developed by students from the bachelor Computer Science at
 15Utrecht University within the Software Project course.
 16© Copyright Utrecht University (Department of Information and Computing Sciences)
 17"""
 18
 19from abc import ABCMeta, abstractmethod
 20from typing import Any, Callable, Dict, List, Optional, Tuple, Union
 21
 22from ..core_constants import KEY_NAME, KEY_PARAMS
 23from .config_parameters import ConfigParameters, create_empty_parameters
 24
 25FUNC_CREATE_OBJ = 'f_create_obj'
 26FUNC_CREATE_PARAMS = 'f_create_params'
 27
 28
 29class BaseFactory(metaclass=ABCMeta):
 30    """Base class for all FairRecKit experiment factories.
 31
 32    The abstraction is intended to support functionality for a tree-like structure.
 33    The actual factory is a leaf and a group of factories that belong together a branch.
 34
 35    Public methods:
 36
 37    get_available
 38    get_available_names
 39    get_name
 40    is_available
 41    """
 42
 43    def __init__(self, factory_name: str):
 44        """Construct the base factory.
 45
 46        Args:
 47            factory_name: the name of the factory.
 48        """
 49        self.factory_name = factory_name
 50        self.factory = {}
 51
 52    @abstractmethod
 53    def get_available(self) -> Any:
 54        """Get the available objects in the factory.
 55
 56        The availability return type depends on the derived class.
 57        """
 58        raise NotImplementedError()
 59
 60    def get_available_names(self) -> List[str]:
 61        """Get the names that are available in the factory.
 62
 63        Returns:
 64            a list of names that is available.
 65        """
 66        result = []
 67
 68        for name, _ in self.factory.items():
 69            result.append(name)
 70
 71        return result
 72
 73    def get_name(self) -> str:
 74        """Get the name of the factory.
 75
 76        Returns:
 77            the factory name.
 78        """
 79        return self.factory_name
 80
 81    def get_num_entries(self) -> int:
 82        """Get the number of entries present in the factory.
 83
 84        Returns:
 85            the amount of entries.
 86        """
 87        return len(self.factory)
 88
 89    @abstractmethod
 90    def is_obj_available(self, obj_name: str) -> bool:
 91        """Is the object with the specified name available.
 92
 93        Args:
 94            obj_name: the name of the object to query for availability.
 95
 96        Returns:
 97            whether the object is available.
 98        """
 99        raise NotImplementedError()
100
101
102class Factory(BaseFactory):
103    """The factory that implements object and parameters creation.
104
105    Public methods:
106
107    add_obj
108    create
109    create_params
110    """
111
112    def add_obj(self,
113                obj_name: str,
114                func_create_obj: Callable[[str, Dict[str, Any]], Any],
115                func_create_obj_params: Optional[Callable[[], ConfigParameters]]=None
116                ) -> None:
117        """Add object with associated parameter creation to the factory.
118
119        Parameter creation is optional and should be None if the object has no parameters.
120
121        Args:
122            obj_name: the name of the object
123            func_create_obj: the function that creates and returns a new object.
124            func_create_obj_params: the function that creates and returns the parameters
125                that are associated with a newly created object or None when the object
126                has no parameters.
127
128        Raises:
129            KeyError: when the object name is already present in the factory.
130        """
131        if obj_name in self.factory:
132            raise KeyError('Factory ' + self.factory_name,
133                           ': object already exists: ' + obj_name)
134
135        if func_create_obj_params is None:
136            func_create_obj_params = create_empty_parameters
137
138        self.factory[obj_name] = {
139            FUNC_CREATE_OBJ: func_create_obj,
140            FUNC_CREATE_PARAMS: func_create_obj_params
141        }
142
143    def create(self, obj_name: str, obj_params: Dict[str, Any]=None, **kwargs) -> Any:
144        """Create and return a new object with the specified name.
145
146        The specified parameters are expected to be of the same structure as the defaults
147        of the ConfigParameters that are associated with the desired object. When no parameters
148        are specified it will use the object's defaults.
149
150        Args:
151            obj_name: the name of the object to create.
152            obj_params: the parameters of the object.
153
154        Keyword Args:
155            Any: extra arguments that need to be passed to the object on creation.
156
157        Returns:
158            the created object or None when it does not exist.
159        """
160        if obj_name not in self.factory:
161            return None
162
163        func_create_obj = self.factory[obj_name][FUNC_CREATE_OBJ]
164        if obj_params is None:
165            obj_params = self.create_params(obj_name).get_defaults()
166
167        return func_create_obj(obj_name, dict(obj_params), **kwargs)
168
169    def create_params(self, obj_name: str) -> ConfigParameters:
170        """Create parameters for the object with the specified name.
171
172        Args:
173            obj_name: name of the object to create parameters for.
174
175        Returns:
176            the configuration parameters of the object or empty parameters when it does not exist.
177        """
178        if obj_name not in self.factory:
179            return create_empty_parameters()
180
181        return self.on_create_params(obj_name)
182
183    def on_create_params(self, obj_name: str) -> ConfigParameters:
184        """Create parameters for the object with the specified name.
185
186        Args:
187            obj_name: name of the object to create parameters for.
188
189        Returns:
190            the configuration parameters of the object or empty parameters when it does not exist.
191        """
192        return self.factory[obj_name][FUNC_CREATE_PARAMS]()
193
194    def get_available(self) -> List[Dict[str, Dict[str, Any]]]:
195        """Get the availability of all object names and their parameters.
196
197        Each object in the factory has a name and parameters that consists of
198        a dictionary with name-value pairs.
199
200        Returns:
201            a list of dictionary entries that includes the name and the parameters.
202        """
203        obj_list = []
204
205        for obj_name, _ in self.factory.items():
206            obj_list.append({
207                KEY_NAME: obj_name,
208                KEY_PARAMS: self.on_create_params(obj_name).to_dict()
209            })
210
211        return obj_list
212
213    def is_obj_available(self, obj_name: str) -> bool:
214        """Is the object with the specified name available.
215
216        Checks the object for existing in this factory.
217
218        Args:
219            obj_name: the name of the object to query for availability.
220
221        Returns:
222            whether the object is available.
223        """
224        return obj_name in self.factory is not None
225
226
227class GroupFactory(BaseFactory):
228    """The factory that groups other factories together.
229
230    Public methods:
231
232    add_factory
233    get_factory
234    get_sub_availability
235    """
236
237    def add_factory(self, factory: BaseFactory) -> None:
238        """Add the specified factory to the group.
239
240        The name of the factory is used as the key.
241
242        Args:
243            factory: to add to the group.
244
245        Raises:
246            KeyError: when the name of the factory already exists in the group.
247        """
248        factory_name = factory.get_name()
249        if factory_name in self.factory:
250            raise KeyError('Factory ' + self.factory_name,
251                           ': factory already exists: ' + factory_name)
252
253        self.factory[factory_name] = factory
254
255    def get_available(self) -> Dict[str, Any]:
256        """Get the availability of all factories in the group.
257
258        Each factory has a name and availability that depends on the
259        type of the factory. Effectively this will generate a tree-like
260        structure of the factory's availability.
261
262        Returns:
263            a dictionary with factory name and availability pairs.
264        """
265        factory_list = {}
266
267        for factory_name, factory in self.factory.items():
268            factory_list[factory_name] = factory.get_available()
269
270        return factory_list
271
272    def get_factory(self, factory_name: str) -> BaseFactory:
273        """Get the factory with the specified name.
274
275        Args:
276            factory_name: the name of the factory to retrieve
277
278        Returns:
279            the requested factory or None when not available.
280        """
281        return self.factory.get(factory_name)
282
283    def get_sub_availability(
284            self,
285            sub_factory_name: str,
286            sub_type: str = None) -> Dict[str, Any]:
287        """Get the sub availability from the factory with the specified name (and type).
288
289        Args:
290            sub_factory_name: the name of the sub-factory to query for availability.
291            sub_type: the subtype of the sub-factory to query for availability or None
292                for the complete availability.
293
294        Returns:
295            a dictionary containing the availability of the sub-factory (type).
296        """
297        sub_factory = self.get_factory(sub_factory_name)
298        if sub_type is None:
299            return sub_factory.get_available()
300
301        type_factory = sub_factory.get_factory(sub_type)
302        if type_factory is None:
303            return {}
304
305        return type_factory.get_available()
306
307    def is_obj_available(self, obj_name: str) -> bool:
308        """Is the object with the specified name available.
309
310        Checks the object for existing in any of the child factories.
311
312        Args:
313            obj_name: the name of the object to query for availability.
314
315        Returns:
316            whether the object is available.
317        """
318        for _, factory in self.factory.items():
319            if factory.is_obj_available(obj_name):
320                return True
321
322        return False
323
324
325def create_factory_from_list(
326        factory_name: str,
327        obj_tuple_list: List[Tuple[
328            str,
329            Callable[[str, Dict[str, Any]], Any],
330            Optional[Callable[[], ConfigParameters]]
331        ]]) -> Factory:
332    """Create and return the factory with the specified tuple entries.
333
334    Each tuple in the list consists of three things; the name of the object,
335    the object creation function and the parameter creation function.
336
337    Args:
338        factory_name: the name of the factory to create.
339        obj_tuple_list: a list of object tuples to add after factory creation.
340
341    Returns:
342        the factory with the added objects.
343    """
344    factory = Factory(factory_name)
345
346    for (obj_name, func_create_obj, func_create_obj_params) in obj_tuple_list:
347        factory.add_obj(
348            obj_name,
349            func_create_obj,
350            func_create_obj_params
351        )
352
353    return factory
354
355
356def resolve_factory(
357        obj_name: str,
358        obj_factory: Union[Factory, GroupFactory]) -> Optional[Factory]:
359    """Resolve the object factory from the name of the object.
360
361    Args:
362        obj_name: the name of the object.
363        obj_factory: the (group) factory to query for the object.
364
365    Returns:
366        the resolved factory or None when not found.
367    """
368    if isinstance(obj_factory, Factory):
369        return obj_factory if obj_factory.is_obj_available(obj_name) else None
370
371    for factory_name in obj_factory.get_available_names():
372        obj_sub_factory = obj_factory.get_factory(factory_name)
373        if obj_sub_factory.is_obj_available(obj_name):
374            if isinstance(obj_sub_factory, GroupFactory):
375                return resolve_factory(obj_name, obj_sub_factory)
376
377            return obj_sub_factory
378
379    return None