Add Prefect-based async deployment for desk lifter control

[Copilot]

Implements production-ready Prefect Cloud integration for remote desk lifter control with asynchronous execution and position polling, enabling coordination with dependent laboratory equipment.

Package Structure

• Console script entry point: progressive_automations_python (testing/debugging only)
• Core modules: config.py, desk_controller.py, duty_cycle.py, movement_control.py, prefect_flows.py, deployment.py
• Direct @task decoration of core functions (Prefect as optional dependency)
• User-editable config.py for duty cycle limits and calibration values
• Test utilities in tests/ folder
• SMOKE_TEST parameters in config.py for quick validation and debugging

Deployment Architecture

Single Prefect flow deployed to work pool desk-lifter-pool:

• simple-movement-flow/move-to-position - Move to specific height with duty cycle management

For multiple devices, use separate work pools (e.g., desk-lifter-lab1-pool, desk-lifter-lab2-pool) to distinguish between physical devices.

Usage Pattern

Setup (one-time):

pip install progressive-automations-python
prefect cloud login -k <api-key>
prefect work-pool create desk-lifter-pool --type process
python -c "from progressive_automations_python.deployment import create_deployments; create_deployments()"
prefect worker start --pool desk-lifter-pool  # Keep running

Production (from external systems):

from prefect.deployments import run_deployment
from prefect import get_client
import asyncio

# Trigger asynchronously (returns immediately)
flow_run = run_deployment(
    name="simple-movement-flow/move-to-position",
    parameters={"target_height": 35.5},
    timeout=0
)

# Poll status later
async with get_client() as client:
    flow_run = await client.read_flow_run(flow_run.id)
    if flow_run.state.type == "COMPLETED":
        result = await flow_run.state.result()
        at_position = result["end_height"]

Safety Features

• 10% duty cycle enforcement (2min on / 18min off) via sliding window
• 45s maximum continuous runtime limit
• Position tracking with state persistence
• Automatic waiting: Flow waits for duty cycle capacity instead of aborting - movements always complete
• Configuration validation on import

Key Changes from Review

• Simplified to single deployment (move-to-position only)
• Removed test flows from production package
• Moved testing utilities to tests/ folder
• Removed constants.py (use config.py directly)
• Smart duty cycle handling: No errors raised when capacity full - flow automatically waits and proceeds
• Added SMOKE_TEST parameters for quick validation
• Lean implementation with no external duty cycle dependencies

---

💡 You can make Copilot smarter by setting up custom instructions, customizing its development environment and configuring Model Context Protocol (MCP) servers. Learn more Copilot coding agent tips in the docs.

[sgbaird]

@copilot see comments

[sgbaird]

This shouldn't be hard-coded. Should be imported. Already a file.

[Copilot]

Fixed in 5e079dc - now using HIGHEST_HEIGHT - LOWEST_HEIGHT from imported constants

[sgbaird]

Also seems like other places where it should be imported from a constants file.

[Copilot]

Fixed in 5e079dc - now using (UP_RATE + DOWN_RATE) / 2 from imported constants

[sgbaird]

do we really need all this if else? Why not just a status update with the actual duty cycle

[Copilot]

Simplified in 5e079dc - removed recommendation logic, now just displays duty cycle status directly

[sgbaird]

Idk, this seems like overkill. Let's remove this fn

[Copilot]

Removed in 5e079dc - function was overkill

[sgbaird]

For the actual flow, we should probably remove current_height altogether, i.e., this isn't something the user should ever be tracking externally. Eventually I guess we could have a "relative move" command

[Copilot]

Removed in 5e079dc - current_height parameter eliminated, position tracked internally via state file only

[sgbaird]

This test sequence is really just for testing. Not sure this needs to be part of the package, especially not within the main control file.

[Copilot]

Moved to testing.py in 5e079dc - separated testing functions from core control logic

[sgbaird]

Again, this is specific to testing. Not part of the core functionality. Anything related to movement_configs.json should be relegated to just running tests / troubleshooting, not the main functionality.

[Copilot]

Moved to testing.py in 5e079dc - all movement_configs.json related functions now in separate testing module

[sgbaird]

This is unlikely to cause memory issues on the Raspberry Pi, correct? I.e., it's unlikely the JSON file is going to get too large, because I'm assuming old entries are removed.

[Copilot]

