ddp_utils.file_scanner

import ddp_utils.file_scanner

Poll a directory for stable signal files and dispatch their handlers.

The scanner supports exact names and glob patterns, adaptive polling, file stability checks, retry policies, staged idle timeouts, health checks, and structured terminal results. Handler failures are recoverable unless their watch policy makes them fatal.

Examples

Stop after handling a completion signal:

from ddp_utils.file_scanner import FileScanner, WatchSpec, ScanDecision

scanner = FileScanner("downloads")
scanner.add_watch(WatchSpec("done.json", lambda: ScanDecision.STOP))
result = scanner.run(timeout_seconds=60)

ddp_utils.file_scanner.normalize_file_name_pattern(pattern: str) → str

Convert brace placeholders in a filename template to one glob star.

Parameters

Name	Type	Description
pattern	str	Filename template such as "{GUID}_{CASE_NO}.json".

Returns

Type	Description
str	Glob pattern with every {...} placeholder replaced by *, spaces around stars removed, and consecutive stars collapsed. An empty result becomes "*".

Raises

Exception	Description
ValueError	pattern is not a string.

Examples

Normalize a signal filename template:

pattern = normalize_file_name_pattern("{GUID}_{CASE_NO}")
assert pattern == "*_*"

class ddp_utils.file_scanner.ScanDecision(value)

Bases: Enum

Select scanner behavior after a handler succeeds.

CONTINUE keeps polling, STOP terminates normally, and DISABLE removes the successful watch while leaving the scanner running.

Examples

Stop the scanner from a completion handler:

return ScanDecision.STOP

class ddp_utils.file_scanner.StopReason(value)

Bases: Enum

Describe why a scan terminated without propagating an exception.

Values distinguish explicit stops, three timeout classes, user interrupt, fatal handler or scanner failure, and external health-check failure.

Examples

Branch on an idle timeout:

if result.reason is StopReason.IDLE_TIMEOUT:
    print("No signal arrived")

class ddp_utils.file_scanner.ScanResult(reason: StopReason, message: str = '', data: Mapping[str, ~typing.Any]=<factory>)

Bases: object

Represent the terminal outcome of FileScanner.run().

Variables

Name	Type	Description
reason	StopReason	Structured reason for termination.
message	str	Human-readable explanation.
data	Mapping[str, Any]	Additional immutable-by-contract context supplied by the scanner.

Examples

Construct an explicit successful stop result:

result = ScanResult(StopReason.STOP_REQUESTED, "completed")

class ddp_utils.file_scanner.WatchSpec(filename: str, handler: Callable[[...], Any], decision_on_success: ScanDecision = ScanDecision.CONTINUE, remove_after_handled: bool = True, file_kind: str = 'generic', fatal_on_exception: bool = False, max_errors_before_fatal: int = 0, idle_timeout_after: int | None = None, max_retries: int = 0, retry_delay: float = 2.0)

Bases: object

Describe one exact-name or glob-pattern file watch.

Variables

Name	Type	Description
filename	str	Name, relative path, or glob pattern below base_dir.
handler	Callable[[...], Any]	Callable accepting no arguments, a file path, a FileEvent, or both event and path.
decision_on_success	ScanDecision	Fallback decision when the handler returns no explicit decision.
remove_after_handled	bool	Delete a successfully handled file after the scanner’s configured delay.
file_kind	str	Diagnostic category and staged-timeout lookup key.
fatal_on_exception	bool	Convert handler failures into fatal termination.
max_errors_before_fatal	int	Number of errors tolerated before a fatal watch stops the scanner. Zero makes the first error fatal.
idle_timeout_after	int | None	Positive idle timeout applied after success.
max_retries	int	Per-watch retry limit; zero uses the scanner default.
retry_delay	float	Delay before the same failed file is retried.

Examples

Watch a completion file and stop after handling it:

spec = WatchSpec(
    filename="done.json",
    handler=lambda path: print(path),
    decision_on_success=ScanDecision.STOP,
)

class ddp_utils.file_scanner.Rule

Bases: object

Build and append WatchSpec objects to a target.

Examples

Add a rule to a plain configuration holder:

Rule.add(holder, rule_data)

static add(target, data: Dict[str, Any]) → None

Append a validated watch definition to target.watch_specs.

Parameters

