01 | OVERLAP INDEPENDENT WORK

A robot can run its intake while driving toward a collection area. It may also position a mechanism while another action is progressing, provided their motion and loads are compatible.

Parallel physical movement does not require multiple Python threads. A cooperative runner can update several actions each cycle, then apply their requests to separate resources.

Build on 09.1: each action starts once, updates promptly and reports RUNNING, DONE or FAILED.

02 | PLAN WHICH ACTIONS MAY OVERLAP

Keep prerequisite actions sequential. For example, lower a mechanism to a verified safe position before driving through a restricted opening.

Use the travel-limit and feedback ideas from 05.5 for arms and lifts. A timed motor request alone is not suitable evidence that a constrained mechanism reached a safe position.

03 | RESERVE RESOURCES BEFORE STARTING

Each action declares its resource: drive or intake. The parallel group rejects overlapping declarations before any output is applied.

DriveChild.resources = {"drive"}
TimedIntake.resources = {"intake"}
# Two children claiming "drive" must not start together.

These names are logical reservations. Keep the mapping to real motors accurate; two different labels for the same motor would bypass this simple check.

The runner is the only code that commits requests to motors. Action methods calculate requests and return them.

04 | DEFINE THE JOIN POLICY

Do not treat a failed child as a successful race winner. Define success and failure separately.

The example waits for both drive and intake before beginning the turn. Whichever finishes first stops its own output while the other continues.

05 | TIMED COMPLETION HAS A LIMITED MEANING

The intake action runs at a configured request for 1,200 ms. Its DONE status means the commanded interval elapsed. It does not prove an object was collected or that the rollers actually turned.

For collection success, add suitable object feedback. For jam detection, use a validated policy such as the one discussed in 05.6.

The drive action uses encoder distance, heading and settling. Those measurements also have physical limitations such as wheel slip.

06 | JOIN SUCCESS BEFORE STARTING THE NEXT STEP

The machine enters MOVE_AND_INTAKE with both children created from the same entry time. It then checks each child on every tick.

If the drive completes first, the group keeps checking that its encoder position and heading remain within the finish window while the intake continues. Drift causes failure rather than silently accepting an old completion.

07 | COMPLETE COOPERATIVE DRIVE-AND-INTAKE STARTER

Paste below # Begin project code, preserving generated imports and Brain setup. Drive motors: left ports 3/4, right 1/2. Two-stage intake: ports 7/6. Inertial: 20.

Verify intake directions and gearing for your build. Avoid duplicate Devices declarations. MOTION_CONFIGURED begins False; verify signs, encoder scale and a clear complete test path before enabling it.

# =============================================================================
# TEAM [MY TEAM NAME / NUMBER] | COOPERATIVE PARALLEL DRIVE AND INTAKE
# =============================================================================
# Keep generated imports and brain above '# Begin project code'.
# Manual configuration below: do not duplicate declarations in Devices.
import math
left_1 = Motor(Ports.PORT3, GearSetting.RATIO_6_1, False)
left_2 = Motor(Ports.PORT4, GearSetting.RATIO_6_1, False)
right_1 = Motor(Ports.PORT1, GearSetting.RATIO_6_1, True)
right_2 = Motor(Ports.PORT2, GearSetting.RATIO_6_1, True)
left_drive = MotorGroup(left_1, left_2)
right_drive = MotorGroup(right_1, right_2)
inertial = Inertial(Ports.PORT20)
controller = Controller(PRIMARY)
# =============================================================================
# TWO-STAGE INTAKE: VERIFY REVERSALS FOR YOUR MECHANISM
# =============================================================================
intake_1 = Motor(Ports.PORT7, GearSetting.RATIO_18_1, False)
intake_2 = Motor(Ports.PORT6, GearSetting.RATIO_18_1, False)
intake = MotorGroup(intake_1, intake_2)
INTAKE_PERCENT = 20.0
INTAKE_DURATION_MS = 1200
INTAKE_TIMEOUT_MS = 2000