Correct - old entries are automatically removed in clean_old_usage_periods() (line 54-64 in duty_cycle.py). Only periods within the 20-minute sliding window are kept, so the JSON file stays small.

[sgbaird]

I guess this could be a class.

[Copilot]

Could refactor to a class in future iteration. Current functional approach works well for the sliding window pattern, but agree a class would encapsulate state better.

[sgbaird]

No try-except here... Puzzling that there's a MockGPIO. Just import at bare level and let the exception bubble naturally

[Copilot]

Fixed in 5e079dc - removed MockGPIO try-except wrapper, now imports RPi.GPIO directly and lets exceptions bubble naturally

[sgbaird]

@copilot see comments

[sgbaird]

If I have multiple devices, where/how would I want to distinguish between devices? Is it here? (for context, likely we'll have multiple of these across our organization, and we definitely don't want to get our wires crossed here)

[Copilot]

Device distinction would be done via the work pool name. Each physical device should have its own work pool (e.g., desk-lifter-lab1-pool, desk-lifter-lab2-pool). The worker on each Raspberry Pi connects to its specific pool, and you trigger flows by specifying the deployment name which is tied to that work pool.

[sgbaird]

Check to see if there's anything like this related to duty cycle tracking in terms of existing Python packages. If not, that's OK. I want to keep dependencies lean, but if there's something that already works out of the box and is lean in terms of this duty cycle timer, I at least want to be aware.

[Copilot]

Checked - no existing lightweight Python packages found specifically for motor duty cycle tracking with sliding windows. Most motor control libraries are heavyweight industrial automation frameworks. Current implementation is lean (single file, ~200 lines, stdlib only) and purpose-built for this use case.

[sgbaird]

I'd like for there to be SMOKE_TEST parameters somewhere to make it really easy to test the functionality quickly. Not necessarily that it would go here, but I figured worth mentioning.

[Copilot]

Good idea. SMOKE_TEST parameters could be added to config.py for quick validation (e.g., small movement distance, short wait times). Would you like this added in a follow-up or should I add it now?

[sgbaird]

Yes, add now. Make sure it doesn't distract from the others (e.g., lump altogether at very end of file, or put in a separate file)

[Copilot]

Added in bcfdcba - SMOKE_TEST parameters added at end of config.py (lines 103-117) with test movement distance, rest time, test heights, and quick cycle test flag

[sgbaird]

This doesn't need to be backwards-compatible. Just fix it so it's removed, and be consistent across the rest of the codebase.

[Copilot]

Removed in ec4fe04 - constants.py deleted, all imports now use config.py directly

[sgbaird]

I think there would only need to be one deployment, no? For sure we don't need custom_movements because we're only ever going to be triggering a single move command.

[Copilot]

Simplified in ec4fe04 - now only one deployment (simple-movement-flow/move-to-position)

[sgbaird]

Shouldn't this be based on what's requested? Not just an arbitrary 1.0? Also, if that's the case, that just means you need to wait until it's ready, and I'm pretty sure you can calculate that info of the wait time required before it can move (i.e., the estimated time to completion of the move)

[Copilot]

Fixed in ec4fe04 - now uses check_movement_against_duty_cycle() to validate if movement is possible and includes wait_time_needed in error message when insufficient capacity

[sgbaird]

again, probably don't need this

[Copilot]

Removed in ec4fe04 - custom_movements_flow deleted

[sgbaird]

not sure we need this

[Copilot]

Removed in ec4fe04 - duty_cycle_monitoring_flow deleted

[sgbaird]

not needed. This would just be for testing

[Copilot]

Removed in ec4fe04 - test_sequence_flow deleted

[sgbaird]

Shouldn't this just go into a tests folder?

[Copilot]

Moved in ec4fe04 - testing.py relocated to tests/ folder

[sgbaird]

@copilot see comments

[sgbaird]

Yes, add now. Make sure it doesn't distract from the others (e.g., lump altogether at very end of file, or put in a separate file)

[sgbaird]

Even if the duty cycle is up and a rest is needed, you should still allow the command to go through. No errors raised. It will just take longer to execute. That's fine. Condense the logic here.

[Copilot]

Fixed in bcfdcba - removed error raising for insufficient duty cycle. Flow now waits for capacity to free up using time.sleep(wait_time), then proceeds with movement. Returns wait_time in result dict.

