aind_behavior_experiment_launcher.data_transfer.watchdog_service.WatchdogDataTransferService

class aind_behavior_experiment_launcher.data_transfer.watchdog_service.WatchdogDataTransferService(destination: PathLike, schedule_time: time | None = datetime.time(20, 0), project_name: str | None = None, platform: Platform = BEHAVIOR(name='Behavior platform', abbreviation='behavior'), capsule_id: str | None = None, script: Dict[str, List[str]] | None = None, s3_bucket: BucketType = BucketType.PRIVATE, mount: str | None = None, force_cloud_sync: bool = True, transfer_endpoint: str = 'http://aind-data-transfer-service/api/v1/submit_jobs', validate: bool = True)

Bases: DataTransferService

__init__(destination: PathLike, schedule_time: time | None = datetime.time(20, 0), project_name: str | None = None, platform: Platform = BEHAVIOR(name='Behavior platform', abbreviation='behavior'), capsule_id: str | None = None, script: Dict[str, List[str]] | None = None, s3_bucket: BucketType = BucketType.PRIVATE, mount: str | None = None, force_cloud_sync: bool = True, transfer_endpoint: str = 'http://aind-data-transfer-service/api/v1/submit_jobs', validate: bool = True) → None

Methods

__init__(destination[, schedule_time, ...])
create_manifest_config(source, ads_session)	Create a ManifestConfig object
create_manifest_from_ads_session(...)
create_watchdog_config(watched_directory, ...)	Create a WatchConfig object
dump_manifest_config(manifest_config[, ...])
force_restart([kill_if_running])
is_running()
is_valid_project_name()
transfer(*args, **kwargs)
validate([create_config])

Attributes

DEFAULT_CONFIG
DEFAULT_EXE

create_manifest_config(source: PathLike, ads_session: Session, ads_schemas: List[PathLike] | None = None, session_name: str | None = None, **kwargs) → ManifestConfig

Create a ManifestConfig object

static create_watchdog_config(watched_directory: PathLike, manifest_complete_directory: PathLike, webhook_url: str | None = None, create_dir: bool = True) → WatchConfig

Create a WatchConfig object