Mini SDK para trabajar con conceptrones CFT (Color, Forma, Tono) en Python: dataclasses, validación, API con FastAPI y conexión con modelos de IA.
Un conceptrón en Python se representa así:
from dataclasses import dataclass from typing import List, Tuple @dataclass class Conceptron: C: str F: str T: Tuple[float, float] chain: List[Conceptron] = [ Conceptron("red", "triangle", (300, 500)), Conceptron("blue","hexagon", (200, 350)), ]
from _future_ import annotations from dataclasses import dataclass, asdict from typing import List, Tuple, Dict, Any COLORS: List[str] = [ "red","orange","yellow","green","cyan","blue", "purple","pink","white","grey","black" ] SHAPES: List[str] = [ "circle","square","triangle","hexagon","star","bar","wave" ] @dataclass class Conceptron: C: str F: str T: Tuple[float, float] # (minHz, maxHz) def validate(self) -> None: if self.C not in COLORS: raise ValueError(f"Color no permitido: {self.C}") if self.F not in SHAPES: raise ValueError(f"Forma no permitida: {self.F}") if len(self.T) != 2: raise ValueError("T debe tener forma (min,max)") t_min, t_max = self.T if not (isinstance(t_min,(int,float)) and isinstance(t_max,(int,float))): raise ValueError("T debe contener números (Hz)") if t_min >= t_max: raise ValueError("T: min debe ser < max") @dataclass class Pattern: id: str descripcion: str chain: List[Conceptron] def validate(self) -> None: if not self.id: raise ValueError("id requerido") if not self.chain: raise ValueError("chain no puede estar vacío") for c in self.chain: c.validate() def to_dict(self) -> Dict[str, Any]: return { "id": self.id, "descripcion": self.descripcion, "chain": [asdict(c) for c in self.chain], }
from cft_core import Conceptron, Pattern riesgo_alto = Pattern( id="riesgo_fiscal_alto", descripcion="Riesgo fiscal alto detectado", chain=[ Conceptron("red", "triangle", (300, 500)), Conceptron("blue", "hexagon", (200, 350)), ] ) riesgo_alto.validate() print(riesgo_alto.to_dict())
VOCABULARIO: Dict[str, Pattern] = { "ok": Pattern( id="ok", descripcion="Estado normal", chain=[Conceptron("green","circle",(600,800))] ), "warning": Pattern( id="warning", descripcion="Atención", chain=[Conceptron("yellow","triangle",(400,700))] ), "danger": riesgo_alto, }
import json from cft_core import Pattern def save_patterns(path: str, patterns: Dict[str, Pattern]) -> None: data = {k: v.to_dict() for k, v in patterns.items()} with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2)
from typing import Dict from cft_core import Conceptron, Pattern def load_patterns(path: str) -> Dict[str, Pattern]: with open(path, "r", encoding="utf-8") as f: data = json.load(f) patterns: Dict[str, Pattern] = {} for key, p in data.items(): chain = [ Conceptron(C=c["C"], F=c["F"], T=tuple(c["T"])) for c in p["chain"] ] pat = Pattern(id=p["id"], descripcion=p["descripcion"], chain=chain) pat.validate() patterns[key] = pat return patterns
# api.py from typing import List, Tuple, Dict from fastapi import FastAPI, HTTPException from pydantic import BaseModel, field_validator from cft_core import COLORS, SHAPES, VOCABULARIO class ConceptronDTO(BaseModel): C: str F: str T: Tuple[float, float] @field_validator("C") def check_color(cls, v): if v not in COLORS: raise ValueError("Color no permitido") return v @field_validator("F") def check_shape(cls, v): if v not in SHAPES: raise ValueError("Forma no permitida") return v @field_validator("T") def check_t(cls, v): if len(v) != 2 or v[0] >= v[1]: raise ValueError("T debe ser [min,max] con min < max") return v class PatternDTO(BaseModel): id: str descripcion: str chain: List[ConceptronDTO] app = FastAPI(title="OpenSymbolic CFT API") @app.get("/patterns/{pattern_id}", response_model=PatternDTO) def get_pattern(pattern_id: str): pat = VOCABULARIO.get(pattern_id) if not pat: raise HTTPException(status_code=404, detail="Pattern not found") return pat.to_dict()
from datetime import datetime from fastapi import Body class PacketDTO(BaseModel): ver: int ts: int src: str dst: str payload: List[ConceptronDTO] meta: Dict[str, str] | None = None @app.post("/osnet/ingest") def ingest_packet(packet: PacketDTO = Body(...)): # Aquí podrías guardar en BD, log, etc. ts = datetime.fromtimestamp(packet.ts / 1000.0) print(f"[{ts}] Packet from {packet.src} to {packet.dst}") print(f"Meta: {packet.meta}") print(f"Conceptrones recibidos: {len(packet.payload)}") return {"ok": True}
Ejemplo genérico llamando a un proveedor de IA (pseudo-código) y devolviendo un Pattern.
import json from typing import Any from cft_core import Conceptron, Pattern SYSTEM_PROMPT = """ Eres un traductor simbólico de OpenSymbolic. Un conceptrón se define como: - C: color en inglés, de esta lista fija: ["red","orange","yellow","green","cyan","blue","purple","pink","white","grey","black"] - F: forma en inglés, de esta lista fija: ["circle","square","triangle","hexagon","star","bar","wave"] - T: rango de tono en Hz, como [min,max]. Devuelve SOLO un objeto JSON con: { "id": "string", "descripcion": "string", "chain": [ { "C": "color", "F": "shape", "T": [minHz, maxHz] } ] } """ def call_llm(messages: list[dict[str, str]]) -> dict[str, Any]: """ Envía mensajes al proveedor de IA y devuelve un dict con la respuesta. Implementa aquí la llamada real (OpenAI, Vertex, etc.). """ raise NotImplementedError def text_to_pattern(text: str) -> Pattern: messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": f'Frase: "{text}"'} ] resp = call_llm(messages) raw = resp["content"] data = json.loads(raw) chain = [ Conceptron(C=c["C"], F=c["F"], T=tuple(c["T"])) for c in data["chain"] ] pat = Pattern( id=data["id"], descripcion=data["descripcion"], chain=chain ) pat.validate() return pat
from fastapi import APIRouter from cft_core import Pattern from api import PatternDTO from ia_integration import text_to_pattern router = APIRouter(prefix="/ai", tags=["ai"]) @router.post("/text-to-pattern", response_model=PatternDTO) def ai_text_to_pattern(payload: dict): text = payload.get("text","").strip() if not text: raise HTTPException(status_code=400, detail="text requerido") pat: Pattern = text_to_pattern(text) return pat.to_dict()
# cli_demo.py import argparse from cft_core import VOCABULARIO def main(): parser = argparse.ArgumentParser() parser.add_argument("pattern_id", help="ID del patrón (ok, warning, danger...)") args = parser.parse_args() pat = VOCABULARIO.get(args.pattern_id) if not pat: print("Patrón no encontrado") return print("ID:", pat.id) print("Descripción:", pat.descripcion) for i, c in enumerate(pat.chain, start=1): print(f" {i}. C={c.C}, F={c.F}, T={c.T[0]}-{c.T[1]} Hz") if _name_ == "_main_": main()