MOTION_CONFIGURED = False
MM_PER_MOTOR_TURN = 219.44  # Example: calibrate wheel size/external gearing
DRIVE_MM = 300.0
TURN_DEG = 90.0            # Relative clockwise turn using continuous rotation
DRIVE_KP = 0.05
TURN_KP = 0.7
MAX_DRIVE = 20.0
MAX_TURN = 10.0
MAX_WHEEL_OUTPUT = 25.0
DISTANCE_TOL_MM = 10.0
HEADING_TOL_DEG = 2.0
SIDE_RATE_TOL_MM_S = 30.0
TURN_RATE_TOL_DEG_S = 8.0
SETTLE_MS = 200
DRIVE_TIMEOUT_MS = 8000
TURN_TIMEOUT_MS = 6000
ROUTE_TIMEOUT_MS = 14000
POLL_MS = 20
MAX_UPDATE_GAP_MS = 100
ready = False

def clamp(v, low, high):
    return max(low, min(high, v))

def angle_error(target, current):
    return (target-current+180)%360-180

def stop_drive():
    left_drive.stop(BRAKE)
    right_drive.stop(BRAKE)

def side_output(group, value):
    if value == 0:
        group.stop(BRAKE)
    else:
        group.spin(FORWARD if value > 0 else REVERSE, abs(value), PERCENT)

def read_snapshot():
    if not inertial.installed() or inertial.is_calibrating():
        raise ValueError("Inertial unavailable")
    turns = (left_1.position(TURNS), left_2.position(TURNS),
             right_1.position(TURNS), right_2.position(TURNS))
    rotation = inertial.rotation(DEGREES)
    if not all(math.isfinite(v) for v in turns+(rotation,)):
        raise ValueError("Invalid encoder/inertial data")
    return {"left": (turns[0]+turns[1])*MM_PER_MOTOR_TURN/2,
            "right": (turns[2]+turns[3])*MM_PER_MOTOR_TURN/2,
            "rotation": rotation}

# =============================================================================
# ACTION: NO LOOPS, WAITS OR DEVICE ACCESS INSIDE update()
# =============================================================================
class MotionAction:
    def __init__(self, mode, now, snapshot):
        if mode not in ("DRIVE", "TURN"):
            raise ValueError("Invalid action mode")
        self.mode = mode
        self.started = now
        self.start = dict(snapshot)  # Capture once; never reset on each update.
        self.settled_since = None
        self.finished = None

    def update(self, now, snapshot, slow):
        if self.finished is not None:
            return self.finished, 0.0, 0.0, ""
        limit = DRIVE_TIMEOUT_MS if self.mode == "DRIVE" else TURN_TIMEOUT_MS
        if now-self.started >= limit:
            self.finished = "FAILED"
            return "FAILED", 0.0, 0.0, self.mode+"_TIMEOUT"
        if self.mode == "DRIVE":
            left_distance = snapshot["left"]-self.start["left"]
            right_distance = snapshot["right"]-self.start["right"]
            distance_error = DRIVE_MM-(left_distance+right_distance)/2
            heading_error = angle_error(self.start["rotation"], snapshot["rotation"])
            if distance_error < -DISTANCE_TOL_MM:
                self.finished = "FAILED"
                return "FAILED", 0.0, 0.0, "DRIVE_OVERSHOOT"
            near = (abs(DRIVE_MM-left_distance) <= DISTANCE_TOL_MM
                    and abs(DRIVE_MM-right_distance) <= DISTANCE_TOL_MM
                    and abs(heading_error) <= HEADING_TOL_DEG)
            base = (clamp(DRIVE_KP*distance_error, 0, MAX_DRIVE)
                    if abs(heading_error) <= 10 else 0.0)
            turn = clamp(TURN_KP*heading_error, -MAX_TURN, MAX_TURN)
        else:
            # Relative target is fixed at entry; continuous rotation preserves +90.
            heading_error = self.start["rotation"]+TURN_DEG-snapshot["rotation"]
            near = abs(heading_error) <= HEADING_TOL_DEG
            base = 0.0
            turn = clamp(TURN_KP*heading_error, -MAX_TURN, MAX_TURN)
        if near:
            if slow:
                if self.settled_since is None:
                    self.settled_since = now
                if now-self.settled_since >= SETTLE_MS:
                    self.finished = "DONE"
                    return "DONE", 0.0, 0.0, ""
            else:
                self.settled_since = None
            return "RUNNING", 0.0, 0.0, ""  # Brake while observing settling.
        self.settled_since = None
        left, right = base+turn, base-turn
        scale = max(1.0, abs(left)/MAX_WHEEL_OUTPUT, abs(right)/MAX_WHEEL_OUTPUT)
        return "RUNNING", left/scale, right/scale, ""

