DriverInstanceCache.py

import logging, threading
from typing import Any, Dict, Optional

from device.service.driver_api.FilterFields import FilterFieldEnum
from ._Driver import _Driver
from .DriverFactory import DriverFactory
from .Exceptions import DriverInstanceCacheTerminatedException

LOGGER = logging.getLogger(__name__)

class DriverInstanceCache:
    def __init__(self, driver_factory : DriverFactory) -> None:
        self._lock = threading.Lock()
        self._terminate = threading.Event()
        self._device_uuid__to__driver_instance : Dict[str, _Driver] = {}
        self._driver_factory = driver_factory

    def get(
        self, device_uuid : str, filter_fields : Dict[FilterFieldEnum, Any] = {}, address : Optional[str] = None,
        port : Optional[int] = None, settings : Dict[str, Any] = {}) -> _Driver:

        if self._terminate.is_set():
            raise DriverInstanceCacheTerminatedException()

        filter_fields = {k.value:v for k,v in filter_fields.items()}

        with self._lock:
            driver_instance = self._device_uuid__to__driver_instance.get(device_uuid)
            if driver_instance is not None: return driver_instance

            if len(filter_fields) == 0: return None
            driver_class = self._driver_factory.get_driver_class(**filter_fields)
            driver_instance : _Driver = driver_class(address, port, **settings)
            self._device_uuid__to__driver_instance[device_uuid] = driver_instance
            return driver_instance

    def delete(self, device_uuid : str) -> None:
        with self._lock:
            device_driver = self._device_uuid__to__driver_instance.pop(device_uuid, None)
            if device_driver is None: return
            device_driver.Disconnect()

    def terminate(self) -> None:
        self._terminate.set()
        with self._lock:
            while len(self._device_uuid__to__driver_instance) > 0:
                try:
                    device_uuid,device_driver = self._device_uuid__to__driver_instance.popitem()
                    device_driver.Disconnect()
                except: # pylint: disable=bare-except
                    msg = 'Error disconnecting Driver({:s}) from device. Will retry later...'
                    LOGGER.exception(msg.format(device_uuid))
                    self._device_uuid__to__driver_instance[device_uuid] = device_driver