led-controller/src/models/sequence.py

from models.model import Model


class Sequence(Model):
    def load(self):
        super().load()
        self._migrate_after_load()

    def _migrate_after_load(self):
        try:
            from models.profile import Profile

            profiles = Profile()
            profile_list = profiles.list()
            default_profile_id = profile_list[0] if profile_list else None
        except Exception:
            default_profile_id = None

        changed = False
        for _sid, doc in list(self.items()):
            if not isinstance(doc, dict):
                continue
            if not isinstance(doc.get("steps"), list):
                presets = doc.get("presets")
                if isinstance(presets, list) and presets:
                    doc["steps"] = [
                        {"preset_id": str(p), "group_ids": []} for p in presets
                    ]
                else:
                    doc["steps"] = []
                changed = True
            if "step_duration_ms" not in doc:
                dur = doc.get("sequence_duration")
                doc["step_duration_ms"] = (
                    int(dur) if isinstance(dur, (int, float)) else 3000
                )
                changed = True
            if "loop" not in doc:
                doc["loop"] = bool(doc.get("sequence_loop", False))
                changed = True
            if "name" not in doc:
                doc["name"] = str(doc.get("group_name") or "")
                changed = True
            if "profile_id" not in doc and default_profile_id is not None:
                doc["profile_id"] = str(default_profile_id)
                changed = True
            if not isinstance(doc.get("lanes"), list):
                steps = doc.get("steps")
                if isinstance(steps, list) and steps:
                    doc["lanes"] = [list(steps)]
                else:
                    doc["lanes"] = [[]]
                changed = True
            if "group_ids" not in doc or not isinstance(doc.get("group_ids"), list):
                doc["group_ids"] = []
                changed = True
            if doc.get("advance_mode") != "beats":
                doc["advance_mode"] = "beats"
                changed = True
            if "simulated_bpm" not in doc:
                doc["simulated_bpm"] = 120
                changed = True
            else:
                try:
                    sb = int(float(doc["simulated_bpm"]))
                    doc["simulated_bpm"] = max(30, min(300, sb))
                except (TypeError, ValueError):
                    doc["simulated_bpm"] = 120
                    changed = True
            if "sequence_transition" not in doc:
                doc["sequence_transition"] = 500
                changed = True
            # Ensure each step has beats (beat-based advance); default 1
            for lane in doc.get("lanes") or []:
                if not isinstance(lane, list):
                    continue
                for step in lane:
                    if not isinstance(step, dict):
                        continue
                    if "beats" not in step:
                        step["beats"] = 1
                        changed = True
            # Per-lane group ids (parallel to ``lanes``)
            lanes_list = [x for x in (doc.get("lanes") or []) if isinstance(x, list)]
            n_lanes = len(lanes_list)
            lg = doc.get("lanes_group_ids")
            if n_lanes and (not isinstance(lg, list) or len(lg) != n_lanes):
                shared = doc.get("group_ids") if isinstance(doc.get("group_ids"), list) else []
                shared_s = [str(x).strip() for x in shared if x is not None and str(x).strip()]
                if n_lanes == 1 and lanes_list[0]:
                    first = lanes_list[0][0] if isinstance(lanes_list[0][0], dict) else {}
                    step_g = (
                        first.get("group_ids")
                        if isinstance(first.get("group_ids"), list)
                        else []
                    )
                    step_s = [
                        str(x).strip() for x in step_g if x is not None and str(x).strip()
                    ]
                    doc["lanes_group_ids"] = [step_s if step_s else list(shared_s)]
                else:
                    doc["lanes_group_ids"] = [list(shared_s) for _ in range(n_lanes)]
                changed = True
        if changed:
            self.save()

    def create(self, profile_id=None):
        next_id = self.get_next_id()
        self[next_id] = {
            "name": "",
            "profile_id": str(profile_id) if profile_id is not None else None,
            "group_ids": [],
            "lanes": [[]],
            "lanes_group_ids": [[]],
            "advance_mode": "beats",
            "steps": [],
            "step_duration_ms": 3000,
            "simulated_bpm": 120,
            "sequence_transition": 500,
            "loop": True,
        }
        self.save()
        return next_id

    def read(self, id):
        id_str = str(id)
        return self.get(id_str, None)

    def update(self, id, data):
        id_str = str(id)
        if id_str not in self:
            return False
        if not isinstance(data, dict):
            return False
        data = dict(data)
        steps = data.get("steps")
        lanes = data.get("lanes")
        if isinstance(steps, list) and steps:
            lanes_ok = (
                isinstance(lanes, list)
                and lanes
                and any(isinstance(x, list) and len(x) > 0 for x in lanes)
            )
            if not lanes_ok:
                data["lanes"] = [list(steps)]
        self[id_str].update(data)
        self.save()
        return True

    def delete(self, id):
        id_str = str(id)
        if id_str not in self:
            return False
        self.pop(id_str)
        self.save()
        return True

    def list(self):
        return list(self.keys())