fix(sequences): target only checked lane groups

Use zone group checkboxes in the editor; empty lane groups no longer
fall back to the whole zone. Remove cross-lane device splitting.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
2026-05-19 00:23:21 +12:00
parent baec87068a
commit 7a7bedc07c
2 changed files with 89 additions and 65 deletions

View File

@@ -78,8 +78,13 @@ def _normalize_sequence_lanes(doc: Dict[str, Any]) -> List[List[Dict[str, Any]]]
def _group_ids_for_lane_step(
sequence_doc: Dict[str, Any], step: Dict[str, Any], lane_index: int, num_lanes: int
sequence_doc: Dict[str, Any],
step: Dict[str, Any],
lane_index: int,
num_lanes: int,
zone_doc: Optional[Dict[str, Any]] = None,
) -> List[str]:
_ = zone_doc
lgs = sequence_doc.get("lanes_group_ids")
if isinstance(lgs, list) and lane_index < len(lgs):
for_lane = lgs[lane_index]
@@ -234,7 +239,7 @@ def _resolve_step_device_names(
) -> List[str]:
z_names, z_macs = _compute_zone_targets(zone_doc, devices, groups)
if not step_group_ids:
return list(z_names)
return []
zone_mac_set = {m for m in (_norm_mac(x) for x in z_macs) if m}
zone_name_by_mac: Dict[str, str] = {}
for i, m in enumerate(z_macs):
@@ -273,12 +278,29 @@ def _lane_has_non_empty_lanes_group_ids(sequence_doc: Dict[str, Any], lane_index
return any(x is not None and str(x).strip() for x in for_lane)
def _partition_devices_for_lane(
num_lanes: int,
*,
lane_has_own_groups: bool,
step_group_ids: List[str],
) -> bool:
"""Split zone devices across lanes only when lanes lack explicit group targeting."""
if num_lanes <= 1:
return False
if lane_has_own_groups:
return False
# No lane groups (whole zone): every lane uses all zone / zone-group devices.
if not step_group_ids:
return False
return False
def _split_device_names_for_lane(
all_names: List[str],
lane_index: int,
num_lanes: int,
*,
partition_shared_zone: bool = True,
partition_shared_zone: bool = False,
) -> List[str]:
names = [n for n in all_names if n and str(n).strip()]
if num_lanes <= 1 or not partition_shared_zone:
@@ -368,15 +390,20 @@ def _resolve_lane_device_names(lane_index: int, ctx: Dict[str, Any]) -> List[str
lane = lanes[lane_index] if 0 <= lane_index < len(lanes) else []
if not lane:
return []
gids = _group_ids_for_lane_step(sequence_doc, lane[0], lane_index, num_lanes)
gids = _group_ids_for_lane_step(
sequence_doc, lane[0], lane_index, num_lanes, zone_doc=zone_doc
)
device_names = _resolve_step_device_names(
zone_doc, gids, devices, groups, sequence_doc=sequence_doc
)
lane_own = _lane_has_non_empty_lanes_group_ids(sequence_doc, lane_index)
return _split_device_names_for_lane(
device_names,
lane_index,
num_lanes,
partition_shared_zone=not _lane_has_non_empty_lanes_group_ids(sequence_doc, lane_index),
partition_shared_zone=_partition_devices_for_lane(
num_lanes, lane_has_own_groups=lane_own, step_group_ids=gids
),
)
@@ -545,16 +572,19 @@ def _union_macs_for_sequence(ctx: Dict[str, Any]) -> List[str]:
for step in lane:
if not isinstance(step, dict):
continue
gids = _group_ids_for_lane_step(sequence_doc, step, lane_index, num_lanes)
gids = _group_ids_for_lane_step(
sequence_doc, step, lane_index, num_lanes, zone_doc=zone_doc
)
device_names = _resolve_step_device_names(
zone_doc, gids, devices, groups, sequence_doc=sequence_doc
)
lane_own = _lane_has_non_empty_lanes_group_ids(sequence_doc, lane_index)
device_names = _split_device_names_for_lane(
device_names,
lane_index,
num_lanes,
partition_shared_zone=not _lane_has_non_empty_lanes_group_ids(
sequence_doc, lane_index
partition_shared_zone=_partition_devices_for_lane(
num_lanes, lane_has_own_groups=lane_own, step_group_ids=gids
),
)
if gids and not device_names:
@@ -563,10 +593,7 @@ def _union_macs_for_sequence(ctx: Dict[str, Any]) -> List[str]:
if m and m not in seen:
seen.add(m)
out.append(m)
if out:
return out
_, z_macs = _compute_zone_targets(zone_doc, devices, groups)
return list(z_macs)
return out
def _coerce_loop(sequence_doc: Dict[str, Any]) -> bool:
@@ -641,15 +668,22 @@ async def _send_lane(
display_preset = _display_preset_for_step(preset_id, presets_map, palette_colors)
if not display_preset:
return
gids = _group_ids_for_lane_step(sequence_doc, step, lane_index, int(ctx["num_lanes"]))
device_names = _resolve_step_device_names(
ctx["zone_doc"], gids, devices, ctx["groups"], sequence_doc=sequence_doc
num_lanes = int(ctx["num_lanes"])
zone_doc = ctx["zone_doc"]
gids = _group_ids_for_lane_step(
sequence_doc, step, lane_index, num_lanes, zone_doc=zone_doc
)
device_names = _resolve_step_device_names(
zone_doc, gids, devices, ctx["groups"], sequence_doc=sequence_doc
)
lane_own = _lane_has_non_empty_lanes_group_ids(sequence_doc, lane_index)
device_names = _split_device_names_for_lane(
device_names,
lane_index,
int(ctx["num_lanes"]),
partition_shared_zone=not _lane_has_non_empty_lanes_group_ids(sequence_doc, lane_index),
num_lanes,
partition_shared_zone=_partition_devices_for_lane(
num_lanes, lane_has_own_groups=lane_own, step_group_ids=gids
),
)
if gids and not device_names:
return