class DeviceActions(Enum):
    UNKNOWN = 0,
    CONNECT = 1,
    DISCONNECT = 2 class Status(Enum):
    FAILURE = 1,
    NOT_FOUND = 2,
    SUCCESS = 3 CONNECTED_DEVICES_FILENAME = "connected_devices.json" CONNECTOR_PARAMETER = "connector" PERSISTENT_GRPC_CONNECTORS_KEY_FILENAME = "persistent_keys.json" class TBClient(threading.Thread):
    def __init__(self, config, config_folder_path):
        super().__init__()
        self.setName('Connection thread.')
        self.daemon = True
        self.__config_folder_path = config_folder_path
        self.__config = config
        self.__host = config["host"]
        self.__port = config.get("port", 1883)
        self.__default_quality_of_service = config.get("qos", 1)
        credentials = config["security"]
        self.__min_reconnect_delay = 1
        self.__tls = bool(credentials.get('tls', False) or credentials.get('caCert', False))
        self.__ca_cert = None
        self.__private_key = None
        self.__cert = None
        self.__token = None
        self.__is_connected = False
        self.__stopped = False
        self.__paused = False
        if credentials.get("accessToken") is not None:
            self.__token = str(credentials["accessToken"])
        self.client = TBGatewayMqttClient(self.__host, self.__port, self.__token, self, quality_of_service=self.__default_quality_of_service)
        if self.__tls:
            self.__ca_cert = self.__config_folder_path + credentials.get("caCert") if credentials.get("caCert") is not None else None
            self.__private_key = self.__config_folder_path + credentials.get("privateKey") if credentials.get("privateKey") is not None else None
            self.__cert = self.__config_folder_path + credentials.get("cert") if credentials.get("cert") is not None else None
            self.client._client.tls_set(ca_certs=self.__ca_cert,
                                        certfile=self.__cert,
                                        keyfile=self.__private_key,
                                        tls_version=PROTOCOL_TLSv1_2,
                                        cert_reqs=CERT_REQUIRED,
                                        ciphers=None)
            if credentials.get("insecure", False):
                self.client._client.tls_insecure_set(True)
        # pylint: disable=protected-access
        # Adding callbacks
        self.client._client._on_connect = self._on_connect
        self.client._client._on_disconnect = self._on_disconnect
        # self.client._client._on_log = self._on_log
        self.start()

    # def _on_log(self, *args):
    #     if "exception" in args[-1]:
    #         log.exception(args)
    #     else:
    #         log.debug(args)

    def pause(self):
        self.__paused = True

    def unpause(self):
        self.__paused = False

    def is_connected(self):
        return self.__is_connected

    def _on_connect(self, client, userdata, flags, result_code, *extra_params):
        log.debug('TB client %s connected to ThingsBoard', str(client))
        if result_code == 0:
            self.__is_connected = True
        # pylint: disable=protected-access
        self.client._on_connect(client, userdata, flags, result_code, *extra_params)

    def _on_disconnect(self, client, userdata, result_code):
        # pylint: disable=protected-access
        if self.client._client != client:
            log.info("TB client %s has been disconnected. Current client for connection is: %s", str(client), str(self.client._client))
            client.disconnect()
            client.loop_stop()
        else:
            self.__is_connected = False
            self.client._on_disconnect(client, userdata, result_code)

    def stop(self):
        # self.disconnect()
        self.client.stop()
        self.__stopped = True

    def disconnect(self):
        self.__paused = True
        self.unsubscribe('*')
        self.client.disconnect()

    def unsubscribe(self, subsription_id):
        self.client.gw_unsubscribe(subsription_id)
        self.client.unsubscribe_from_attribute(subsription_id)

    def connect(self, min_reconnect_delay=10):
        self.__paused = False
        self.__stopped = False
        self.__min_reconnect_delay = min_reconnect_delay

    def run(self):
        keep_alive = self.__config.get("keep_alive", 120)
        try:
            while not self.client.is_connected() and not self.__stopped:
                if not self.__paused:
                    if self.__stopped:
                        break
                    log.debug("connecting to ThingsBoard")
                    try:
                        self.client.connect(keepalive=keep_alive,
                                            min_reconnect_delay=self.__min_reconnect_delay)
                    except ConnectionRefusedError:
                        pass
                    except Exception as e:
                        log.exception(e)
                sleep(1)
        except Exception as e:
            log.exception(e)
            sleep(10)

        while not self.__stopped:
            try:
                if not self.__stopped:
                    sleep(.2)
                else:
                    break
            except KeyboardInterrupt:
                self.__stopped = True
            except Exception as e:
                log.exception(e)

    def get_config_folder_path(self):
        return self.__config_folder_path class FileEventStorage(EventStorage):
    def __init__(self, config):
        self.settings = FileEventStorageSettings(config)
        self.init_data_folder_if_not_exist()
        self.event_storage_files = self.init_data_files()
        self.data_files = self.event_storage_files.get_data_files()
        self.state_file = self.event_storage_files.get_state_file()
        self.__writer = EventStorageWriter(self.event_storage_files, self.settings)
        self.__reader = EventStorageReader(self.event_storage_files, self.settings)
        self.__stopped = False

    def put(self, event):
        success = False
        if not self.__stopped:
            try:
                self.__writer.write(event)
            except DataFileCountError as e:
                log.error(e)
            except Exception as e:
                log.exception(e)
            else:
                success = True
        else:
            log.error("Storage is closed!")
        return success

    def get_event_pack(self):
        return self.__reader.read()

    def event_pack_processing_done(self):
        self.__reader.discard_batch()

    def init_data_folder_if_not_exist(self):
        path = self.settings.get_data_folder_path()
        if not os.path.exists(path):
            try:
                os.makedirs(path)
            except OSError as e:
                log.error('Failed to create data folder! Error: %s', e)

    def init_data_files(self):
        data_files = []
        state_file = None
        data_files_size = 0
        _dir = self.settings.get_data_folder_path()
        event_storage_files = None
        if os.path.isdir(_dir):
            for file in os.listdir(_dir):
                if file.startswith('data_'):
                    data_files.append(file)
                    data_files_size += os.path.getsize(_dir + file)
                elif file.startswith('state_'):
                    state_file = file
            if data_files_size == 0:
                data_files.append(self.create_new_datafile())
            if not state_file:
                state_file = self.create_file('state_', 'file')
                with open(self.settings.get_data_folder_path() + state_file, 'w') as state_file_obj:
                    dump({"position": 0, "file": sorted(data_files)[0]}, state_file_obj)
            event_storage_files = EventStorageFiles(state_file, data_files)
        return event_storage_files

    def create_new_datafile(self):
        return self.create_file('data_', str(round(time.time() * 1000)))

    def create_file(self, prefix, filename):
        file_path = self.settings.get_data_folder_path() + prefix + filename + '.txt'
        try:
            file = open(file_path, 'w')
            file.close()
            return prefix + filename + '.txt'
        except IOError as e:
            log.error("Failed to create a new file! Error: %s", e)

    def stop(self):
        self.__stopped = True class MemoryEventStorage(EventStorage):
    def __init__(self, config):
        self.__queue_len = config.get("max_records_count", 10000)
        self.__events_per_time = config.get("read_records_count", 1000)
        self.__events_queue = Queue(self.__queue_len)
        self.__event_pack = []
        self.__stopped = False
        log.debug("Memory storage created with following configuration: \nMax size: %i\n Read records per time: %i",
                  self.__queue_len, self.__events_per_time)

    def put(self, event):
        success = False
        if not self.__stopped:
            try:
                self.__events_queue.put(event)
                success = True
            except Full:
                log.error("Memory storage is full!")
        else:
            log.error("Storage is stopped!")
        return success

    def get_event_pack(self):
        try:
            if not self.__event_pack:
                self.__event_pack = [self.__events_queue.get(False) for _ in
                                     range(min(self.__events_per_time, self.__events_queue.qsize()))]
        except Empty:
            pass
        return self.__event_pack

    def event_pack_processing_done(self):
        self.__event_pack = []

    def stop(self):
        self.__stopped = True class SQLiteEventStorage(EventStorage):
    """
    HIGH level api for thingsboard_gateway main loop
    """

    def __init__(self, config):
        log.info("Sqlite Storage initializing...")
        log.info("Initializing read and process queues")
        self.processQueue = Queue(-1)
        self.db = Database(config, self.processQueue)
        self.db.setProcessQueue(self.processQueue)
        self.db.init_table()
        log.info("Sqlite storage initialized!")
        self.delete_time_point = None
        self.last_read = time()
        self.stopped = False

    def get_event_pack(self):
        if not self.stopped:
            self.delete_time_point = self.last_read
            data_from_storage = self.read_data(self.last_read)
            self.last_read = time()

            return [item[0] for item in data_from_storage or []]
        else:
            return []

    def event_pack_processing_done(self):
        if not self.stopped:
            self.delete_data(self.delete_time_point)

    def read_data(self, ts):
        self.db.__stopped = True
        data = self.db.read_data(ts)
        self.db.__stopped = False
        return data

    def delete_data(self, ts):
        return self.db.delete_data(ts)

    def put(self, message):
        try:
            if not self.stopped:
                _type = DatabaseActionType.WRITE_DATA_STORAGE
                request = DatabaseRequest(_type, message)

                log.info("Sending data to storage")
                self.processQueue.put(request)
                self.db.process()
                return True
            else:
                return False
        except Exception as e:
            log.exception(e)

    def stop(self):
        self.stopped = True
        self.db.__stopped = True
        self.db.closeDB() class RemoteConfigurator:
    def __init__(self, gateway, config):
        self.__gateway = gateway
        self.__new_configuration = None
        self.__old_configuration = None
        self.__apply_timeout = 10
        self.__old_tb_client = None
        self.__old_logs_configuration = self.__get_current_logs_configuration()
        self.__new_logs_configuration = None
        self.__old_connectors_configs = {}
        self.__new_connectors_configs = {}
        self.__old_general_configuration_file = config
        self.__new_general_configuration_file = {}
        self.__old_event_storage = None
        self.__new_event_storage = None
        self.in_process = False

    def process_configuration(self, configuration):
        try:
            if not self.in_process:
                self.in_process = True
                # while not self.__gateway._published_events.empty():
                #     LOG.debug("Waiting for end of the data processing...")
                #     sleep(1)
                decoded_configuration = b64decode(configuration)
                self.__new_configuration = loads(decoded_configuration)
                self.__old_connectors_configs = self.__gateway.connectors_configs
                self.__new_general_configuration_file = self.__new_configuration.get("thingsboard")
                
                # To maintain RemoteShell status
                if not isinstance(self.__old_configuration, dict):
                    self.__old_configuration = loads(b64decode(self.__old_configuration))
                if self.__old_configuration.get("thingsboard").get("thingsboard").get("remoteShell"):
                    self.__new_configuration["thingsboard"]["thingsboard"]["remoteShell"] = True
                    
                self.__new_logs_configuration = b64decode(self.__new_general_configuration_file.pop("logs")).decode('UTF-8').replace('}}', '\n')
                if self.__old_configuration != decoded_configuration:
                    LOG.info("Remote configuration received: \n %s", decoded_configuration)
                    result = self.__process_connectors_configuration()
                    self.in_process = False
                    if result:
                        self.__old_configuration = self.__new_configuration
                        return True
                    else:
                        return False
                else:
                    LOG.info("Remote configuration is the same.")
            else:
                LOG.error("Remote configuration is already in processing")
                return False
        except Exception as e:
            self.in_process = False
            LOG.exception(e)

    def send_current_configuration(self):
        try:
            current_configuration = {}
            for connector in self.__gateway.connectors_configs:
                if current_configuration.get(connector) is None:
                    current_configuration[connector] = []
                for config in self.__gateway.connectors_configs[connector]:
                    for config_file in config['config']:
                        current_configuration[connector].append({'name': config['name'], 'config': config['config'][config_file]})
            current_configuration["thingsboard"] = self.__old_general_configuration_file
            current_configuration["thingsboard"]["logs"] = b64encode(self.__old_logs_configuration.replace('\n', '}}').encode("UTF-8"))
            json_current_configuration = dumps(current_configuration)
            encoded_current_configuration = b64encode(json_current_configuration.encode())
            self.__old_configuration = encoded_current_configuration
            self.__gateway.tb_client.client.send_attributes(
                {"current_configuration": encoded_current_configuration.decode("UTF-8")})
            LOG.debug('Current configuration has been sent to ThingsBoard: %s', json_current_configuration)
        except Exception as e:
            LOG.exception(e)

    def __process_connectors_configuration(self):
        LOG.info("Processing remote connectors configuration...")
        if self.__apply_new_connectors_configuration():
            self.__write_new_configuration_files()
        self.__apply_storage_configuration()
        if self.__safe_apply_connection_configuration():
            LOG.info("Remote configuration has been applied.")
            with open(self.__gateway.get_config_path() + "tb_gateway.yaml", "w", encoding="UTF-8") as general_configuration_file:
                safe_dump(self.__new_general_configuration_file, general_configuration_file)
            self.__old_connectors_configs = {}
            self.__new_connectors_configs = {}
            self.__old_general_configuration_file = self.__new_general_configuration_file
            self.__old_logs_configuration = self.__new_logs_configuration
            self.__update_logs_configuration()
            self.__new_logs_configuration = None
            self.__new_general_configuration_file = {}
            return True
        else:
            self.__update_logs_configuration()
            self.__old_general_configuration_file.pop("logs")
            with open(self.__gateway.get_config_path() + "tb_gateway.yaml", "w", encoding="UTF-8") as general_configuration_file:
                safe_dump(self.__old_general_configuration_file, general_configuration_file)
            LOG.error("A remote general configuration applying has been failed.")
            self.__old_connectors_configs = {}
            self.__new_connectors_configs = {}
            self.__new_logs_configuration = None
            self.__new_general_configuration_file = {}
            return False

    def __prepare_connectors_configuration(self, input_connector_config):
        try:
            self.__gateway.connectors_configs = {}
            for connector in input_connector_config['thingsboard']['connectors']:
                for input_connector in input_connector_config[connector['type']]:
                    if input_connector['name'] == connector['name']:
                        if not self.__gateway.connectors_configs.get(connector['type']):
                            self.__gateway.connectors_configs[connector['type']] = []
                        config_file_path = self.__gateway.get_config_path() + connector['configuration']
                        # Create the configuration json file if not exists
                        open(config_file_path, 'w')
                        self.__gateway.connectors_configs[connector['type']].append(
                            {"name": connector["name"], 
                            "config": {connector['configuration']: input_connector["config"]},
                            "config_updated": stat(config_file_path),
                            "config_file_path": config_file_path})
                        connector_class = TBModuleLoader.import_module(connector["type"],
                                                                       self.__gateway._default_connectors.get(connector["type"], connector.get("class")))
                        self.__gateway._implemented_connectors[connector["type"]] = connector_class
        except Exception as e:
            LOG.exception(e)

    def __apply_new_connectors_configuration(self):
        try:
            self.__prepare_connectors_configuration(self.__new_configuration)
            for connector_name in self.__gateway.available_connectors:
                try:
                    self.__gateway.available_connectors[connector_name].close()
                except Exception as e:
                    LOG.exception(e)
            self.__gateway._connect_with_connectors()
            LOG.debug("New connectors configuration has been applied")
            self.__old_connectors_configs = {}
            return True
        except Exception as e:
            self.__gateway.connectors_configs = self.__old_connectors_configs
            for connector_name in self.__gateway.available_connectors:
                self.__gateway.available_connectors[connector_name].close()
            self.__gateway._load_connectors(self.__old_general_configuration_file)
            self.__gateway._connect_with_connectors()
            LOG.exception(e)
            return False

    def __write_new_configuration_files(self):
        try:
            self.__new_connectors_configs = self.__new_connectors_configs if self.__new_connectors_configs else self.__gateway.connectors_configs
            new_connectors_files = []
            for connector_type in self.__new_connectors_configs:
                for connector_config_section in self.__new_connectors_configs[connector_type]:
                    for connector_file in connector_config_section["config"]:
                        connector_config = connector_config_section["config"][connector_file]
                        with open(self.__gateway.get_config_path() + connector_file, "w", encoding="UTF-8") as config_file:
                            dump(connector_config, config_file, sort_keys=True, indent=2)
                        new_connectors_files.append(connector_file)
                        LOG.debug("Saving new configuration for \"%s\" connector to file \"%s\"", connector_type,
                                  connector_file)
                        break
            self.__old_general_configuration_file["connectors"] = self.__new_general_configuration_file["connectors"]
            for old_connector_type in self.__old_connectors_configs:
                for old_connector_config_section in self.__old_connectors_configs[old_connector_type]:
                    for old_connector_file in old_connector_config_section["config"]:
                        if old_connector_file not in new_connectors_files:
                            remove(self.__gateway.get_config_path() + old_connector_file)
                        LOG.debug("Remove old configuration file \"%s\" for \"%s\" connector ", old_connector_file,
                                  old_connector_type)
        except Exception as e:
            LOG.exception(e)

    def __safe_apply_connection_configuration(self):
        apply_start = time() * 1000
        self.__old_tb_client = self.__gateway.tb_client
        try:
            self.__old_tb_client.unsubscribe('*')
            self.__old_tb_client.stop()
            self.__old_tb_client.disconnect()
            self.__gateway.tb_client = TBClient(self.__new_general_configuration_file["thingsboard"], self.__old_tb_client.get_config_folder_path())
            self.__gateway.tb_client.connect()
            connection_state = False
            while time() * 1000 - apply_start < self.__apply_timeout * 1000 and not connection_state:
                connection_state = self.__gateway.tb_client.is_connected()
                sleep(.1)
            if not connection_state:
                self.__revert_configuration()
                LOG.info("The gateway cannot connect to the ThingsBoard server with a new configuration.")
                return False
            else:
                self.__old_tb_client.stop()
                self.__gateway.subscribe_to_required_topics()
                return True
        except Exception as e:
            LOG.exception(e)
            self.__revert_configuration()
            return False

    def __apply_storage_configuration(self):
        if self.__old_general_configuration_file["storage"] != self.__new_general_configuration_file["storage"]:
            self.__old_event_storage = self.__gateway._event_storage
            try:
                storage_class = self.__gateway._event_storage_types[self.__new_general_configuration_file["storage"]["type"]]
                self.__gateway._event_storage = storage_class(self.__new_general_configuration_file["storage"])
                self.__old_event_storage = None
            except Exception as e:
                LOG.exception(e)
                self.__gateway._event_storage = self.__old_event_storage

    def __revert_configuration(self):
        try:
            LOG.info("Remote general configuration will be restored.")
            self.__new_general_configuration_file = self.__old_general_configuration_file
            self.__gateway.tb_client.disconnect()
            self.__gateway.tb_client.stop()
            self.__gateway.tb_client = TBClient(self.__old_general_configuration_file["thingsboard"])
            self.__gateway.tb_client.connect()
            self.__gateway.subscribe_to_required_topics()
            LOG.debug("%s connection has been restored", str(self.__gateway.tb_client.client._client))
        except Exception as e:
            LOG.exception("Exception on reverting configuration occurred:")
            LOG.exception(e)

    def __get_current_logs_configuration(self):
        try:
            with open(self.__gateway.get_config_path() + 'logs.conf', 'r', encoding="UTF-8") as logs:
                current_logs_configuration = logs.read()
            return current_logs_configuration
        except Exception as e:
            LOG.exception(e)

    def __update_logs_configuration(self):
        global LOG
        try:
            LOG = getLogger('service')
            logs_conf_file_path = self.__gateway.get_config_path() + 'logs.conf'
            new_logging_level = findall(r'level=(.*)', self.__new_logs_configuration.replace("NONE", "NOTSET"))[-1]
            new_logging_config = self.__new_logs_configuration.replace("NONE", "NOTSET").replace("\r\n", linesep)
            logs_config = ConfigParser(allow_no_value=True)
            logs_config.read_string(new_logging_config)
            for section in logs_config:
                if "handler_" in section and section != "handler_consoleHandler":
                    args = tuple(logs_config[section]["args"]
                                 .replace('(', '')
                                 .replace(')', '')
                                 .split(', '))
                    path = args[0][1:-1]
                    LOG.debug("Checking %s...", path)
                    if not exists(dirname(path)):
                        raise FileNotFoundError
            with open(logs_conf_file_path, 'w', encoding="UTF-8") as logs:
                logs.write(self.__new_logs_configuration.replace("NONE", "NOTSET") + "\r\n")
            fileConfig(logs_config)
            LOG = getLogger('service')
            # self.__gateway.remote_handler.deactivate()
            self.__gateway.remote_handler = TBLoggerHandler(self.__gateway)
            self.__gateway.main_handler.setLevel(new_logging_level)
            self.__gateway.main_handler.setTarget(self.__gateway.remote_handler)
            if new_logging_level == "NOTSET":
                self.__gateway.remote_handler.deactivate()
            else:
                self.__gateway.remote_handler.activate(new_logging_level)
            LOG.debug("Logs configuration has been updated.")
        except Exception as e:
            LOG.error("Remote logging configuration is wrong!")
            LOG.exception(e) class TBModuleLoader:
    PATHS = []
    LOADED_CONNECTORS = {}

    @staticmethod
    def find_paths():
        root_path = path.abspath(path.dirname(path.dirname(__file__)))
        log.debug("Root path is: " + root_path)
        if path.exists(DEB_INSTALLATION_EXTENSION_PATH):
            log.debug("Debian installation extensions folder exists.")
            TBModuleLoader.PATHS.append(DEB_INSTALLATION_EXTENSION_PATH)
        TBModuleLoader.PATHS.append(root_path + EXTENSIONS_FOLDER)
        TBModuleLoader.PATHS.append(root_path + CONNECTORS_FOLDER)

    @staticmethod
    def import_module(extension_type, module_name):
        if len(TBModuleLoader.PATHS) == 0:
            TBModuleLoader.find_paths()
        buffered_module_name = extension_type + module_name
        if TBModuleLoader.LOADED_CONNECTORS.get(buffered_module_name) is not None:
            return TBModuleLoader.LOADED_CONNECTORS[buffered_module_name]
        try:
            for current_path in TBModuleLoader.PATHS:
                current_extension_path = current_path + path.sep + extension_type
                if path.exists(current_extension_path):
                    for file in listdir(current_extension_path):
                        if not file.startswith('__') and file.endswith('.py'):
                            try:
                                module_spec = spec_from_file_location(module_name, current_extension_path + path.sep + file)
                                log.debug(module_spec)

                                if module_spec is None:
                                    continue

                                module = module_from_spec(module_spec)
                                module_spec.loader.exec_module(module)
                                for extension_class in getmembers(module, isclass):
                                    if module_name in extension_class:
                                        log.info("Import %s from %s.", module_name, current_extension_path)
                                        TBModuleLoader.LOADED_CONNECTORS[buffered_module_name] = extension_class[1]
                                        return extension_class[1]
                            except ImportError:
                                continue
        except Exception as e:
            log.exception(e) class TBLoggerHandler(logging.Handler):
    def __init__(self, gateway):
        self.current_log_level = 'INFO'
        super().__init__(logging.getLevelName(self.current_log_level))
        self.setLevel(logging.getLevelName('DEBUG'))
        self.__gateway = gateway
        self.activated = False
        self.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - [%(filename)s] - %(module)s - %(lineno)d - %(message)s'))
        self.loggers = ['service',
                        'extension',
                        'converter',
                        'connector',
                        'tb_connection'
                        ]
        for logger in self.loggers:
            log = logging.getLogger(logger)
            log.addHandler(self.__gateway.main_handler)
            log.debug("Added remote handler to log %s", logger)

    def activate(self, log_level=None):
        try:
            for logger in self.loggers:
                if log_level is not None and logging.getLevelName(log_level) is not None:
                    if logger == 'tb_connection' and log_level == 'DEBUG':
                        log = logging.getLogger(logger)
                        log.setLevel(logging.getLevelName('INFO'))
                    else:
                        log = logging.getLogger(logger)
                        self.current_log_level = log_level
                        log.setLevel(logging.getLevelName(log_level))
        except Exception as e:
            log = logging.getLogger('service')
            log.exception(e)
        self.activated = True

    def handle(self, record):
        if self.activated and not self.__gateway.stopped:
            record = self.formatter.format(record)
            self.__gateway.send_to_storage(self.__gateway.name, {"deviceName": self.__gateway.name, "telemetry": [{"ts": int(time()*1000), "values":{'LOGS': record}}]})

    def deactivate(self):
        self.activated = False

    @staticmethod
    def set_default_handler():
        logger_names = [
            'service',
            'storage',
            'extension',
            'converter',
            'connector',
            'tb_connection'
            ]
        for logger_name in logger_names:
            logger = logging.getLogger(logger_name)
            handler = logging.StreamHandler(stdout)
            handler.setFormatter(logging.Formatter('[STREAM ONLY] %(asctime)s - %(levelname)s - [%(filename)s] - %(module)s - %(lineno)d - %(message)s'))
            logger.addHandler(handler) class RemoteShell:
    def __init__(self, platform, release):
        self.__session_active = False
        self.__platform = platform
        self.__release = release
        self.shell_commands = {
            "getTermInfo": self.get_term_info,
            "sendCommand": self.send_command,
            "getCommandStatus": self.get_command_status,
            "terminateCommand": self.terminate_command,
            }
        self.command_in_progress = None
        self.__previous_stdout = b""
        self.__previous_stderr = b""

    def get_term_info(self, *args):
        return {"platform": self.__platform, "release": self.__release, "cwd": str(getcwd())}

    def send_command(self, *args):
        result = {"ok": False, "qos": 0}
        log.debug("Received command to shell with args: %r", args)
        command = args[0]['command']
        cwd = args[0].get('cwd')
        if cwd is not None and str(getcwd()) != cwd:
            chdir(cwd)
        if command.split():
            if self.command_in_progress is not None:
                log.debug("Received a new command: \"%s\", during old command is running, terminating old command...", command)
                old_command = self.command_in_progress.args
                self.terminate_command()
                log.debug("Old command: \"%s\" terminated.", old_command)
            if command.split()[0] in ["quit", "exit"]:
                self.command_in_progress = None
            elif command.split()[0] == "cd":
                chdir(command.split()[1])
                self.command_in_progress = "cd"
            else:
                log.debug("Run command in remote shell: %s", command)
                self.command_in_progress = Popen(command, shell=True, stdout=PIPE, stdin=PIPE, stderr=STDOUT, universal_newlines=True)
            result.update({"ok": True})
        return result

    def get_command_status(self, *args):
        result = {"data": [{"stdout": "",
                            "stderr": ""}],
                  "cwd": str(getcwd()),
                  "done": True,
                  "qos": 0,
                  }
        done = False
        if self.command_in_progress == "cd":
            done = True
        elif self.command_in_progress is not None:
            stdout_value = b""
            stderr_value = b""
            done = True if self.command_in_progress.poll() is not None else False
            try:
                stdout_value, stderr_value = self.command_in_progress.communicate(timeout=.1)
            except TimeoutExpired as e:
                log.debug("Process is run")
                stdout_value = b"" if e.stdout is None else e.stdout.replace(self.__previous_stdout, b"")
                stderr_value = b"" if e.stderr is None else e.stderr.replace(self.__previous_stderr, b"")
                stdout_value = stdout_value[:-1] if len(stdout_value) and stdout_value[-1] == b"\n" else stdout_value
                stderr_value = stderr_value[:-1] if len(stderr_value) and stderr_value[-1] == b"\n" else stderr_value
            self.__previous_stderr = self.__previous_stderr + b"" if stderr_value is None else stderr_value
            self.__previous_stdout = self.__previous_stdout + b"" if stdout_value is None else stdout_value
            str_stdout = str(stdout_value, "UTF-8") if isinstance(stdout_value, bytes) else stdout_value
            str_stderr = str(stderr_value, "UTF-8") if isinstance(stderr_value, bytes) else stderr_value
            result.update({"data": [{"stdout": str_stdout,
                                     "stderr": str_stderr}]
                           })
        result.update({"done": done})
        if done:
            self.command_in_progress = None

        return result

    def terminate_command(self, *args):
        result = {"ok": False}
        if self.command_in_progress is not None:
            try:
                self.command_in_progress.terminate()
                self.__previous_stderr = b""
                self.__previous_stdout = b""
                result.update({"ok": True})
            except Exception as e:
                log.exception(e)
                result["error"] = str(e)
        else:
            result["error"] = "Process for termination not found."
        return result class TBUpdater(Thread):
    def __init__(self):
        super().__init__()
        self.__version = {"current_version": get_distribution('thingsboard_gateway').version,
                          "latest_version": get_distribution('thingsboard_gateway').version}
        self.__instance_id = str(uuid1())
        self.__platform = system()
        self.__release = release()
        self.__os_version = platform()
        self.__previous_check = 0
        self.__check_period = 3600.0
        self.__request_timeout = 5
        self.__stopped = True
        self.start()

    def run(self):
        self.__stopped = False
        while not self.__stopped:
            if time() >= self.__previous_check + self.__check_period:
                self.check_for_new_version()
                self.__previous_check = time()
            else:
                sleep(1)

    def stop(self):
        self.__stopped = True

    def get_version(self):
        return self.__version

    def get_platform(self):
        return self.__platform

    def get_release(self):
        return self.__release

    def check_for_new_version(self):
        log.debug("Checking for new version")
        request_args = self.form_request_params()
        try:
            response = post(**request_args)
            content = None
            content = loads(response.content)
            if content is not None and content.get("updateAvailable", False):
                new_version = content["message"].replace("New version ", "").replace(" is available!", "")
                if new_version > self.__version["current_version"]:
                    log.info(content["message"])
                    self.__version["latest_version"] = new_version
                    log.info("\n\n[===UPDATE===]\n\n New version %s is available! \n\n[===UPDATE===]\n",
                             self.__version["latest_version"])
        except ConnectionRefusedError:
            log.warning("Cannot connect to the update service. PLease check your internet connection.")
        except ConnectionError:
            log.warning("Cannot connect to the update service. PLease check your internet connection.")
        except Exception as e:
            log.exception(e)

    def form_request_params(self):
        json_data = {
            "version": self.__version["current_version"],
            "platform": self.__platform,
            "instanceId": self.__instance_id,
            "osVersion": self.__os_version,
            }
        url = UPDATE_SERVICE_BASE_URL + "/api/tb-gateway/updates"
        request_args = {
            "url": url,
            "json": json_data,
            "timeout": self.__request_timeout

            }
        return request_args

    def update(self):
        if self.__version["latest_version"] != self.__version["current_version"]:
            result = TBUtility.install_package("thingsboard-gateway", self.__version["latest_version"])
        else:
            result = "Congratulations! You have the latest version."
        return result class TBUtility:

    @staticmethod
    def decode(message):
        try:
            if isinstance(message.payload, bytes):
                content = loads(message.payload.decode("utf-8", "ignore"))
            else:
                content = loads(message.payload)
        except JSONDecodeError:
            try:
                content = message.payload.decode("utf-8", "ignore")
            except JSONDecodeError:
                content = message.payload
        return content

    @staticmethod
    def validate_converted_data(data):
        error = None
        if error is None and not data.get("deviceName"):
            error = 'deviceName is empty in data: '

        if error is None:
            got_attributes = False
            got_telemetry = False

            if data.get("attributes") is not None and len(data.get("attributes")) > 0:
                got_attributes = True

            if data.get("telemetry") is not None:
                for entry in data.get("telemetry"):
                    if (entry.get("ts") is not None and len(entry.get("values")) > 0) or entry.get("ts") is None:
                        got_telemetry = True
                        break

            if got_attributes is False and got_telemetry is False:
                error = 'No telemetry and attributes in data: '

        if error is not None:
            json_data = dumps(data)
            if isinstance(json_data, bytes):
                log.error(error + json_data.decode("UTF-8"))
            else:
                log.error(error + json_data)
            return False
        return True

    @staticmethod
    def topic_to_regex(topic):
        return topic.replace("+", "[^/]+").replace("#", ".+")

    @staticmethod
    def regex_to_topic(regex):
        return regex.replace("[^/]+", "+").replace(".+", "#")

    @staticmethod
    def get_value(expression, body=None, value_type="string", get_tag=False, expression_instead_none=False):
        if isinstance(body, str):
            body = loads(body)
        if not expression:
            return ''
        positions = search(r'\${(?:(.*))}', expression)
        if positions is not None:
            p1 = positions.regs[-1][0]
            p2 = positions.regs[-1][1]
        else:
            p1 = 0
            p2 = len(expression)
        target_str = str(expression[p1:p2])
        if get_tag:
            return target_str
        full_value = None
        try:
            if isinstance(body, dict) and target_str.split()[0] in body:
                if value_type.lower() == "string":
                    full_value = str(expression[0: max(p1 - 2, 0)]) + str(body[target_str.split()[0]]) + str(expression[
                                                                                                             p2 + 1:len(
                                                                                                                 expression)])
                else:
                    full_value = body.get(target_str.split()[0])
            elif isinstance(body, (dict, list)):
                try:
                    jsonpath_expression = parse(target_str)
                    jsonpath_match = jsonpath_expression.find(body)
                    if jsonpath_match:
                        full_value = jsonpath_match[0].value
                except Exception as e:
                    log.debug(e)
            elif isinstance(body, (str, bytes)):
                search_result = search(expression, body)
                if search_result.groups():
                    full_value = search_result.group(0)
            if expression_instead_none and full_value is None:
                full_value = expression
        except Exception as e:
            log.exception(e)
        return full_value

    @staticmethod
    def get_values(expression, body=None, value_type="string", get_tag=False, expression_instead_none=False):
        expression_arr = findall(r'\$\{[${A-Za-z0-9.^\]\[*_]*\}', expression)

        values = [TBUtility.get_value(exp, body, value_type=value_type, get_tag=get_tag,
                                      expression_instead_none=expression_instead_none) for exp in expression_arr]

        if '${' not in expression:
            values.append(expression)

        return values

    @staticmethod
    def install_package(package, version="upgrade"):
        from sys import executable
        from subprocess import check_call, CalledProcessError
        result = False
        if version.lower() == "upgrade":
            try:
                result = check_call([executable, "-m", "pip", "install", package, "--upgrade", "--user"])
            except CalledProcessError:
                result = check_call([executable, "-m", "pip", "install", package, "--upgrade"])
        else:
            from pkg_resources import get_distribution
            current_package_version = None
            try:
                current_package_version = get_distribution(package)
            except Exception:
                pass
            if current_package_version is None or current_package_version != version:
                installation_sign = "==" if ">=" not in version else ""
                try:
                    result = check_call(
                        [executable, "-m", "pip", "install", package + installation_sign + version, "--user"])
                except CalledProcessError:
                    result = check_call([executable, "-m", "pip", "install", package + installation_sign + version])
        return result

    @staticmethod
    def replace_params_tags(text, data):
        if '${' in text:
            for item in text.split('/'):
                if '${' in item:
                    tag = '${' + TBUtility.get_value(item, data['data'], 'params', get_tag=True) + '}'
                    value = TBUtility.get_value(item, data['data'], 'params', expression_instead_none=True)
                    text = text.replace(tag, str(value))
        return text

    @staticmethod
    def get_dict_key_by_value(dictionary: dict, value):
        return list(dictionary.values())[list(dictionary.values()).index(value)]
