Watchdog Module

mosaicolabs.watchdog.watchdog.Watchdog

Watchdog(watchdog_config)

Scans a folder and injects every new MCAP or ROS bag file into a Mosaico server.

The files are listed by a LocalFileSource (filtered by supported extension, glob_pattern and min_file_age_s), skipped if the FileStateStore already knows them, and injected one at a time with the injector chosen by the InjectorDispatcher. Each outcome is recorded: loaded in .already_loaded.txt, quarantined in .quarantine_file.txt.

• Single-pass: run() scans once in the calling thread, injects every new file, then returns.
• Daemon: run() starts a producer thread that scans the folder every polling_interval_s seconds and fills the queue, while the calling thread injects the queued files, until Ctrl-C or an error stops it.

Example

from pathlib import Path

watchdog = Watchdog(WatchdogConfig(path_to_monitor=Path("/data/bags")))
watchdog.run()

Parameters:

Name	Type	Description	Default
watchdog_config	WatchdogConfig	The configuration of the Watchdog.	required

Raises:

Type	Description
ValueError	If path_to_monitor does not exist or is not a folder.
RuntimeError	If a state file exists in path_to_monitor but cannot be read.

file_store_state instance-attribute

file_store_state = FileStateStore(path_to_monitor)

Records which files are loaded, quarantined or in the queue.

run

run()

Runs the Watchdog in the configured mode, blocking the calling thread.

• Single-pass: scans the folder once, injects every new file, then returns.
• Daemon: starts the producer thread and injects the queued files until Ctrl-C or an error stops it. The producer thread is always stopped and joined before run() raises.

Raises:

Type	Description
KeyboardInterrupt	On Ctrl-C. The file being injected is not marked as loaded.
Exception	Any injection error, when on_error is RAISE.

mosaicolabs.watchdog.watchdog.WatchdogConfig dataclass

WatchdogConfig(
    path_to_monitor,
    mode=SINGLEPASS,
    polling_interval_s=5.0,
    glob_pattern=None,
    min_file_age_s=None,
    on_error=QUARANTINE,
    retry_number=2,
    max_queue_size=100,
    host="localhost",
    port=6726,
    mosaico_api_key=None,
    tls_cert_path=None,
    enable_tls=False,
    log_level="INFO",
)

The configuration of a Watchdog: the folder to watch, how to scan it, what to do on errors, and how to reach the Mosaico server.

Attributes:

Name	Type	Description
path_to_monitor	Path	The folder to watch, subfolders included. It must exist and be writable: the state files (.already_loaded.txt, .quarantine_file.txt) are written inside it.
mode	WatchdogMode	SINGLEPASS scans once and returns, DAEMON keeps scanning until stopped. Default: WatchdogMode.SINGLEPASS
polling_interval_s	float	Seconds between the start of two scans. Daemon mode only. Default: 5.0
glob_pattern	Optional[str]	Only files matching this glob pattern are injected; it is matched at any depth (e.g. "run_*/*.mcap"). None keeps every file with a supported extension. Default: None
min_file_age_s	Optional[float]	Files modified less than this many seconds ago are skipped, since they may still be written; they are considered again at the next scan. None disables the check. Default: None
on_error	WatchdogError	What to do when the injection of a file fails (see WatchdogError). Default: WatchdogError.QUARANTINE
retry_number	int	How many times a failed file is tried again before being quarantined. Used only when on_error is WatchdogError.RETRY. Default: 2
max_queue_size	int	Maximum number of files waiting to be injected. When the queue is full, the remaining new files wait for the next scan. Daemon mode only: in single-pass mode the queue is unbounded. Default: 100
host	str	Hostname or IP of the Mosaico server. Default: "localhost"
port	int	Port of the Mosaico server. Default: 6726
mosaico_api_key	Optional[str]	API key for the Mosaico server. If provided, it must have the write permission. Default: None
tls_cert_path	Optional[str]	Path to the TLS certificate file for a secure connection to the Mosaico server. Default: None
enable_tls	bool	Use TLS for the connection to the Mosaico server. Default: False
log_level	str	Logging verbosity level ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"). It is also passed to the injectors, which reconfigure logging when they start. Default: "INFO"