# =============================================================================
# COMMON ACTION INTERFACE: STATUS, RESOURCE REQUESTS, REASON
# =============================================================================
def zero_commands():
    return {"drive": (0.0, 0.0), "intake": 0.0}

def stop_all():
    stop_drive()
    intake.stop(BRAKE)

class DriveChild:
    resources = {"drive"}
    def __init__(self, now, snapshot):
        self.motion = MotionAction("DRIVE", now, snapshot)
    def update(self, now, snapshot, slow):
        status, left, right, reason = self.motion.update(now, snapshot, slow)
        return status, {"drive": (left, right)}, reason
    def still_at_target(self, snapshot):
        start = self.motion.start
        return (abs(snapshot["left"]-start["left"]-DRIVE_MM) <= DISTANCE_TOL_MM
                and abs(snapshot["right"]-start["right"]-DRIVE_MM) <= DISTANCE_TOL_MM
                and abs(angle_error(start["rotation"],snapshot["rotation"])) <= HEADING_TOL_DEG)

class TimedIntake:
    resources = {"intake"}
    def __init__(self, now):
        self.started = now
        self.finished = None
    def update(self, now, snapshot, slow):
        if self.finished is not None:
            return self.finished, {"intake": 0.0}, ""
        elapsed = now-self.started
        if elapsed >= INTAKE_TIMEOUT_MS:
            self.finished = "FAILED"
            return "FAILED", {"intake": 0.0}, "INTAKE_TIMEOUT"
        if elapsed >= INTAKE_DURATION_MS:
            self.finished = "DONE"
            return "DONE", {"intake": 0.0}, ""
        return "RUNNING", {"intake": INTAKE_PERCENT}, ""

class ParallelAll:
    def __init__(self, children):
        if not children:
            raise ValueError("Need parallel children")
        reserved = set()
        for child in children:
            if reserved.intersection(child.resources):
                raise ValueError("Resource conflict before group execution")
            reserved.update(child.resources)
        self.children = children
        self.finished = None
        self.reason = ""
    def update(self, now, snapshot, slow):
        if self.finished is not None:
            return self.finished, zero_commands(), self.reason
        commands = zero_commands()
        all_done = True
        for child in self.children:
            status, requests, reason = child.update(now, snapshot, slow)
            if status == "FAILED":
                self.finished, self.reason = "FAILED", reason
                return "FAILED", zero_commands(), reason  # Cancel sibling output.
            if status != "DONE":
                all_done = False
            # A completed drive must not drift away while another child runs.
            if isinstance(child, DriveChild) and status == "DONE" and not child.still_at_target(snapshot):
                self.finished, self.reason = "FAILED", "DRIVE_DRIFTED_WHILE_WAITING"
                return "FAILED", zero_commands(), self.reason
            commands.update(requests)
        if all_done:
            self.finished = "DONE"
            return "DONE", zero_commands(), ""
        return "RUNNING", commands, ""