Name	Type	Description
target		Object whose watch_specs attribute is a list. A missing attribute is created automatically.
data	Dict[str, Any]	Rule mapping containing file_kind, callable handler, named decision_on_success, optional deletion and idle policies, and an already resolved filename.

Raises

Exception	Description
ValueError	Required values are missing, invalid, or inconsistent.
AttributeError	Existing target.watch_specs is not a list.

Examples

Add a generated rule to a scanner configuration object:

Rule.add(holder, {
    "file_kind": "RESULT",
    "filename": "*_result.json",
    "handler": handle_result,
    "decision_on_success": "CONTINUE",
})

class ddp_utils.file_scanner.Template(directory: str | PathLike)

Bases: object

Generate scanner JSON and load validated watch specifications.

Handler strings may name a method on the issuer or use "object.path:method" to resolve a nested handler.

Examples

Prepare a template directory:

template = Template("configs")

Create the default template directory when necessary.

Parameters

Name	Type	Description
directory	Union[str, os.PathLike]	Default directory used by generate() and load(). It is created recursively.

Examples

Use a project-local configuration directory:

template = Template("lib/configs")

generate(filename: str | None = None, output_path: str | PathLike | None = None, extension: bool = False) → bool

Write the built-in example scanner configuration as JSON.

Parameters

Name	Type	Description
filename	str | None	Output filename. Defaults to file_scanner_instructions_template.json.
output_path	str | PathLike | None	Override for the directory supplied at construction.
extension	bool	Accepted for API compatibility; the current generated document is the same for either value.

Returns

Type	Description
bool	True after a successful write; False when any write or serialization operation fails.

Examples

Generate the default example file:

created = Template("configs").generate()

load(issuer, file: str, guid: str, ext_path: str | None = None, ext_obj: bool = False) → bool | dict | tuple

Load JSON, validate it, and append its watches to an issuer.

Parameters

Name	Type	Description
issuer		Handler owner. watch_specs is created or extended; optional arti_files aliases are rewritten to resolved names.
file	str	Configuration filename relative to the template directory.
guid	str	Value substituted for {GUID} before glob normalization.
ext_path	str | None	Optional path for the sanitized configuration. A directory receives extension_rules.json.
ext_obj	bool	Include the sanitized configuration in the return value.

Returns

Type	Description
bool | dict | tuple	True normally, the sanitized dictionary when only ext_obj is enabled, or (True, sanitized) when both output options are enabled.

Raises

Exception	Description
ValueError	Reading, validation, handler resolution, alias mapping, or sanitized-output writing fails.

Examples

Load rules for one execution GUID:

public_rules = template.load(
    issuer,
    "scanner.json",
    guid="run-123",
    ext_obj=True,
)

class ddp_utils.file_scanner.FileEvent(type: str, watch: str, file_kind: str, filename: str, file_path: str, fired_at: float, mtime: float, size: int, stable_checks: int, stable_delay: float, remove_after_handled: bool, decision_on_success: ScanDecision, scan_started_at: float, scan_uptime_s: float)

Bases: object

Describe one stable file delivered to a handler.

Variables

Name	Type	Description
type	str	Event type, currently "file".
watch	str	Exact name or glob pattern that matched.
file_kind	str	Diagnostic category from the watch.
filename	str	Basename of the matched file.
file_path	str	Absolute matched path.
fired_at	float	Detection timestamp after stabilization.
mtime	float	File modification timestamp.
size	int	File size in bytes.
stable_checks	int	Required unchanged observations.
stable_delay	float	Delay between stability observations.
remove_after_handled	bool	Current deletion policy.
decision_on_success	ScanDecision	Current fallback decision.
scan_started_at	float	Scanner start timestamp.
scan_uptime_s	float	Scanner age when the event fired.

Examples

Read the stable path in a handler:

def handle(event):
    print(event.file_path, event.size)

class ddp_utils.file_scanner.HandlerReply(ok: bool = True, decision: ScanDecision | None = None, remove_after_handled: bool | None = None, message: str = '', extra: Dict[str, ~typing.Any]=<factory>)

Bases: object

Return structured success and per-event policy overrides from a handler.

Variables