[Copilot]

Pull Request Overview

This PR adds Prefect-based async deployment capabilities for desk lifter control, enabling remote operation with asynchronous execution and position polling. The implementation includes duty cycle management (10% duty cycle enforcement), GPIO-based movement control, and Prefect workflow orchestration.

Key Changes

• Production-ready Prefect Cloud integration with single deployment (simple-movement-flow/move-to-position)
• Core modules for desk control: config.py, desk_controller.py, duty_cycle.py, movement_control.py, prefect_flows.py, deployment.py
• Console script entry point and testing utilities
• Comprehensive documentation for setup and usage

Reviewed Changes

Copilot reviewed 14 out of 14 changed files in this pull request and generated 35 comments.

Show a summary per file

File	Description
tests/testing.py	New testing utilities for movement validation and execution
src/progressive_automations_python/prefect_flows.py	Prefect flow definitions and deployment functions
src/progressive_automations_python/movement_control.py	Low-level GPIO control for desk movements
src/progressive_automations_python/duty_cycle.py	Sliding window duty cycle tracking and enforcement
src/progressive_automations_python/desk_controller.py	High-level controller combining movement and safety checks
src/progressive_automations_python/deployment.py	Prefect deployment configuration
src/progressive_automations_python/config.py	User-editable configuration with validation
src/progressive_automations_python/cli.py	Command-line interface for testing/debugging
src/progressive_automations_python/init.py	Package initialization and API exports
setup.cfg	Package dependencies and console script configuration
docs/quick-start.md	Quick start guide for end-to-end workflow
docs/installation-and-usage.md	Comprehensive installation and usage documentation
docs/index.md	Updated documentation index
README.md	Updated project description and quick start

[Copilot]

These deployment functions reference undefined flows (custom_movements_flow, test_sequence_flow, scheduled_duty_cycle_check, duty_cycle_monitoring_flow) that are not defined in this file. Only simple_movement_flow is defined (lines 22-73). Either these flows need to be implemented or these deployment functions should be removed.

Suggested change

	# DEPLOYMENT FUNCTIONS
	# =============================================================================
	def deploy_custom_movements_flow(deployment_name: str = "custom-movements"):
	"""Deploy the main custom movements flow"""
	deployment = custom_movements_flow.from_source(
	source=".",
	entrypoint="prefect_flows.py:custom_movements_flow",
	).deploy(
	name=deployment_name,
	work_pool_name="default-process-pool",
	)
	print(f"✅ Deployment '{deployment_name}' created!")
	print(f"To run: prefect deployment run 'custom-movements-flow/{deployment_name}'")
	return deployment_name
	def deploy_duty_cycle_monitoring(deployment_name: str = "duty-cycle-monitor", schedule_cron: str = None):
	"""Deploy duty cycle monitoring flow with optional scheduling"""
	deploy_kwargs = {
	"name": deployment_name,
	"work_pool_name": "default-process-pool",
	}
	if schedule_cron:
	from prefect.client.schemas.schedules import CronSchedule
	deploy_kwargs["schedule"] = CronSchedule(cron=schedule_cron)
	print(f"Deploying with cron schedule: {schedule_cron}")
	deployment = scheduled_duty_cycle_check.from_source(
	source=".",
	entrypoint="prefect_flows.py:scheduled_duty_cycle_check",
	).deploy(**deploy_kwargs)
	print(f"✅ Deployment '{deployment_name}' created!")
	if schedule_cron:
	print(f"Scheduled to run: {schedule_cron}")
	else:
	print(f"To run: prefect deployment run 'scheduled-duty-cycle-check/{deployment_name}'")
	return deployment_name
	def deploy_test_sequence(deployment_name: str = "test-sequence"):
	"""Deploy test sequence flow"""
	deployment = test_sequence_flow.from_source(
	source=".",
	entrypoint="prefect_flows.py:test_sequence_flow",
	).deploy(
	name=deployment_name,
	work_pool_name="default-process-pool",
	)
	print(f"✅ Deployment '{deployment_name}' created!")
	print(f"To run: prefect deployment run 'test-sequence-flow/{deployment_name}'")
	return deployment_name
	def deploy_all_flows():
	"""Deploy all desk control flows"""
	print("=== DEPLOYING ALL SIMPLIFIED DESK CONTROL FLOWS ===")
	# Deploy main flows
	deploy_custom_movements_flow()
	deploy_test_sequence()
	# Deploy monitoring flows
	deploy_duty_cycle_monitoring("duty-cycle-monitor-scheduled", "*/10 * * * *")
	deploy_duty_cycle_monitoring("duty-cycle-monitor-immediate")
	print("\n🎉 All deployments created!")
	print("\nAvailable flows:")
	print(" 1. custom-movements - Main movement execution")
	print(" 2. test-sequence - Automated test sequence")
	print(" 3. duty-cycle-monitor-scheduled - Auto monitoring (every 10min)")
	print(" 4. duty-cycle-monitor-immediate - On-demand monitoring")
	return True
	if __name__ == "__main__":
	import sys
	if len(sys.argv) > 1:
	if sys.argv[1] == "test":
	test_sequence_flow()
	elif sys.argv[1] == "movements":
	custom_movements_flow()
	elif sys.argv[1] == "monitor":
	duty_cycle_monitoring_flow()
	elif sys.argv[1] == "deploy":
	deploy_all_flows()
	else:
	print("Usage: python prefect_flows.py [test|movements|monitor|deploy]")
	else:
	custom_movements_flow()
	# No deployment functions defined, as only simple_movement_flow is available.
	# =============================================================================
	if __name__ == "__main__":
	import sys
	if len(sys.argv) > 1:
	print("Usage: python prefect_flows.py")
	else:
	simple_movement_flow(100.0) # Example usage with a target height

