Module: sarpyx.snapflow.snap2stamps

File: sarpyx/snapflow/snap2stamps.py

Processing-only SNAP2StaMPS workflows implemented on top of :mod:`snapflow`. This module covers the upstream SNAP2StaMPS processing branches without recreating the original script/config runner UX. The focus is: - processing branch selection for TOPSAR and Stripmap data - reusable Python-callable helpers for the graph families used upstream - backward-compatible access to the narrower workflow helpers that previously lived in ``snap2stamps_pipelines.py``

Exported Symbols (__all__)

• PAIRWISE_GRAPH_NAMES
• PairProducts
• PipelineDefinition
• PipelineStep
• SNAP2STAMPS_GRAPH_PIPELINES
• SNAP2STAMPS_PIPELINES
• SNAP2STAMPS_WORKFLOWS
• SNAP2STAMPS_WORKFLOW_INPUTS
• TopsarCoregIfgResult
• build_gpt
• get_pipeline_definition
• list_pipeline_names
• pipeline_requires_multi_input
• pipeline_requires_pair
• prepare_pair
• run_graph_pipeline
• run_pair_graph_pipeline
• run_pair_workflow
• run_processing_pipeline
• run_stripmap_coreg
• run_stripmap_export
• run_stripmap_ifg
• run_stripmap_subset
• run_topsar_coreg_ifg
• run_topsar_export
• run_topsar_split_apply_orbit
• run_workflow
• select_stripmap_coreg_pipeline
• select_stripmap_ifg_pipeline
• select_topsar_coreg_ifg_pipeline
• select_topsar_export_pipeline
• select_topsar_split_pipeline

Public Functions (23)

list_pipeline_names(branch: 'BranchName | None' = None) -> 'tuple[str, ...]' -> tuple[str, ...]

from sarpyx.snapflow.snap2stamps import list_pipeline_names

result = list_pipeline_names()

get_pipeline_definition(name: 'str') -> 'PipelineDefinition' -> PipelineDefinition

from sarpyx.snapflow.snap2stamps import get_pipeline_definition

result = get_pipeline_definition(name=<name>)

pipeline_requires_pair(name: 'str') -> 'bool' -> bool

from sarpyx.snapflow.snap2stamps import pipeline_requires_pair

result = pipeline_requires_pair(name=<name>)

pipeline_requires_multi_input(name: 'str') -> 'bool' -> bool

from sarpyx.snapflow.snap2stamps import pipeline_requires_multi_input

result = pipeline_requires_multi_input(name=<name>)

select_topsar_split_pipeline(source_count: 'int') -> 'str' -> str

from sarpyx.snapflow.snap2stamps import select_topsar_split_pipeline

result = select_topsar_split_pipeline(source_count=<source_count>)

select_topsar_coreg_ifg_pipeline(*, master_count: 'int', burst_count: 'int', external_dem_file: 'str | Path | None' = None) -> 'str' -> str

from sarpyx.snapflow.snap2stamps import select_topsar_coreg_ifg_pipeline

result = select_topsar_coreg_ifg_pipeline(master_count=<master_count>, burst_count=<burst_count>)

select_topsar_export_pipeline(*, master_count: 'int', external_dem_file: 'str | Path | None' = None) -> 'str' -> str

from sarpyx.snapflow.snap2stamps import select_topsar_export_pipeline

result = select_topsar_export_pipeline(master_count=<master_count>)

select_stripmap_coreg_pipeline(external_dem_file: 'str | Path | None' = None) -> 'str' -> str

from sarpyx.snapflow.snap2stamps import select_stripmap_coreg_pipeline

result = select_stripmap_coreg_pipeline()

select_stripmap_ifg_pipeline(external_dem_file: 'str | Path | None' = None) -> 'str' -> str

from sarpyx.snapflow.snap2stamps import select_stripmap_ifg_pipeline

result = select_stripmap_ifg_pipeline()

build_gpt(product: 'str | Path', outdir: 'str | Path', *, format: 'str' = 'BEAM-DIMAP', gpt_path: 'str' = '/usr/local/snap/bin/gpt', memory: 'str | None' = None, parallelism: 'int | None' = 14, timeout: 'int | None' = 7200, snap_userdir: 'str | Path | None' = None) -> 'GPT' -> GPT

from sarpyx.snapflow.snap2stamps import build_gpt

result = build_gpt(product=<product>, outdir=<outdir>)

run_graph_pipeline(gpt: 'GPT', graph_name: 'str', overrides: 'dict[str, dict[str, Any]] | None' = None) -> 'str' -> str

from sarpyx.snapflow.snap2stamps import run_graph_pipeline