Example

from pathlib import Path

config = WatchdogConfig(
    path_to_monitor=Path("/data/bags"),
    mode=WatchdogMode.DAEMON,
    min_file_age_s=10.0,
    on_error=WatchdogError.RETRY,
)

path_to_monitor instance-attribute

path_to_monitor

The folder to watch, subfolders included. It must exist and be writable: the state files (.already_loaded.txt, .quarantine_file.txt) are written inside it.

mode class-attribute instance-attribute

mode = SINGLEPASS

SINGLEPASS scans once and returns, DAEMON keeps scanning until stopped.

polling_interval_s class-attribute instance-attribute

polling_interval_s = 5.0

Seconds between the start of two scans. Daemon mode only.

glob_pattern class-attribute instance-attribute

glob_pattern = None

Only files matching this glob pattern are injected; it is matched at any depth (e.g. "run_*/*.mcap"). None keeps every file with a supported extension.

min_file_age_s class-attribute instance-attribute

min_file_age_s = None

Files modified less than this many seconds ago are skipped, since they may still be written; they are considered again at the next scan. None disables the check.

on_error class-attribute instance-attribute

on_error = QUARANTINE

What to do when the injection of a file fails (see WatchdogError).

retry_number class-attribute instance-attribute

retry_number = 2

How many times a failed file is tried again before being quarantined. Used only when on_error is WatchdogError.RETRY.

max_queue_size class-attribute instance-attribute

max_queue_size = 100

Maximum number of files waiting to be injected. When the queue is full, the remaining new files wait for the next scan. Daemon mode only: in single-pass mode the queue is unbounded.

host class-attribute instance-attribute

host = 'localhost'

Hostname or IP of the Mosaico server.

port class-attribute instance-attribute

port = 6726

Port of the Mosaico server.

mosaico_api_key class-attribute instance-attribute

mosaico_api_key = None

API key for the Mosaico server. If provided, it must have the write permission.

tls_cert_path class-attribute instance-attribute

tls_cert_path = None

Path to the TLS certificate file for a secure connection to the Mosaico server.

enable_tls class-attribute instance-attribute

enable_tls = False

Use TLS for the connection to the Mosaico server.

log_level class-attribute instance-attribute

log_level = 'INFO'

Logging verbosity level ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"). It is also passed to the injectors, which reconfigure logging when they start.

is_deamon_mode

is_deamon_mode()

Returns True if the Watchdog runs in daemon mode.

mosaicolabs.watchdog.watchdog.WatchdogMode

Bases: Enum

How long the Watchdog runs.

SINGLEPASS class-attribute instance-attribute

SINGLEPASS = 'single-pass'

Scan the folder once, inject every new file, then return.

DAEMON class-attribute instance-attribute

DAEMON = 'daemon'

Scan the folder every polling_interval_s seconds until stopped (Ctrl-C or an error).

mosaicolabs.watchdog.watchdog.WatchdogError

Bases: Enum

What the Watchdog does when the injection of a file fails.

RAISE class-attribute instance-attribute

RAISE = 'raise'

Stop the Watchdog and re-raise the error to the caller of run().

RETRY class-attribute instance-attribute

RETRY = 'retry'

Try the file again, up to retry_number more times, then quarantine it.

QUARANTINE class-attribute instance-attribute

QUARANTINE = 'quarantine'

Quarantine the file at the first failure and move on to the next one.

mosaicolabs.watchdog.file_source.FileRef dataclass

FileRef(relative_path, uri, size, mtime_ns)

Immutable description of a candidate file found by a File Source. It holds no content.

Only a File Source creates it. Two FileRef are the same file when their relative_path is equal: uri, size and mtime_ns are excluded from comparison and hashing, since size and modification time change between scans.

relative_path instance-attribute

relative_path

Path relative to the monitored folder, with / as separator and the extension (e.g. run_01/drive.mcap). It is the key used by the FileStateStore and the base of the sequence name.

uri class-attribute instance-attribute

uri = field(compare=False)

Where the file is: a file:// URI for local files (e.g. file:///data/bags/run_01/drive.mcap), or s3://bucket/key for object stores (not implemented yet). Use FileSource.get_local_path() to get a local path.

