cursus.steps.scripts.temporal_sequence_normalization

Temporal Sequence Normalization Script

This script normalizes temporal sequence data for machine learning models, providing configurable operations for sequence ordering, validation, missing value handling, time delta computation, and sequence padding/truncation.

Supports multiple data formats (CSV, TSV, JSON, Parquet) and provides extensive configurability through environment variables.

load_signature_columns(signature_path)

Load column names from signature file.

peek_json_format(file_path, open_func=<built-in function open>)

Check if the JSON file is in JSON Lines or regular format.

combine_shards(input_dir, signature_columns=None)

Detect and combine all supported data shards in a directory.

class SequenceOrderingOperation(temporal_field, id_field, logger=None)

Bases: object

Handles temporal ordering of sequences.

process(df)

Sort sequences by temporal field and handle duplicates.

class DataValidationOperation(validation_strategy, temporal_field, id_field, missing_indicators, logger=None)

Bases: object

Validates sequence data integrity.

process(df, sequence_fields)

Validate sequence data based on strategy.

class MissingValueHandlingOperation(missing_indicators, logger=None)

Bases: object

Handles missing values in sequences.

process(df, sequence_fields)

Handle missing values in sequence fields.

class TimeDeltaComputationOperation(temporal_field, max_seconds, logger=None)

Bases: object

Computes time deltas for temporal sequences.

process(sequence_data)

Compute time deltas for numerical sequences.

class SequencePaddingOperation(target_length, padding_strategy, truncation_strategy, include_attention_masks, logger=None)

Bases: object

Handles sequence padding and truncation.

process(sequence_data)

Pad or truncate sequences to target length.

detect_sequence_fields(df, sequence_separator, entity_id_field, secondary_entity_field, sequence_naming_pattern, enable_multi_sequence, temporal_field, missing_indicators)

Automatically detect sequence fields based on naming patterns.

parse_sequence_data(df, sequence_fields, sequence_separator, missing_indicators, logger=None)

Parse sequence data from DataFrame into numpy arrays.

save_normalized_sequences(sequence_data, output_dir, output_format, sequence_length, sequence_separator, temporal_field, entity_id_field, id_field, include_attention_masks, logger=None)

Save normalized sequences in the specified format.

main(input_paths, output_paths, environ_vars, job_args, logger=None)

Main logic for temporal sequence normalization.

Parameters:

• input_paths (Dict[str, str]) – Dictionary of input paths with logical names

• output_paths (Dict[str, str]) – Dictionary of output paths with logical names

• environ_vars (Dict[str, str]) – Dictionary of environment variables

• job_args (Namespace) – Command line arguments

• logger (Callable[[str], None] | None) – Optional logger object (defaults to print if None)

Returns:

Dictionary of normalized sequence arrays

Return type:

Dict[str, ndarray]