result = run_graph_pipeline(gpt=<gpt>, graph_name=<graph_name>)

run_pair_graph_pipeline(gpt: 'GPT', graph_name: 'str', pair: 'PairProducts', overrides: 'dict[str, dict[str, Any]] | None' = None) -> 'str' -> str

from sarpyx.snapflow.snap2stamps import run_pair_graph_pipeline

result = run_pair_graph_pipeline(gpt=<gpt>, graph_name=<graph_name>, pair=<pair>)

prepare_pair(pair: 'PairProducts', outdir: 'str | Path', *, preprocess_graphs: 'tuple[str, ...]' = ('split_orbit',), format: 'str' = 'BEAM-DIMAP', gpt_path: 'str' = '/usr/local/snap/bin/gpt', memory: 'str | None' = None, parallelism: 'int | None' = 14, timeout: 'int | None' = 7200, snap_userdir: 'str | Path | None' = None, overrides: 'dict[str, dict[str, Any]] | None' = None) -> 'PairProducts' -> PairProducts

from sarpyx.snapflow.snap2stamps import prepare_pair

result = prepare_pair(pair=<pair>, outdir=<outdir>)

run_pair_workflow(pair: 'PairProducts', outdir: 'str | Path', workflow: 'str' = 'stamps_prep', *, format: 'str' = 'BEAM-DIMAP', gpt_path: 'str' = '/usr/local/snap/bin/gpt', memory: 'str | None' = None, parallelism: 'int | None' = 14, timeout: 'int | None' = 7200, snap_userdir: 'str | Path | None' = None, overrides: 'dict[str, dict[str, Any]] | None' = None) -> 'str' -> str

from sarpyx.snapflow.snap2stamps import run_pair_workflow

result = run_pair_workflow(pair=<pair>, outdir=<outdir>)

run_workflow(gpt: 'GPT', workflow: 'str', overrides: 'dict[str, dict[str, Any]] | None' = None) -> 'str' -> str

from sarpyx.snapflow.snap2stamps import run_workflow

result = run_workflow(gpt=<gpt>, workflow=<workflow>)

run_topsar_split_apply_orbit(source_products: 'Iterable[str | Path]', outdir: 'str | Path', *, subswath: 'str', polarisation: 'str | None' = None, polygon_wkt: 'str | None' = None, output_name: 'str | None' = None, format: 'str' = 'BEAM-DIMAP', gpt_path: 'str' = '/usr/local/snap/bin/gpt', memory: 'str | None' = None, parallelism: 'int | None' = 14, timeout: 'int | None' = 7200, snap_userdir: 'str | Path | None' = None, apply_orbit_kwargs: 'dict[str, Any] | None' = None) -> 'str' -> str

from sarpyx.snapflow.snap2stamps import run_topsar_split_apply_orbit

result = run_topsar_split_apply_orbit(source_products=<source_products>, outdir=<outdir>, subswath=<subswath>)

run_processing_pipeline(pipeline_name: 'str', *, outdir: 'str | Path', format: 'str' = 'BEAM-DIMAP', gpt_path: 'str' = '/usr/local/snap/bin/gpt', memory: 'str | None' = None, parallelism: 'int | None' = 14, timeout: 'int | None' = 7200, snap_userdir: 'str | Path | None' = None, source_product: 'str | Path | None' = None, source_products: 'Iterable[str | Path] | None' = None, pair: 'PairProducts | None' = None, target_folder: 'str | Path | None' = None, **kwargs: 'Any') -> 'str | TopsarCoregIfgResult' -> str | TopsarCoregIfgResult

from sarpyx.snapflow.snap2stamps import run_processing_pipeline

result = run_processing_pipeline(pipeline_name=<pipeline_name>, outdir=<outdir>, kwargs=<kwargs>)

run_topsar_coreg_ifg(pair: 'PairProducts', outdir: 'str | Path', *, master_count: 'int | None' = None, burst_count: 'int | None' = None, pipeline_name: 'str | None' = None, polygon_wkt: 'str | None' = None, subset_region: 'str' = '0,0,0,0', external_dem_file: 'str | Path | None' = None, format: 'str' = 'BEAM-DIMAP', gpt_path: 'str' = '/usr/local/snap/bin/gpt', memory: 'str | None' = None, parallelism: 'int | None' = 14, timeout: 'int | None' = 7200, snap_userdir: 'str | Path | None' = None, output_name_prefix: 'str' = 'pair') -> 'TopsarCoregIfgResult' -> TopsarCoregIfgResult

from sarpyx.snapflow.snap2stamps import run_topsar_coreg_ifg