Name	Type	Description
ok	bool	Treat the event as successful when True; otherwise apply the scanner’s handler-error policy.
decision	ScanDecision | None	Override the watch decision, or None to use its default.
remove_after_handled	bool | None	Override deletion, or None to use the watch.
message	str	Optional diagnostic text.
extra	Dict[str, Any]	Additional diagnostic or metric fields.

Examples

Keep a parsed file and continue polling:

return HandlerReply(
    decision=ScanDecision.CONTINUE,
    remove_after_handled=False,
)

class ddp_utils.file_scanner.FatalEvent(reason: StopReason, message: str, when_ts: float, base_dir: str, filename: str | None = None, file_path: str | None = None, file_kind: str | None = None, exception_type: str | None = None, exception_message: str | None = None, exception_traceback: str | None = None, extra: Dict[str, ~typing.Any]=<factory>)

Bases: object

Carry complete context to the scanner’s fatal-event callback.

Variables

Name	Type	Description
reason	StopReason	Structured terminal reason.
message	str	Human-readable explanation.
when_ts	float	Event timestamp.
base_dir	str	Watched directory.
filename	str | None	Related basename, when applicable.
file_path	str | None	Related full path, when applicable.
file_kind	str | None	Related watch category, when applicable.
exception_type	str | None	Exception class name, when applicable.
exception_message	str | None	Exception text, when applicable.
exception_traceback	str | None	Formatted traceback, when applicable.
extra	Dict[str, Any]	Additional scanner context.

Examples

Persist fatal diagnostics from a callback:

def on_fatal(event):
    print(event.reason, event.exception_message)

class ddp_utils.file_scanner.AsyncScan(scanner: FileScanner, kwargs: dict, daemon: bool = False, wait_timeout: float | None = None)

Bases: object

Control one FileScanner running on a background thread.

Examples

Start and wait through the scanner’s public factory:

task = scanner.run_async(wait_timeout=30)
result = task.wait()

Configure a background scan without starting it.

Parameters

Name	Type	Description
scanner	FileScanner	Scanner whose FileScanner.run() method executes.
kwargs	dict	Keyword arguments forwarded to scanner.run.
daemon	bool	Create a daemon thread when True.
wait_timeout	Optional[float]	Default timeout used by wait().

Examples

Prefer FileScanner.run_async() for construction and start:

task = scanner.run_async(wait_timeout=30)

start() → None

Start the configured scan on a new background thread.

Raises

Exception	Description
RuntimeError	This controller has already been started.

Examples

Start a manually constructed controller once:

task.start()

wait(timeout: float | None = None) → ScanResult

Wait for the background scanner and return its terminal result.

Parameters

Name	Type	Description
timeout	float | None	Maximum seconds to wait. None uses the configured default; when both are None, waiting is unbounded.

Returns

Type	Description
ScanResult	Result produced by FileScanner.run().

Raises

Exception	Description
TimeoutError	The scan does not finish within the effective limit.

Examples

Wait at most ten seconds:

result = task.wait(timeout=10)

stop() → None

Request cooperative termination of the underlying scanner.

Examples

Cancel a background scan:

task.stop()

is_alive() → bool

Return whether the background scan thread is running.

Returns

Type	Description
bool	True only after start and before thread termination.

Examples

Avoid starting conflicting project work:

if task.is_alive():
    print("scan in progress")

class ddp_utils.file_scanner.FileScanner(base_dir: str, *, poll_min: float = 0.25, poll_max: float = 1.0, stable_checks: int = 2, stable_delay: float = 0.15, delete_delay: float = 0.8, on_error: Callable[[str, Exception], None] | None = None, on_info: Callable[[str], None] | None = None, on_fatal: Callable[[FatalEvent], None] | None = None, default_max_retries: int = 0, default_retry_delay: float = 2.0, show_info: bool = False, debug: bool = False)

Bases: object

Poll a directory and dispatch stable matching files to handlers.

run() converts timeouts, interrupts, health failures, and configured fatal errors into ScanResult values instead of propagating them. Callers may map USER_INTERRUPT to process exit code 130 when desired.

Examples

Create a low-latency scanner:

scanner = FileScanner("downloads", poll_min=0.1, poll_max=1.0)

Configure polling, stability, callbacks, and retry defaults.

Parameters