size class-attribute instance-attribute

size = field(compare=False)

File size in bytes, at scan time.

mtime_ns class-attribute instance-attribute

mtime_ns = field(compare=False)

Last modification time, in nanoseconds since the Unix epoch, at scan time.

name property

name

The file name, with extension: run_01/drive.mcap -> drive.mcap.

mosaicolabs.watchdog.file_source.FileSource

Bases: Protocol

Contract of a File Source: lists the candidate files of a local folder or an object store, and gives access to their content as local files.

list_files

list_files()

Lists the candidate files of the location, subfolders included.

Returns:

Type	Description
Iterable[FileRef]	Iterable[FileRef]: One FileRef per candidate file.

get_local_path

get_local_path(ref)

Returns a local path to read the file described by ref.

A local source returns the path of the file itself. An object store source downloads the file to a staging folder and returns the path of the copy.

Parameters:

Name	Type	Description	Default
ref	FileRef	A FileRef returned by this source's list_files().	required

Returns:

Name	Type	Description
Path	Path	A local path to the content of the file.

mosaicolabs.watchdog.file_source.LocalFileSource

LocalFileSource(
    path_to_monitor,
    accepted_extensions,
    glob_pattern=None,
    min_file_age_s=None,
)

FileSource for a local folder.

Lists every file in path_to_monitor and, recursively, its subfolders, keeping only the files that:

• have one of the supported extensions;
• match glob_pattern, if set;
• were last modified at least min_file_age_s seconds ago, if set, since a more recent file may still be being written.

A skipped file is not recorded anywhere, so it is evaluated again at the next scan.

Parameters:

Name	Type	Description	Default
path_to_monitor	Path	The folder to list. It is resolved to an absolute path once, so later changes of the working directory do not affect it.	required
accepted_extensions	Tuple[str, ...]	The extensions to keep, with or without the leading dot (e.g. InjectorDispatcher.list_supported_ext()). They are normalized to lower case with a leading dot, and file names are matched case-sensitively, so drive.MCAP is not listed.	required
glob_pattern	Optional[str]	A glob pattern the files must also match. As with Path.rglob, it is matched at any depth (e.g. "run_*/*.mcap" also matches a/b/run_01/x.mcap). Default: None (all files).	None
min_file_age_s	Optional[float]	Minimum time, in seconds, since the last modification of a file for it to be listed. None or 0 disables the check. Default: None	None

Raises:

Type	Description
ValueError	If path_to_monitor does not exist or is not a folder.

list_files

list_files()

Lists the files of the monitored folder that pass the filters (see the class docstring).

The listing is lazy: the folder is scanned while the result is iterated, and the result can be iterated only once, so call list_files() again for each scan. A file deleted during the scan is skipped.

Yields:

Name	Type	Description
FileRef	Iterable[FileRef]	One per file, with relative_path relative to path_to_monitor.

get_local_path

get_local_path(ref)

Returns the path of the file described by ref, inside the monitored folder. Nothing is copied: the path points to the file itself.

Parameters:

Name	Type	Description	Default
ref	FileRef	A FileRef returned by list_files().	required

Returns:

Name	Type	Description
Path	Path	path_to_monitor / ref.relative_path.

mosaicolabs.watchdog.file_state_store.FileStateStore

FileStateStore(watched_dir)

Keeps the state of every file the Watchdog has seen: loaded, quarantined or in_queue. A file with no state is new.

The loaded and quarantined states are persisted in .already_loaded.txt and .quarantine_file.txt inside watched_dir: UTF-8, one relative path per line, only appended to by the store. Both files are read again by refresh(), so manual edits take effect while the Watchdog runs, and a missing file counts as empty. The in_queue state lives in memory only and starts empty at every Watchdog start.

A path in more than one state counts as quarantined, then loaded, then in_queue. The public methods hold a lock, since the Producer and the Consumer threads use the store at the same time.

Parameters:

Name	Type	Description	Default
watched_dir	Path	The folder holding the state files. The Watchdog passes the monitored folder itself (path_to_monitor), so it must be writable.	required

Raises:

Type	Description
ValueError	If watched_dir does not exist or is not a folder.
RuntimeError	If a state file exists but cannot be read.