# =============================================================================
# SEQUENCE: PARALLEL DRIVE/INTAKE -> TURN -> DONE
# =============================================================================
class ParallelAuto:
    def __init__(self, now, snapshot):
        self.state = "MOVE_AND_INTAKE"
        self.reason = ""
        self.deadline = now+ROUTE_TIMEOUT_MS
        self.group = ParallelAll([DriveChild(now,snapshot), TimedIntake(now)])
        self.turn = None
    def fail(self, reason):
        self.state, self.reason = "FAULT", reason
        return zero_commands()
    def tick(self, now, snapshot, slow):
        if self.state in ("DONE", "FAULT"):
            return zero_commands()
        if now >= self.deadline:
            return self.fail("ROUTE_DEADLINE")
        if self.state == "MOVE_AND_INTAKE":
            status, commands, reason = self.group.update(now,snapshot,slow)
            if status == "FAILED":
                return self.fail(reason)
            if status == "DONE":
                self.state = "TURN"
                self.turn = MotionAction("TURN",now,snapshot)
                return zero_commands()
            return commands
        status,left,right,reason = self.turn.update(now,snapshot,slow)
        if status == "FAILED":
            return self.fail(reason)
        if status == "DONE":
            self.state = "DONE"
            return zero_commands()
        return {"drive": (left,right), "intake": 0.0}

# =============================================================================
# SINGLE RUNNER AND OUTPUT COMMIT
# =============================================================================
def run_parallel_auto():
    stop_all()
    machine = None
    try:
        settings = (MM_PER_MOTOR_TURN, DRIVE_MM, TURN_DEG, DRIVE_KP, TURN_KP,
                    MAX_DRIVE,MAX_TURN,MAX_WHEEL_OUTPUT,DISTANCE_TOL_MM,
                    HEADING_TOL_DEG,SIDE_RATE_TOL_MM_S,TURN_RATE_TOL_DEG_S,
                    SETTLE_MS,DRIVE_TIMEOUT_MS,TURN_TIMEOUT_MS,ROUTE_TIMEOUT_MS,
                    POLL_MS,MAX_UPDATE_GAP_MS,INTAKE_PERCENT,INTAKE_DURATION_MS,INTAKE_TIMEOUT_MS)
        if (not all(math.isfinite(v) for v in settings)
                or min(MM_PER_MOTOR_TURN,DRIVE_MM,DRIVE_KP,TURN_KP,MAX_DRIVE,
                       MAX_TURN,MAX_WHEEL_OUTPUT,DISTANCE_TOL_MM,HEADING_TOL_DEG,
                       SETTLE_MS,DRIVE_TIMEOUT_MS,TURN_TIMEOUT_MS,ROUTE_TIMEOUT_MS) <= 0
                or SIDE_RATE_TOL_MM_S < 0 or TURN_RATE_TOL_DEG_S < 0
                or not 0 < POLL_MS < MAX_UPDATE_GAP_MS
                or max(MAX_DRIVE,MAX_TURN) > MAX_WHEEL_OUTPUT or MAX_WHEEL_OUTPUT > 25
                or not 0 < abs(INTAKE_PERCENT) <= 30
                or not 0 < INTAKE_DURATION_MS < INTAKE_TIMEOUT_MS):
            return "INVALID_SETTINGS"
        previous = None
        previous_ms = brain.timer.time(MSEC)
        while True:
            if (not ready or not MOTION_CONFIGURED or not competition.is_enabled()
                    or not competition.is_autonomous() or controller.buttonB.pressing()):
                return "CANCELLED_OR_NOT_READY"
            snapshot = read_snapshot()
            now = brain.timer.time(MSEC)
            if now-previous_ms > MAX_UPDATE_GAP_MS:
                return "UPDATE_GAP"
            if (not competition.is_enabled() or not competition.is_autonomous()
                    or controller.buttonB.pressing()):
                return "CANCELLED_OR_NOT_READY"
            dt = (now-previous_ms)/1000.0
            slow = (previous is not None and dt > 0 and
                    max(abs(snapshot["left"]-previous["left"]),
                        abs(snapshot["right"]-previous["right"]))/dt <= SIDE_RATE_TOL_MM_S
                    and abs(snapshot["rotation"]-previous["rotation"])/dt <= TURN_RATE_TOL_DEG_S)
            previous,previous_ms = snapshot,now
            if machine is None:
                machine = ParallelAuto(now,snapshot)
                print("STATE",machine.state)
            old_state = machine.state
            commands = machine.tick(now,snapshot,slow)
            if machine.state != old_state:
                print("STATE",old_state,"->",machine.state,"REASON",machine.reason)
            # All action results have been checked before any motor receives output.
            left,right = commands["drive"]
            side_output(left_drive,left)
            side_output(right_drive,right)
            side_output(intake,commands["intake"])
            if machine.state in ("DONE","FAULT"):
                return machine.state+": "+machine.reason
            wait(POLL_MS,MSEC)
    except Exception:
        return "SENSOR_RESOURCE_OR_ACTION_FAULT"
    finally:
        stop_all()