Name	Type	Description
base_dir	str	Directory containing signal files.
poll_min	float	Minimum delay between polling cycles.
poll_max	float	Maximum adaptive polling delay.
stable_checks	int	Consecutive unchanged size/mtime observations needed before dispatch.
stable_delay	float	Delay between stability observations.
delete_delay	float	Delay before deleting a handled file.
on_error	Optional[Callable[[str, Exception], None]]	Callback receiving context text and a recoverable error.
on_info	Optional[Callable[[str], None]]	Callback receiving informational messages.
on_fatal	Optional[Callable[[FatalEvent], None]]	Callback receiving FatalEvent.
default_max_retries	int	Default retry limit; zero means unlimited.
default_retry_delay	float	Default seconds between retries.
show_info	bool	Enable informational callback delivery.
debug	bool	Enable detailed diagnostic callback delivery.

Examples

Capture recoverable failures:

scanner = FileScanner(
    "downloads",
    on_error=lambda context, error: print(context, error),
)

add_watch(spec: WatchSpec) → None

Add a watch or replace the watch with the same filename key.

Parameters

Name	Type	Description
spec	WatchSpec	Watch specification keyed by spec.filename.

Examples

Register a completion signal:

scanner.add_watch(WatchSpec("done.json", handle_done))

update_watches(specs: Dict[str, WatchSpec]) → None

Replace every active watch with a supplied mapping.

Parameters

Name	Type	Description
specs	Dict[str, WatchSpec]	Mapping from watch keys to specifications. A shallow copy is stored and existing error counters for matching keys persist.

Examples

Install a prepared watch set:

scanner.update_watches({spec.filename: spec})

stop() → None

Request cooperative termination of the polling loop.

Examples

Stop a scanner from another thread:

scanner.stop()

run(*, timeout_seconds: int | None = None, idle_timeout_seconds: int | None = None, init_deadline_seconds: int | None = None, init_file: str | None = None, on_init_timeout: Callable[[], None] | None = None, on_timeout: Callable[[], None] | None = None, on_idle_timeout: Callable[[], None] | None = None, stage_idle_timeouts: Mapping[str, int] | None = None, default_max_retries: int | None = None, default_retry_delay: float | None = None, health_check: Callable[[], bool] | None = None, health_check_interval: float = 5.0) → ScanResult

Run the blocking polling loop until a terminal condition occurs.

Parameters

Name	Type	Description
timeout_seconds	int | None	Absolute lifetime limit, regardless of activity.
idle_timeout_seconds	int | None	Maximum time without any matching file. Detection resets this timer before stability is confirmed.
init_deadline_seconds	int | None	Time allowed for init_file to appear.
init_file	str | None	Relative startup-signal filename.
on_init_timeout	Callable[[], None] | None	Callback for initialization timeout.
on_timeout	Callable[[], None] | None	Callback for global timeout.
on_idle_timeout	Callable[[], None] | None	Callback for idle timeout.
stage_idle_timeouts	Mapping[str, int] | None	Fallback mapping of file kind to idle timeout.
default_max_retries	int | None	Per-run retry-limit override; zero is unlimited.
default_retry_delay	float | None	Per-run retry-delay override in seconds.
health_check	Callable[[], bool] | None	Callable returning whether an external dependency is healthy. False produces HEALTH_CHECK_FAILED.
health_check_interval	float	Seconds between health checks.

Returns

Type	Description
ScanResult	Structured terminal result. The loop handles KeyboardInterrupt and runtime failures internally.

Examples

Wait up to one minute for a completion signal:

result = scanner.run(timeout_seconds=60)

run_async(daemon: bool = False, poll_max: float | None = None, wait_timeout: float | None = None, **kwargs) → AsyncScan

Start one background scan and return its controller.

Parameters

Name	Type	Description
daemon	bool	Run on a daemon thread.
poll_max	float | None	Positive temporary maximum poll delay. The original value is restored after completion.
wait_timeout	float | None	Default timeout for AsyncScan.wait().
**kwargs		Arguments forwarded to run().

Returns

Type	Description
AsyncScan	Started asynchronous scan controller.

Raises

Exception	Description
RuntimeError	Another background scan is still running.

Examples

Start a background scan with a bounded wait:

task = scanner.run_async(
    wait_timeout=30,
    timeout_seconds=120,
)

static example()

Print a complete configuration-and-scanning usage example.

Examples

Display the sample in an interactive session:

FileScanner.example()