result = run_topsar_coreg_ifg(pair=<pair>, outdir=<outdir>)

run_topsar_export(coreg_products: 'Iterable[str | Path]', ifg_products: 'Iterable[str | Path]', outdir: 'str | Path', *, target_folder: 'str | Path', master_count: 'int', polygon_wkt: 'str | None' = None, subset_region: 'str' = '0,0,0,0', external_dem_file: 'str | Path | None' = None, selected_polarisations: 'list[str] | None' = None, format: 'str' = 'BEAM-DIMAP', gpt_path: 'str' = '/usr/local/snap/bin/gpt', memory: 'str | None' = None, parallelism: 'int | None' = 14, timeout: 'int | None' = 7200, snap_userdir: 'str | Path | None' = None) -> 'str' -> str

from sarpyx.snapflow.snap2stamps import run_topsar_export

result = run_topsar_export(coreg_products=<coreg_products>, ifg_products=<ifg_products>, outdir=<outdir>, target_folder=<target_folder>, master_count=<master_count>)

run_stripmap_subset(product: 'str | Path', outdir: 'str | Path', *, region: 'str | None' = None, polygon_wkt: 'str | None' = None, output_name: 'str | None' = None, format: 'str' = 'BEAM-DIMAP', gpt_path: 'str' = '/usr/local/snap/bin/gpt', memory: 'str | None' = None, parallelism: 'int | None' = 14, timeout: 'int | None' = 7200, snap_userdir: 'str | Path | None' = None) -> 'str' -> str

from sarpyx.snapflow.snap2stamps import run_stripmap_subset

result = run_stripmap_subset(product=<product>, outdir=<outdir>)

run_stripmap_coreg(pair: 'PairProducts', outdir: 'str | Path', *, external_dem_file: 'str | Path | None' = None, output_name: 'str | None' = None, format: 'str' = 'BEAM-DIMAP', gpt_path: 'str' = '/usr/local/snap/bin/gpt', memory: 'str | None' = None, parallelism: 'int | None' = 14, timeout: 'int | None' = 7200, snap_userdir: 'str | Path | None' = None) -> 'str' -> str

from sarpyx.snapflow.snap2stamps import run_stripmap_coreg

result = run_stripmap_coreg(pair=<pair>, outdir=<outdir>)

run_stripmap_ifg(coreg_product: 'str | Path', outdir: 'str | Path', *, external_dem_file: 'str | Path | None' = None, output_name_prefix: 'str' = 'stripmap', format: 'str' = 'BEAM-DIMAP', gpt_path: 'str' = '/usr/local/snap/bin/gpt', memory: 'str | None' = None, parallelism: 'int | None' = 14, timeout: 'int | None' = 7200, snap_userdir: 'str | Path | None' = None) -> 'str' -> str

from sarpyx.snapflow.snap2stamps import run_stripmap_ifg

result = run_stripmap_ifg(coreg_product=<coreg_product>, outdir=<outdir>)

run_stripmap_export(coreg_product: 'str | Path', ifg_product: 'str | Path', outdir: 'str | Path', *, target_folder: 'str | Path', format: 'str' = 'BEAM-DIMAP', gpt_path: 'str' = '/usr/local/snap/bin/gpt', memory: 'str | None' = None, parallelism: 'int | None' = 14, timeout: 'int | None' = 7200, snap_userdir: 'str | Path | None' = None) -> 'str' -> str

from sarpyx.snapflow.snap2stamps import run_stripmap_export

result = run_stripmap_export(coreg_product=<coreg_product>, ifg_product=<ifg_product>, outdir=<outdir>, target_folder=<target_folder>)

Public Classes (4)

class PipelineStep(method: 'str', kwargs: 'dict[str, Any]' = <factory>) -> None

from sarpyx.snapflow.snap2stamps import PipelineStep

obj = PipelineStep(...)  # inferred from implementation

class PairProducts(master: 'str | Path', slave: 'str | Path') -> None

from sarpyx.snapflow.snap2stamps import PairProducts

obj = PairProducts(...)  # inferred from implementation

class PipelineDefinition(name: 'str', branch: 'BranchName', input_kind: 'InputKind', stages: 'tuple[str, ...]', upstream_graph: 'str | None' = None, description: 'str' = '') -> None

from sarpyx.snapflow.snap2stamps import PipelineDefinition

obj = PipelineDefinition(...)  # inferred from implementation

class TopsarCoregIfgResult(coreg_path: 'str', ifg_path: 'str', pipeline_name: 'str') -> None

from sarpyx.snapflow.snap2stamps import TopsarCoregIfgResult

obj = TopsarCoregIfgResult(...)  # inferred from implementation