# =============================================================================
# COMPETITION CALLBACKS
# =============================================================================
def autonomous():
    stop_all()
    if ready and MOTION_CONFIGURED:
        print("parallel result",run_parallel_auto())

def driver_control():
    stop_all()  # Teaching project: no driver movement.

competition = Competition(driver_control,autonomous)
stop_all()
if MOTION_CONFIGURED and not competition.is_enabled() and inertial.installed():
    inertial.calibrate()  # Stationary disabled setup; may block.
    ready = not inertial.is_calibrating() and not competition.is_enabled()
while True:
    wait(100,MSEC)

The routine drives 300 mm while running the intake for 1.2 seconds. Once both actions complete, it turns clockwise 90° and ends.

The drive and turn actions retain their own deadlines. The whole sequence shares fourteen seconds, and every required failure prevents later normal movement.

08 | COOPERATIVE PARALLELISM IS NOT ATOMIC OUTPUT

The runner updates all child results before sending output. A detected child failure returns zero requests for both resources in that cycle.

Motor API calls are still sequential. They are not an atomic hardware transaction. If one command fails, finally attempts to stop every participating motor.

Do not add a second intake or drivetrain thread that also writes commands. If you later need separate tasks, define explicit ownership and cancellation rather than relying on timing.

09 | CHOOSE STOP MODES FOR THE MECHANISM

The starter brakes the drive and intake when their requests become zero. A roller often does not need active position holding.

A loaded lift may need a different policy, mechanical support or controlled holding. Do not copy the intake’s stop policy to a gravity-loaded mechanism without reviewing its behaviour.

HOLD actively maintains motor position; it is not a substitute for travel limits or a universal way to protect a mechanism.

10 | TEST FAILURE IN EACH CHILD

1. Test drive alone and intake alone with conservative settings.
2. Check the declared resource reservations with mocked actions.
3. Test one child finishing before the other.
4. Inject a drive failure while the intake is active.
5. Inject an intake failure while drive is active.
6. Test B, mode changes and an expired route budget.
7. Measure a combined run and compare it with separate runs.

Log child status, resource requests, state transitions and measured mechanism outcomes. Increased battery demand or mechanical load can change performance during overlap.

11 | COMMON MISTAKES

• Starting a child every cycle: create it once on entry.
• Sharing motors under different labels: reserve actual resources consistently.
• Continuing a sibling after required failure: define and enforce group cancellation.
• Calling timed intake DONE a collected object: distinguish timing from sensor evidence.
• Stopping the whole group when the first action completes: that is not an ALL policy.
• Ignoring completed-action drift: recheck required join conditions.
• Using the intake stop mode for a loaded lift: choose mechanism-specific cleanup.

CHALLENGES | OVERLAP, JOIN AND CANCEL

1. Trace: show how drive and intake requests coexist during MOVE_AND_INTAKE.
2. Reserve: demonstrate rejection of two drive-resource children.
3. Mock: make the intake finish first and confirm drive continues.
4. Reverse the order: make drive finish first and check its position while waiting.
5. Fail: inject failure in each child and verify all output becomes zero.
6. Measure: compare combined and sequential durations on a clear practice course.
7. Extend: propose an object-detection finish rule for the intake.
8. Evaluate: explain when a drive-and-lift overlap should be forbidden even with separate motor resources.

OFFICIAL REFERENCES

• Motor commands, feedback and stopping modes
• Competition callback and mode API

The cooperative scheduler, resource declarations and join policy are independently written teaching examples requiring physical configuration and testing.