Image processing framework on MAX and Mojo with accelerated operations
You are working with MAX-CV, an accelerated image processing framework built on MAX and Mojo, inspired by the GPUImage series. This framework provides a high-level Python API with Mojo-based custom operations for realtime video processing and machine vision tasks.
Always consult these resources when working on the project:
1. **Python API Layer** (`max_cv/`)
- `ImagePipeline`: Main class for constructing image processing graphs
- `io.py`: Image loading/saving utilities with `load_image_into_tensor()`
- `operations/`: Python wrappers invoking Mojo custom ops via `ops.custom()`
2. **Mojo Custom Operations** (`max_cv/operations_mojo/`)
- Low-level image processing operations in Mojo
- Registered with `@compiler.register("op_name")` decorator
- Uses `foreach` with element-wise kernels for GPU/CPU execution
- Organized by category: blend, color_correction, draw, edge_detection, effects, transform
3. **Pipeline Flow**
- Load images into `Buffer` objects on CPU or accelerator
- `ImagePipeline` constructs a MAX `Graph` (images normalized to 0.0-1.0 float32)
- Operations chained with `pipeline.input_image`
- Compile once with `pipeline.compile()` for optimized `Model`
- Execute multiple times with `pipeline(buffer)`
- Output automatically restored to uint8 0-255 colorspace
- Examples: `pixi run showcase pixellate --value 15`
- `pixi run showcase brightness --value 0.3`
- `pixi run showcase gaussian_blur --kernel_size 16 --sigma 4.0`
```python
from max.driver import Accelerator, CPU, accelerator_count
device = CPU() if accelerator_count() == 0 else Accelerator()
```
```python
from max_cv import ImagePipeline, load_image_into_tensor, operations as ops
from max.dtype import DType
image_tensor = load_image_into_tensor(image_path, device)
with ImagePipeline(
"pipeline_name",
image_tensor.shape,
pipeline_dtype=DType.float32,
device=device,
) as pipeline:
result = ops.brightness(device, pipeline.input_image, 0.5)
pipeline.output(result)
pipeline.compile()
result = pipeline(image_tensor)
```
```python
with ImagePipeline(..., num_inputs=2) as pipeline:
result = ops.dissolve_blend(
device,
pipeline.input_images[0],
pipeline.input_images[1],
0.5
)
pipeline.output(result)
result = pipeline(image1_buffer, image2_buffer)
```
```mojo
@compiler.register("operation_name")
struct OperationName:
@staticmethod
fn execute[target: StaticString](
output: OutputTensor,
param: Float32,
image: InputTensor[dtype = output.dtype, rank = output.rank],
ctx: DeviceContextPtr,
) raises:
@parameter
@always_inline
fn kernel[width: Int](idx: IndexList[image.rank]) -> SIMD[image.dtype, width]:
# Process pixels here
return image.load[width](idx) + param
foreach[kernel, target=target](output, ctx)
```
Leave a review
No reviews yet. Be the first to review this skill!
# Download SKILL.md from killerskills.ai/api/skills/max-cv-development/raw