refresh

refresh()

Reads .already_loaded.txt and .quarantine_file.txt again, keeping the in_queue entries. The Watchdog calls it at the start of every scan, so manual edits take effect while it runs.

The cache is replaced only once both files have been read: if one cannot be read, a warning is logged and the previous state is kept until the next call.

mark_loaded

mark_loaded(file_ref)

Marks the file as loaded: appends its relative path to .already_loaded.txt and replaces its in_queue state. The Consumer calls it after a successful injection.

Parameters:

Name	Type	Description	Default
file_ref	FileRef	The file loaded successfully.	required

Raises:

Type	Description
RuntimeError	If its relative path is blank or contains a line break.
OSError	If .already_loaded.txt cannot be written.

mark_quarantined

mark_quarantined(file_ref)

Marks the file as quarantined: appends its relative path to .quarantine_file.txt and replaces its in_queue state. The Consumer calls it when the injection failed (see on_error).

Parameters:

Name	Type	Description	Default
file_ref	FileRef	The file whose injection failed.	required

Raises:

Type	Description
RuntimeError	If its relative path is blank or contains a line break.
OSError	If .quarantine_file.txt cannot be written.

mark_in_queue

mark_in_queue(file_ref)

Marks the file as in_queue, in memory only. The Producer calls it before appending the file to the queue, so later scans do not queue it again until its outcome is recorded with mark_loaded() or mark_quarantined().

Parameters:

Name	Type	Description	Default
file_ref	FileRef	The file about to be queued.	required

Raises:

Type	Description
RuntimeError	If its relative path is blank or contains a line break.

get_state

get_state(file_ref)

Returns the state of a file.

Parameters:

Name	Type	Description	Default
file_ref	FileRef	The file you want to know the state of.	required

Returns:

Type	Description
Optional[FileState]	Optional[FileState]: The state of the file, or None if the file is new.

mosaicolabs.watchdog.file_state_store.FileState

Bases: Enum

The state of a file known to the FileStateStore. A file with no state is new.

LOADED class-attribute instance-attribute

LOADED = 1

Injected successfully: listed in .already_loaded.txt.

QUARANTINED class-attribute instance-attribute

QUARANTINED = 2

Injection failed: listed in .quarantine_file.txt.

IN_QUEUE class-attribute instance-attribute

IN_QUEUE = 3

Queued or currently being loaded. Kept in memory only.

mosaicolabs.watchdog.dispatcher.InjectorDispatcher

Chooses the injector for a file, from its extension and, for .mcap files, from the encoding of its channels:

• .bag and .db3: RosbagInjector.
• .mcap: RosbagInjector if the file has at least one channel and every channel has a ROS 2 schema (ros2msg or ros2idl). Otherwise MCAPInjector, which skips the channels it cannot decode: this covers schemaless channels, other encodings and ROS 1 .mcap files.

Injectors are not registered: the dispatcher depends on them directly.

SUPPORTED_EXTENSIONS class-attribute instance-attribute

SUPPORTED_EXTENSIONS = ('.mcap', '.db3', '.bag')

The file extensions the Watchdog can load. Matching is case-sensitive.

list_supported_ext classmethod

list_supported_ext()

Returns the supported file extensions, to pass to the File Source so that it only lists files the dispatcher can handle.

Returns:

Type	Description
Tuple[str, ...]	Tuple[str, ...]: The extensions, with the leading dot (e.g. ".mcap").

get_injector classmethod

get_injector(absolute_file_path)

Returns the injector class to load a file with. For .mcap files the file is opened to read its summary (channels and schemas), so it must be available locally: the Watchdog passes the path returned by FileSource.get_local_path().

Parameters:

Name	Type	Description	Default
absolute_file_path	Path	The local path of the file.	required

Returns:

Type	Description
Type[RosbagInjector] | Type[MCAPInjector]	Type[RosbagInjector] | Type[MCAPInjector]: The injector class, not an instance.

Raises:

Type	Description
RuntimeError	If the extension is not supported, or if an .mcap file has no summary section.
FileNotFoundError	If an .mcap file does not exist (e.g. deleted after the scan).
McapError	If an .mcap file is corrupt or truncated.