import logging
import logging.config
import logging.handlers
from copy import deepcopy
from os import execv, listdir, path, pathsep, stat, system
from queue import SimpleQueue
from random import choice
from string import ascii_lowercase, hexdigits
from sys import argv, executable, getsizeof
from threading import RLock, Thread
from time import sleep, time
from simplejson import JSONDecodeError, dumps, load, loads
from yaml import safe_load
from thingsboard_gateway.gateway.constant_enums import DeviceActions, Status
from thingsboard_gateway.gateway.constants import CONNECTED_DEVICES_FILENAME, CONNECTOR_PARAMETER, PERSISTENT_GRPC_CONNECTORS_KEY_FILENAME
from thingsboard_gateway.gateway.tb_client import TBClient
from thingsboard_gateway.storage.file.file_event_storage import FileEventStorage
from thingsboard_gateway.storage.memory.memory_event_storage import MemoryEventStorage
from thingsboard_gateway.storage.sqlite.sqlite_event_storage import SQLiteEventStorage
from thingsboard_gateway.tb_utility.tb_gateway_remote_configurator import RemoteConfigurator
from thingsboard_gateway.tb_utility.tb_loader import TBModuleLoader
from thingsboard_gateway.tb_utility.tb_logger import TBLoggerHandler
from thingsboard_gateway.tb_utility.tb_remote_shell import RemoteShell
from thingsboard_gateway.tb_utility.tb_updater import TBUpdater
from thingsboard_gateway.tb_utility.tb_utility import TBUtility
    from thingsboard_gateway.gateway.grpc_service.grpc_connector import GrpcConnector
    from thingsboard_gateway.gateway.grpc_service.tb_grpc_manager import TBGRPCServerManager
            self.__grpc_manager = TBGRPCServerManager(self, self.__grpc_config)
            self.__grpc_manager.set_gateway_read_callbacks(self.__register_connector, self.__unregister_connector)
        self._load_connectors()
        self._connect_with_connectors()
        self.__load_persistent_devices()

        self.__devices_idle_checker = self.__config['thingsboard'].get('checkingDeviceActivity', {})
        self.__check_devices_idle = self.__devices_idle_checker.get('checkDeviceInactivity', False)
        if self.__check_devices_idle:
            thread = Thread(name='Checking devices idle time', target=self.__check_devices_idle_time, daemon=True)
            thread.start()
            log.info('Start checking devices idle time')

        self._published_events = SimpleQueue()
        self._send_thread = Thread(target=self.__read_data_from_storage, daemon=True,
                                   name="Send data to Thingsboard Thread")
        self._send_thread.start()
        self.__min_pack_send_delay_ms = self.__config['thingsboard'].get('minPackSendDelayMS', 500) / 1000.0
        log.info("Gateway started.")

        try:
            gateway_statistic_send = 0
            connectors_configuration_check_time = 0
            while not self.stopped:
                cur_time = time() * 1000
                if not self.tb_client.is_connected() and self.__subscribed_to_rpc_topics:
                    self.__subscribed_to_rpc_topics = False
                if self.tb_client.is_connected() and not self.__subscribed_to_rpc_topics:
                    for device in self.__saved_devices:
                        self.add_device(device, {"connector": self.__saved_devices[device]["connector"]},
                                        device_type=self.__saved_devices[device]["device_type"])
                    self.subscribe_to_required_topics()
                    self.__subscribed_to_rpc_topics = True
                if self.__scheduled_rpc_calls:
                    for rpc_call_index in range(len(self.__scheduled_rpc_calls)):
                        rpc_call = self.__scheduled_rpc_calls[rpc_call_index]
                        if cur_time > rpc_call[0]:
                            rpc_call = self.__scheduled_rpc_calls.pop(rpc_call_index)
                            result = None
                            try:
                                result = rpc_call[1]["function"](*rpc_call[1]["arguments"])
                            except Exception as e:
                                log.exception(e)
                            if result == 256:
                                log.warning("Error on RPC command: 256. Permission denied.")
                if (self.__rpc_requests_in_progress or not self.__rpc_register_queue.empty()) and self.tb_client.is_connected():
                    new_rpc_request_in_progress = {}
                    if self.__rpc_requests_in_progress:
                        for rpc_in_progress, data in self.__rpc_requests_in_progress.items():
                            if cur_time >= data[1]:
                                data[2](rpc_in_progress)
                                self.cancel_rpc_request(rpc_in_progress)
                                self.__rpc_requests_in_progress[rpc_in_progress] = "del"
                        new_rpc_request_in_progress = {key: value for key, value in
                                                       self.__rpc_requests_in_progress.items() if value != 'del'}
                    if not self.__rpc_register_queue.empty():
                        rpc_request_from_queue = self.__rpc_register_queue.get(False)
                        topic = rpc_request_from_queue["topic"]
                        data = rpc_request_from_queue["data"]
                        new_rpc_request_in_progress[topic] = data
                    self.__rpc_requests_in_progress = new_rpc_request_in_progress
                else:
                    try:
                        sleep(0.2)
                    except Exception as e:
                        log.exception(e)
                        break
                if not self.__request_config_after_connect and self.tb_client.is_connected() and not self.tb_client.client.get_subscriptions_in_progress():
                    self.__request_config_after_connect = True
                    self.__check_shared_attributes()

                if cur_time - gateway_statistic_send > self.__config["thingsboard"].get("statsSendPeriodInSeconds",
                                                                                        3600) * 1000 and self.tb_client.is_connected():
                    summary_messages = self.__form_statistics()
                    # with self.__lock:
                    self.tb_client.client.send_telemetry(summary_messages)
                    gateway_statistic_send = time() * 1000
                    # self.__check_shared_attributes()

                if cur_time - connectors_configuration_check_time > self.__config["thingsboard"].get(
                        "checkConnectorsConfigurationInSeconds", 60) * 1000:
                    self.check_connector_configuration_updates()
                    connectors_configuration_check_time = time() * 1000

                if cur_time - self.__updates_check_time >= self.__updates_check_period_ms:
                    self.__updates_check_time = time() * 1000
                    self.version = self.__updater.get_version()

        except KeyboardInterrupt:
            self.__stop_gateway()
        except Exception as e:
            log.exception(e)
            self.__stop_gateway()
            self.__close_connectors()
            log.info("The gateway has been stopped.")
            self.tb_client.stop()

    def __close_connectors(self):
        for current_connector in self.available_connectors:
            try:
                self.available_connectors[current_connector].close()
                log.debug("Connector %s closed connection.", current_connector)
            except Exception as e:
                log.exception(e)

    def __stop_gateway(self):
        self.stopped = True
        self.__updater.stop()
        log.info("Stopping...")
        if self.__grpc_manager is not None:
            self.__grpc_manager.stop()
        self.__close_connectors()
        self._event_storage.stop()
        log.info("The gateway has been stopped.")
        self.tb_client.disconnect()
        self.tb_client.stop()

    def __init_remote_configuration(self, force=False):
        if (self.__config["thingsboard"].get("remoteConfiguration") or force) and self.__remote_configurator is None:
            try:

                self.__remote_configurator = RemoteConfigurator(self, self.__config)