Implement Pod activation command flow

omnipod-dash/src/main/java/info/nightscout/androidaps/plugins/pump/omnipod/dash/driver/OmnipodDashManager.kt

@@ -2,6 +2,7 @@ package info.nightscout.androidaps.plugins.pump.omnipod.dash.driver
 
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.event.PodEvent
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.definition.AlertConfiguration
+import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.definition.AlertTrigger
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.definition.AlertType
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.definition.BasalProgram
 import io.reactivex.Observable
@@ -9,9 +10,9 @@ import java.util.*
 
 interface OmnipodDashManager {
 
-    fun activatePodPart1(): Observable<PodEvent>
+    fun activatePodPart1(lowReservoirAlertTrigger: AlertTrigger.ReservoirVolumeTrigger?): Observable<PodEvent>
 
-    fun activatePodPart2(): Observable<PodEvent>
+    fun activatePodPart2(basalProgram: BasalProgram): Observable<PodEvent>
 
     fun getStatus(): Observable<PodEvent>
 

omnipod-dash/src/main/java/info/nightscout/androidaps/plugins/pump/omnipod/dash/driver/OmnipodDashManagerImpl.kt

@@ -4,21 +4,15 @@ import info.nightscout.androidaps.logging.AAPSLogger
 import info.nightscout.androidaps.logging.LTag
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.comm.OmnipodDashBleManager
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.event.PodEvent
-import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.command.GetVersionCommand
+import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.command.*
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.command.GetVersionCommand.Companion.DEFAULT_UNIQUE_ID
-import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.definition.ActivationProgress
+import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.definition.*
-import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.definition.AlertConfiguration
+import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.response.*
-import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.definition.AlertType
-import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.definition.BasalProgram
-import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.response.AlarmStatusResponse
-import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.response.DefaultStatusResponse
-import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.response.Response
-import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.response.SetUniqueIdResponse
-import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.response.VersionResponse
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.state.OmnipodDashPodStateManager
 import info.nightscout.androidaps.utils.rx.AapsSchedulers
 import info.nightscout.androidaps.utils.rx.retryWithBackoff
 import io.reactivex.Observable
+import io.reactivex.functions.Action
 import io.reactivex.functions.Consumer
 import java.util.*
 import java.util.concurrent.TimeUnit
@@ -42,9 +36,109 @@ class OmnipodDashManagerImpl @Inject constructor(
             }
         }
 
+    private val observePodReadyForActivationPart2: Observable<PodEvent>
+        get() = Observable.defer {
+            if (podStateManager.activationProgress.isAtLeast(ActivationProgress.PHASE_1_COMPLETED) && podStateManager.activationProgress.isBefore(ActivationProgress.COMPLETED)) {
+                Observable.empty()
+            } else {
+                Observable.error(IllegalStateException("Pod is in an incorrect state"))
+            }
+        }
+
     private val observeConnectToPod: Observable<PodEvent>
         get() = Observable.defer { bleManager.connect().retryWithBackoff(retries = 2, delay = 3, timeUnit = TimeUnit.SECONDS) } // TODO are these reasonable values?
 