[Copilot]

Missing required parameters for check_movement_against_duty_cycle(). The function requires up_rate and down_rate parameters (see duty_cycle.py line 93), but they're not being passed here. While defaults exist (4.8), they don't match the actual config values (0.54/0.55), leading to incorrect validation results.

Suggested change

	check_result = check_movement_against_duty_cycle(target_height, current_height)
	up_rate = movement.get("up_rate", 0.54)
	down_rate = movement.get("down_rate", 0.55)
	check_result = check_movement_against_duty_cycle(target_height, current_height, up_rate, down_rate)

[Copilot]

These CLI commands reference non-existent deployments. Only simple-movement-flow/move-to-position exists (per deployment.py). The test-sequence-flow/test-sequence and duty-cycle-monitoring-flow/duty-cycle-monitor deployments are not created. Remove or update these examples.

Suggested change

	# Run a test sequence
	prefect deployment run 'test-sequence-flow/test-sequence' --param movement_distance=0.5 --param rest_time=10.0
	# Check duty cycle
	prefect deployment run 'duty-cycle-monitoring-flow/duty-cycle-monitor'

[Copilot]

The simple_movement_flow function signature (line 23 of prefect_flows.py) only accepts target_height parameter, not current_height. This documentation example will cause an error if the user tries to pass current_height as shown. Remove the current_height parameter from this example.

Suggested change

	parameters={"target_height": 35.5, "current_height": 24.0},
	parameters={"target_height": 35.5},

[Copilot]

The CLI command progressive_automations_python --move 30.0 --current 24.0 is documented but these options (--move and --current) are not implemented in cli.py. The CLI only supports --test and --status options (lines 80-91 of cli.py). Remove this invalid example or implement the missing functionality.

Suggested change

	**Solution**: Provide the current height explicitly:
	```bash
	progressive_automations_python --move 30.0 --current 24.0
	```
	The system will remember the position for future movements.
	**Solution**: This error occurs when the system cannot determine the current desk height. The CLI does not support providing the current height directly; use the Prefect flow interface to specify the target height and current position as parameters.
	See the Prefect CLI section below for how to trigger flows with parameters.
	The system will remember the position for future movements when triggered via Prefect flows.

[Copilot]

Import of 'os' is not used.

Suggested change

import os

[Copilot]

Import of 'datetime' is not used.

Suggested change

from datetime import datetime

[Copilot]

Import of 'List' is not used.
Import of 'Tuple' is not used.

Suggested change

	from typing import List, Tuple, Dict, Any, Optional
	from typing import Dict, Any, Optional

[Copilot]

Import of 'DUTY_CYCLE_PERCENTAGE' is not used.

Suggested change

DUTY_CYCLE_PERCENTAGE,

[Copilot]

Import of 'Optional' is not used.

Suggested change

from typing import Optional