Skip to content

topmark.pipeline.steps.prober

topmark / pipeline / steps / prober

Probe file type and processor resolution.

Runs the shared resolution probe and applies the same effective resolution mapping as ResolverStep, then halts after successful probing because the probe pipeline is intentionally resolution-only.

ProberStep 

ProberStep()

Bases: BaseStep

Run resolution probing and stop the probe pipeline.

This step is intended for the probe pipeline. It records the full resolution explanation on ctx.resolution_probe, mirrors the effective resolution outcome onto the normal resolve axis, and halts after successful resolution.

Axes written
  • resolve
Sets
  • ctx.resolution_probe
  • ctx.file_type
  • ctx.header_processor
  • ctx.status.resolve
Source code in src/topmark/pipeline/steps/prober.py 
48
49
50
51
52
53
def __init__(self) -> None:
    super().__init__(
        name=self.__class__.__name__,
        primary_axis=Axis.RESOLVE,
        axes_written=(Axis.RESOLVE,),
    )

may_proceed 

may_proceed(ctx)

Return True because probing is the first and only probe step.

Parameters:

Name Type Description Default
ctx ProcessingContext

The processing context for the current file.

 required

Returns:

Type Description
bool

True.
Source code in src/topmark/pipeline/steps/prober.py 
55
56
57
58
59
60
61
62
63
64
def may_proceed(self, ctx: ProcessingContext) -> bool:
    """Return True because probing is the first and only probe step.

    Args:
        ctx: The processing context for the current file.

    Returns:
        True.
    """
    return True

run 

run(ctx)

Resolve and store probe-visible diagnostic details.

Parameters:

Name Type Description Default
ctx ProcessingContext

Processing context representing the file being probed.

 required
Source code in src/topmark/pipeline/steps/prober.py 
66
67
68
69
70
71
72
73
74
def run(self, ctx: ProcessingContext) -> None:
    """Resolve and store probe-visible diagnostic details.

    Args:
        ctx: Processing context representing the file being probed.
    """
    apply_probe_resolution_to_context(ctx=ctx, step=self)
    if ctx.status.resolve == ResolveStatus.RESOLVED:
        ctx.request_halt(reason="Resolution probe completed.", at_step=self)

hint 

hint(ctx)

Advise about probe resolution outcome.

Parameters:

Name Type Description Default
ctx ProcessingContext

The processing context.

 required
Source code in src/topmark/pipeline/steps/prober.py 
76
77
78
79
80
81
82
def hint(self, ctx: ProcessingContext) -> None:
    """Advise about probe resolution outcome.

    Args:
        ctx: The processing context.
    """
    return