+    private fun observeSendProgramBolusCommand(units: Double, rateInEighthPulsesPerSeconds: Byte, confirmationBeeps: Boolean, completionBeeps: Boolean): Observable<PodEvent> {
+        return Observable.defer {
+            bleManager.sendCommand(ProgramBolusCommand.Builder()
+                .setUniqueId(podStateManager.uniqueId!!.toInt())
+                .setSequenceNumber(podStateManager.messageSequenceNumber)
+                .setNonce(1229869870) // TODO
+                .setNumberOfUnits(units)
+                .setDelayBetweenPulsesInEighthSeconds(rateInEighthPulsesPerSeconds)
+                .setProgramReminder(ProgramReminder(confirmationBeeps, completionBeeps, 0))
+                .build()
+            )
+        }
+    }
+
+    private fun observeSendGetPodStatusCommand(type: ResponseType.StatusResponseType = ResponseType.StatusResponseType.DEFAULT_STATUS_RESPONSE): Observable<PodEvent> {
+        return Observable.defer {
+            bleManager.sendCommand(
+                GetStatusCommand.Builder()
+                    .setUniqueId(podStateManager.uniqueId!!.toInt())
+                    .setSequenceNumber(podStateManager.messageSequenceNumber)
+                    .setStatusResponseType(type)
+                    .build()
+            )
+        }
+    }
+
+    private val observeVerifyCannulaInsertion: Observable<PodEvent>
+        get() = Observable.defer {
+            observeSendGetPodStatusCommand()
+                .ignoreElements() //
+                .andThen(Observable.defer {
+                    if (podStateManager.podStatus == PodStatus.RUNNING_ABOVE_MIN_VOLUME) {
+                        Observable.empty()
+                    } else {
+                        Observable.error(IllegalStateException("Unexpected Pod status"))
+                    }
+                })
+        }
+
+    private fun observeSendProgramAlertsCommand(alertConfigurations: List<AlertConfiguration>, multiCommandFlag: Boolean = false): Observable<PodEvent> {
+        return Observable.defer {
+            bleManager.sendCommand(
+                ProgramAlertsCommand.Builder()
+                    .setUniqueId(podStateManager.uniqueId!!.toInt())
+                    .setSequenceNumber(podStateManager.messageSequenceNumber)
+                    .setNonce(1229869870) // TODO
+                    .setAlertConfigurations(alertConfigurations)
+                    .build()
+            )
+        }
+    }
+
+    private fun observeProgramBasalCommand(basalProgram: BasalProgram): Observable<PodEvent> {
+        return Observable.defer {
+            bleManager.sendCommand(
+                ProgramBasalCommand.Builder()
+                    .setUniqueId(podStateManager.uniqueId!!.toInt())
+                    .setSequenceNumber(podStateManager.messageSequenceNumber)
+                    .setNonce(1229869870) // TODO
+                    .setProgramReminder(ProgramReminder(atStart = false, atEnd = false, atInterval = 0))
+                    .setBasalProgram(basalProgram)
+                    .setCurrentTime(Date())
+                    .build()
+            )
+        }
+    }
+
+    private val observeVerifyPrime: Observable<PodEvent>
+        get() = Observable.defer {
+            observeSendGetPodStatusCommand()
+                .ignoreElements() //
+                .andThen(Observable.defer {
+                    if (podStateManager.podStatus == PodStatus.CLUTCH_DRIVE_ENGAGED) {
+                        Observable.empty()
+                    } else {
+                        Observable.error(IllegalStateException("Unexpected Pod status"))
+                    }
+                })
+        }
+
+    private val observeSendSetUniqueIdCommand: Observable<PodEvent>
+        get() = Observable.defer {
+            bleManager.sendCommand(SetUniqueIdCommand.Builder() //
+                .setSequenceNumber(podStateManager.messageSequenceNumber) //
+                .setUniqueId(podStateManager.uniqueId!!.toInt()) //
+                .setLotNumber(podStateManager.lotNumber!!.toInt()) //
+                .setPodSequenceNumber(podStateManager.podSequenceNumber!!.toInt())
+                .setInitializationTime(Date())
+                .build()) //
+        }
+
     private val observeSendGetVersionCommand: Observable<PodEvent>
         get() = Observable.defer {
             bleManager.sendCommand(GetVersionCommand.Builder() //
@@ -53,22 +147,169 @@ class OmnipodDashManagerImpl @Inject constructor(
                 .build()) //
         }
 
-    override fun activatePodPart1(): Observable<PodEvent> {
+    override fun activatePodPart1(lowReservoirAlertTrigger: AlertTrigger.ReservoirVolumeTrigger?): Observable<PodEvent> {
         return Observable.concat(
             observePodReadyForActivationPart1,
             observeConnectToPod,
-            observeSendGetVersionCommand
+            observeActivationPart1Commands(lowReservoirAlertTrigger)
-            // ... Send more commands
+        ).doOnComplete(ActivationProgressUpdater(ActivationProgress.PHASE_1_COMPLETED))
-        ) //
             // TODO these would be common for any observable returned in a public function in this class
             .doOnNext(PodEventInterceptor()) //
             .doOnError(ErrorInterceptor())
             .subscribeOn(aapsSchedulers.io)
     }
 
-    override fun activatePodPart2(): Observable<PodEvent> {
+    private fun observeActivationPart1Commands(lowReservoirAlertTrigger: AlertTrigger.ReservoirVolumeTrigger?): Observable<PodEvent> {
-        // TODO
+        val observables = createActivationPart1Observables(lowReservoirAlertTrigger)
-        return Observable.empty()
+
+        return if (observables.isEmpty()) {
+            Observable.empty()
+        } else {
+            Observable.concat(observables)
+        }
+    }
+
+    private fun createActivationPart1Observables(lowReservoirAlertTrigger: AlertTrigger.ReservoirVolumeTrigger?): List<Observable<PodEvent>> {
+        val observables = ArrayList<Observable<PodEvent>>()
+
+        if (podStateManager.activationProgress.isBefore(ActivationProgress.PRIME_COMPLETED)) {
+            observables.add(
+                observeVerifyPrime.doOnComplete(ActivationProgressUpdater(ActivationProgress.PRIME_COMPLETED))
+            )
+        }
+        if (podStateManager.activationProgress.isBefore(ActivationProgress.PRIMING)) {
+            observables.add(
+                observeSendProgramBolusCommand(
+                    podStateManager.firstPrimeBolusVolume!! * 0.05,
+                    podStateManager.primePulseRate!!.toByte(),
+                    confirmationBeeps = false,
+                    completionBeeps = false
+                ).doOnComplete(ActivationProgressUpdater(ActivationProgress.PRIMING))
+            )
+        }
+        if (podStateManager.activationProgress.isBefore(ActivationProgress.REPROGRAMMED_LUMP_OF_COAL_ALERT)) {
+            observables.add(
+                observeSendProgramAlertsCommand(
+                    listOf(
+                        AlertConfiguration(
+                            AlertType.EXPIRATION,
+                            enabled = true,
+                            durationInMinutes = 55,
+                            autoOff = false,
+                            AlertTrigger.TimerTrigger(5),
+                            BeepType.FOUR_TIMES_BIP_BEEP,
+                            BeepRepetitionType.XXX5
+                        )
+                    )
+                ).doOnComplete(ActivationProgressUpdater(ActivationProgress.REPROGRAMMED_LUMP_OF_COAL_ALERT))
+            )
+        }
+        if (lowReservoirAlertTrigger != null && podStateManager.activationProgress.isBefore(ActivationProgress.PROGRAMMED_LOW_RESERVOIR_ALERTS)) {
+            observables.add(
+                observeSendProgramAlertsCommand(
+                    listOf(
+                        AlertConfiguration(
+                            AlertType.LOW_RESERVOIR,
+                            enabled = true,
+                            durationInMinutes = 0,
+                            autoOff = false,
+                            lowReservoirAlertTrigger,
+                            BeepType.FOUR_TIMES_BIP_BEEP,
+                            BeepRepetitionType.XXX
+                        )
+                    )
+                ).doOnComplete(ActivationProgressUpdater(ActivationProgress.PROGRAMMED_LOW_RESERVOIR_ALERTS))
+            )
+        }
+        if (podStateManager.activationProgress.isBefore(ActivationProgress.SET_UNIQUE_ID)) {
+            observables.add(
+                observeSendSetUniqueIdCommand.doOnComplete(ActivationProgressUpdater(ActivationProgress.SET_UNIQUE_ID))
+            )
+        }
+        if (podStateManager.activationProgress.isBefore(ActivationProgress.GOT_POD_VERSION)) {
+            observables.add(
+                observeSendGetVersionCommand.doOnComplete(ActivationProgressUpdater(ActivationProgress.GOT_POD_VERSION))
+            )
+        }
+
+        return observables.reversed()
+    }
+
+    override fun activatePodPart2(basalProgram: BasalProgram): Observable<PodEvent> {
+        return Observable.concat(
+            observePodReadyForActivationPart2,
+            observeConnectToPod,
+            observeActivationPart2Commands(basalProgram)
+        ).doOnComplete(ActivationProgressUpdater(ActivationProgress.COMPLETED))
+            // TODO these would be common for any observable returned in a public function in this class
+            .doOnNext(PodEventInterceptor()) //
+            .doOnError(ErrorInterceptor())
+            .subscribeOn(aapsSchedulers.io)
+    }
+
+    private fun observeActivationPart2Commands(basalProgram: BasalProgram): Observable<PodEvent> {
+        val observables = createActivationPart2Observables(basalProgram)
+
+        return if (observables.isEmpty()) {
+            Observable.empty()
+        } else {
+            Observable.concat(observables)
+        }
+    }
+
+    private fun createActivationPart2Observables(basalProgram: BasalProgram): List<Observable<PodEvent>> {
+        val observables = ArrayList<Observable<PodEvent>>()
+
+        if (podStateManager.activationProgress.isBefore(ActivationProgress.CANNULA_INSERTED)) {
+            observables.add(
+                observeVerifyCannulaInsertion
+                    .doOnComplete(ActivationProgressUpdater(ActivationProgress.CANNULA_INSERTED))
+            )
+        }
+        if (podStateManager.activationProgress.isBefore(ActivationProgress.INSERTING_CANNULA)) {
+            observables.add(
+                observeSendProgramBolusCommand(
+                    podStateManager.secondPrimeBolusVolume!! * 0.05,
+                    podStateManager.primePulseRate!!.toByte(),
+                    confirmationBeeps = false,
+                    completionBeeps = false
+                ).doOnComplete(ActivationProgressUpdater(ActivationProgress.INSERTING_CANNULA))
+            )
+        }
+        if (podStateManager.activationProgress.isBefore(ActivationProgress.UPDATED_EXPIRATION_ALERTS)) {
+            observables.add(observeSendProgramAlertsCommand(
+                listOf(
+                    // FIXME use user configured expiration alert
+                    AlertConfiguration(
+                        AlertType.EXPIRATION,
+                        enabled = true,
+                        durationInMinutes = TimeUnit.HOURS.toMinutes(7).toShort(),
+                        autoOff = false,
+                        AlertTrigger.TimerTrigger(TimeUnit.HOURS.toMinutes(73).toShort()), // FIXME use activation time
+                        BeepType.FOUR_TIMES_BIP_BEEP,
+                        BeepRepetitionType.XXX3
+                    ),
+                    AlertConfiguration(
+                        AlertType.EXPIRATION_IMMINENT,
+                        enabled = true,
+                        durationInMinutes = TimeUnit.HOURS.toMinutes(1).toShort(),
+                        autoOff = false,
+                        AlertTrigger.TimerTrigger(TimeUnit.HOURS.toMinutes(79).toShort()), // FIXME use activation time
+                        BeepType.FOUR_TIMES_BIP_BEEP,
+                        BeepRepetitionType.XXX4
+                    )
+                ),
+                multiCommandFlag = true
+            ).doOnComplete(ActivationProgressUpdater(ActivationProgress.UPDATED_EXPIRATION_ALERTS)))
+        }
+        if (podStateManager.activationProgress.isBefore(ActivationProgress.PROGRAMMED_BASAL)) {
+            observables.add(
+                observeProgramBasalCommand(basalProgram)
+                    .doOnComplete(ActivationProgressUpdater(ActivationProgress.PROGRAMMED_BASAL))
+            )
+        }
+
+        return observables.reversed()
     }
 
     override fun getStatus(): Observable<PodEvent> {
@@ -149,6 +390,10 @@ class OmnipodDashManagerImpl @Inject constructor(
                     podStateManager.uniqueId = event.uniqueId
                 }
 
+                is PodEvent.CommandSent -> {
+                    podStateManager.increaseMessageSequenceNumber()
+                }
+
                 is PodEvent.ResponseReceived -> {
                     podStateManager.increaseMessageSequenceNumber()
                     handleResponse(event.response)
@@ -189,4 +434,12 @@ class OmnipodDashManagerImpl @Inject constructor(
         }
 
     }
+
+    inner class ActivationProgressUpdater(private val value: ActivationProgress) : Action {
+
+        override fun run() {
+            podStateManager.activationProgress = value
+        }
+
+    }
 }

omnipod-dash/src/main/java/info/nightscout/androidaps/plugins/pump/omnipod/dash/driver/pod/definition/ActivationProgress.kt

@@ -8,10 +8,9 @@ enum class ActivationProgress {
     REPROGRAMMED_LUMP_OF_COAL_ALERT,
     PRIMING,
     PRIME_COMPLETED,
-    PROGRAMMED_USER_SET_EXPIRATION_ALERT,
     PHASE_1_COMPLETED,
     PROGRAMMED_BASAL,
-    PROGRAMMED_CANCEL_LOC_ETC_ALERT,
+    UPDATED_EXPIRATION_ALERTS,
     INSERTING_CANNULA,
     CANNULA_INSERTED,
     COMPLETED;

omnipod-dash/src/main/java/info/nightscout/androidaps/plugins/pump/omnipod/dash/driver/pod/state/OmnipodDashPodStateManager.kt

@@ -21,7 +21,7 @@ interface OmnipodDashPodStateManager {
     val messageSequenceNumber: Short
     val sequenceNumberOfLastProgrammingCommand: Short?
     val activationTime: Long?
-    var uniqueId: Long?
+    var uniqueId: Long? // TODO make Int
     var bluetoothAddress: String?
 
     val bluetoothVersion: SoftwareVersion?

omnipod-dash/src/main/java/info/nightscout/androidaps/plugins/pump/omnipod/dash/driver/pod/state/OmnipodDashPodStateManagerImpl.kt

@@ -3,6 +3,8 @@ package info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.state
 import com.google.gson.Gson
 import info.nightscout.androidaps.logging.AAPSLogger
 import info.nightscout.androidaps.logging.LTag
+import info.nightscout.androidaps.plugins.bus.RxBusWrapper
+import info.nightscout.androidaps.plugins.pump.omnipod.dash.EventOmnipodDashPumpValuesChanged
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.R
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.definition.*
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.response.AlarmStatusResponse
@@ -18,7 +20,8 @@ import javax.inject.Singleton
 @Singleton
 class OmnipodDashPodStateManagerImpl @Inject constructor(
     private val logger: AAPSLogger,
-    private val sharedPreferences: SP
+    private val sharedPreferences: SP,
+    private val rxBus: RxBusWrapper
 ) : OmnipodDashPodStateManager {
 
     private var podState: PodState
@@ -158,6 +161,7 @@ class OmnipodDashPodStateManagerImpl @Inject constructor(
 
         podState.lastUpdated = System.currentTimeMillis()
         store()
+        rxBus.send(EventOmnipodDashPumpValuesChanged())
     }
 
     override fun updateFromVersionResponse(response: VersionResponse) {
@@ -169,6 +173,7 @@ class OmnipodDashPodStateManagerImpl @Inject constructor(
 
         podState.lastUpdated = System.currentTimeMillis()
         store()
+        rxBus.send(EventOmnipodDashPumpValuesChanged())
     }
 
     override fun updateFromSetUniqueIdResponse(response: SetUniqueIdResponse) {
@@ -186,10 +191,15 @@ class OmnipodDashPodStateManagerImpl @Inject constructor(
 
         podState.lastUpdated = System.currentTimeMillis()
         store()
+        rxBus.send(EventOmnipodDashPumpValuesChanged())
     }
 
     override fun updateFromAlarmStatusResponse(response: AlarmStatusResponse) {
-        TODO("Not yet implemented")
+        // TODO
+        logger.error(LTag.PUMP, "Not implemented: OmnipodDashPodStateManagerImpl.updateFromAlarmStatusResponse(AlarmStatusResponse)")
+
+        store()
+        rxBus.send(EventOmnipodDashPumpValuesChanged())
     }
 
     override fun reset() {

omnipod-dash/src/main/java/info/nightscout/androidaps/plugins/pump/omnipod/dash/ui/wizard/activation/viewmodel/action/DashInitializePodViewModel.kt

@@ -8,6 +8,7 @@ import info.nightscout.androidaps.logging.LTag
 import info.nightscout.androidaps.plugins.pump.omnipod.common.ui.wizard.activation.viewmodel.action.InitializePodViewModel
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.R
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.OmnipodDashManager
+import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.definition.AlertTrigger
 import io.reactivex.Single
 import io.reactivex.rxkotlin.subscribeBy
 import javax.inject.Inject
@@ -26,7 +27,8 @@ class DashInitializePodViewModel @Inject constructor(
 
     override fun doExecuteAction(): Single<PumpEnactResult> =
         Single.create { source ->
-            val disposable = omnipodManager.activatePodPart1().subscribeBy(
+            // TODO use configured value for low reservoir trigger
+            val disposable = omnipodManager.activatePodPart1(AlertTrigger.ReservoirVolumeTrigger(200)).subscribeBy(
                 onNext = { podEvent -> logger.debug(LTag.PUMP, "Received PodEvent in Pod activation part 1: $podEvent") },
                 onError = { throwable ->
                     logger.error(LTag.PUMP, "Error in Pod activation part 1", throwable)

omnipod-dash/src/main/java/info/nightscout/androidaps/plugins/pump/omnipod/dash/ui/wizard/activation/viewmodel/action/DashInsertCannulaViewModel.kt

@@ -5,12 +5,17 @@ import dagger.android.HasAndroidInjector
 import info.nightscout.androidaps.data.PumpEnactResult
 import info.nightscout.androidaps.interfaces.ProfileFunction
 import info.nightscout.androidaps.logging.AAPSLogger
+import info.nightscout.androidaps.logging.LTag
 import info.nightscout.androidaps.plugins.pump.omnipod.common.ui.wizard.activation.viewmodel.action.InsertCannulaViewModel
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.R
+import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.OmnipodDashManager
+import info.nightscout.androidaps.plugins.pump.omnipod.dash.util.mapProfileToBasalProgram
 import io.reactivex.Single
+import io.reactivex.rxkotlin.subscribeBy
 import javax.inject.Inject
 
 class DashInsertCannulaViewModel @Inject constructor(
+    private val omnipodManager: OmnipodDashManager,
     private val profileFunction: ProfileFunction,
     injector: HasAndroidInjector,
     logger: AAPSLogger
@@ -22,7 +27,24 @@ class DashInsertCannulaViewModel @Inject constructor(
 
     override fun isPodDeactivatable(): Boolean = true // TODO
 
-    override fun doExecuteAction(): Single<PumpEnactResult> = Single.just(PumpEnactResult(injector).success(false).comment("TODO")) // TODO
+    override fun doExecuteAction(): Single<PumpEnactResult> = Single.create { source ->
+        val profile = profileFunction.getProfile()
+        if (profile == null) {
+            source.onError(IllegalStateException("No profile set"))
+        } else {
+            val disposable = omnipodManager.activatePodPart2(mapProfileToBasalProgram(profile)).subscribeBy(
+                onNext = { podEvent -> logger.debug(LTag.PUMP, "Received PodEvent in Pod activation part 2: $podEvent") },
+                onError = { throwable ->
+                    logger.error(LTag.PUMP, "Error in Pod activation part 2", throwable)
+                    source.onSuccess(PumpEnactResult(injector).success(false).comment(throwable.message))
+                },
+                onComplete = {
+                    logger.debug("Pod activation part 2 completed")
+                    source.onSuccess(PumpEnactResult(injector).success(true))
+                }
+            )
+        }
+    }
 
     @StringRes
     override fun getTitleId(): Int = R.string.omnipod_common_pod_activation_wizard_insert_cannula_title