MESAVAULT/40-clients
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ucepsa): add read-only Shop Floor session probe

Browse Source
This commit is contained in:
Victor Fraile Garcia 2026-07-17 09:07:05 +02:00
parent 4586e29afe
commit 36d67d99b0
3 changed files with 790 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
ucepsa/edge-oee-demo/deploy/ucepsa-local/docker-compose.shopfloor-session-probe.yml Normal file
View File

@ -0,0 +1,35 @@
services:
mv_ucepsa_shopfloor_session_probe:
image: edge-oee-ucepsa-mv_ucepsa_edge_oee_mqtt_pg_sink
container_name: mv_ucepsa_shopfloor_session_probe
profiles:
- diagnostic
env_file:
- .env
environment:
ODOO_DB_OVERRIDE: ucepsa_prod
PYTHONUNBUFFERED: "1"
PYTHONDONTWRITEBYTECODE: "1"
volumes:
- ./:/app:ro
- /tmp:/probe-output
working_dir: /app
command:
- python
- tools/odoo_shopfloor_session_probe.py
- --machine
- CORT-00
- --watch
- --duration-minutes
- "30"
- --interval
- "2"
- --output
- /probe-output/odoo_shopfloor_session_probe.jsonl
networks:
- mv_ucepsa_net
networks:
mv_ucepsa_net:
external: true
name: mv_ucepsa_net

93
ucepsa/edge-oee-demo/docs/runbooks/092_odoo_shopfloor_session_discovery.md Normal file
View File

@ -0,0 +1,93 @@
# UCEPSA — Descubrimiento de sesiones Odoo Shop Floor v0.1
## Objetivo
Identificar, sin escribir en Odoo, qué registros y campos cambian cuando una operaria:
1. se identifica en Shop Floor;
2. inicia una orden en una cortadora;
3. pausa la orden;
4. la reanuda;
5. termina o abandona la orden;
6. es sustituida por otra operaria.
El resultado servirá para diseñar el publicador productivo de sesiones Shop Floor de MESAVAULT.
## Hipótesis técnica que se debe verificar
La fuente candidata principal es `mrp.workcenter.productivity`.
Una sesión activa debería aparecer como una fila con:
- `workorder_id`;
- `workcenter_id`;
- `date_start`;
- `date_end` vacío mientras sigue activa;
- `user_id` u otro campo de identidad;
- una pérdida/tipo compatible con tiempo productivo.
No se dará esta hipótesis por válida hasta observarla en UCEPSA PROD.
## Seguridad
La sonda:
- solo usa XML-RPC de lectura;
- no invoca `create`, `write`, `unlink` ni botones de Odoo;
- no publica MQTT;
- no modifica PostgreSQL;
- no altera balizas;
- no modifica Shop Floor.
## Secuencia recomendada
### 1. Descubrimiento estático
Ejecutar `--discover` y conservar la salida.
### 2. Instantánea inicial
Con CORT-00 sin una prueba en curso, ejecutar `--snapshot`.
### 3. Vigilancia
Iniciar `--watch` durante 30 minutos.
### 4. Acciones en Shop Floor
Anotar la hora local exacta de cada acción:
- T0: operaria se identifica;
- T1: abre CORT-00;
- T2: inicia la orden;
- T3: pausa;
- T4: reanuda;
- T5: finaliza;
- T6: otra operaria inicia o se incorpora, si el flujo lo permite.
Esperar al menos 20–30 segundos entre acciones para distinguirlas con claridad.
### 5. Cierre
Detener la sonda o esperar a que termine y copiar:
- salida de descubrimiento;
- instantánea inicial;
- JSONL de vigilancia;
- lista manual T0–T6;
- orden, producto y máquina utilizados.
## Criterios de aceptación
Se considera fuente oficial viable si se puede reconstruir de forma inequívoca:
- operaria;
- máquina;
- orden de trabajo;
- orden de fabricación;
- inicio;
- pausa/cierre;
- reanudación;
- relevo.
Debe confirmarse además que una sesión abierta tiene un estado detectable y que una pausa o fin deja una marca histórica, no solo un estado actual.

662
ucepsa/edge-oee-demo/tools/odoo_shopfloor_session_probe.py Normal file
View File

@ -0,0 +1,662 @@
#!/usr/bin/env python3
"""
Read-only probe for Odoo Shop Floor operator/work-order timing.
Purpose:
- Discover the actual Odoo models/fields available in UCEPSA PROD.
- Observe which records change when an operator signs in, starts, pauses,
resumes, and finishes a work order in Shop Floor.
- Produce a JSONL audit trail without writing anything to Odoo.
This script performs only XML-RPC reads.
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import time
import xmlrpc.client
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple
MACHINE_MAP = {
"CORT-00": "Cortadora 0",
"CORT-01": "Cortadora 01",
"CORT-02": "Cortadora 02",
}
RELEVANT_FIELDS = {
"mrp.workcenter.productivity": [
"id",
"display_name",
"description",
"workorder_id",
"workcenter_id",
"date_start",
"date_end",
"duration",
"user_id",
"employee_id",
"loss_id",
"loss_type",
"company_id",
"create_date",
"write_date",
],
"mrp.workorder": [
"id",
"display_name",
"name",
"state",
"workcenter_id",
"production_id",
"product_id",
"date_start",
"date_finished",
"duration",
"duration_expected",
"working_user_ids",
"last_working_user_id",
"is_user_working",
"time_ids",
"create_date",
"write_date",
],
"res.users": [
"id",
"display_name",
"name",
"login",
"active",
"employee_id",
"employee_ids",
"create_date",
"write_date",
],
"hr.employee": [
"id",
"display_name",
"name",
"active",
"user_id",
"create_date",
"write_date",
],
"mrp.workcenter.productivity.loss": [
"id",
"display_name",
"name",
"loss_type",
"manual",
"sequence",
"create_date",
"write_date",
],
}
def read_env(path: str = ".env") -> Dict[str, str]:
env: Dict[str, str] = {}
p = Path(path)
if not p.exists():
return env
for raw in p.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
env[key.strip()] = value.strip().strip('"').strip("'")
return env
def utc_now() -> datetime:
return datetime.now(timezone.utc)
def odoo_datetime(dt: datetime) -> str:
return dt.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
def json_default(value: Any) -> str:
return str(value)
def connect_odoo(env: Dict[str, str]) -> Dict[str, Any]:
required = ["ODOO_URL", "ODOO_DB", "ODOO_USER", "ODOO_PASSWORD"]
missing = [key for key in required if not (os.getenv(key) or env.get(key))]
if missing:
raise RuntimeError(f"Faltan variables Odoo: {', '.join(missing)}")
url = (os.getenv("ODOO_URL") or env["ODOO_URL"]).rstrip("/")
db = os.getenv("ODOO_DB_OVERRIDE") or os.getenv("ODOO_DB") or env["ODOO_DB"]
user = os.getenv("ODOO_USER") or env["ODOO_USER"]
password = os.getenv("ODOO_PASSWORD") or env["ODOO_PASSWORD"]
common = xmlrpc.client.ServerProxy(
f"{url}/xmlrpc/2/common",
allow_none=True,
)
uid = common.authenticate(db, user, password, {})
if not uid:
raise RuntimeError("Autenticación Odoo fallida")
models = xmlrpc.client.ServerProxy(
f"{url}/xmlrpc/2/object",
allow_none=True,
)
return {
"url": url,
"db": db,
"user": user,
"password": password,
"uid": uid,
"models": models,
}
def execute_kw(
odoo: Dict[str, Any],
model: str,
method: str,
args: List[Any],
kwargs: Optional[Dict[str, Any]] = None,
) -> Any:
return odoo["models"].execute_kw(
odoo["db"],
odoo["uid"],
odoo["password"],
model,
method,
args,
kwargs or {},
)
def model_fields(odoo: Dict[str, Any], model: str) -> Dict[str, Any]:
try:
return execute_kw(
odoo,
model,
"fields_get",
[],
{"attributes": ["string", "type", "relation", "required", "readonly"]},
)
except xmlrpc.client.Fault as exc:
print(
f"[WARN] No se puede consultar fields_get de {model}: {exc.faultString}",
file=sys.stderr,
)
return {}
def existing_fields(
odoo: Dict[str, Any],
model: str,
wanted: Iterable[str],
) -> List[str]:
fields = model_fields(odoo, model)
return [field for field in wanted if field in fields]
def search_read(
odoo: Dict[str, Any],
model: str,
domain: List[Any],
wanted_fields: Iterable[str],
*,
limit: int = 200,
order: str = "id asc",
) -> List[Dict[str, Any]]:
fields = existing_fields(odoo, model, wanted_fields)
if not fields:
return []
try:
return execute_kw(
odoo,
model,
"search_read",
[domain],
{
"fields": fields,
"limit": limit,
"order": order,
},
)
except xmlrpc.client.Fault as exc:
print(
f"[WARN] No se puede leer {model}: {exc.faultString}",
file=sys.stderr,
)
return []
def m2o_id(value: Any) -> Optional[int]:
if isinstance(value, list) and value:
try:
return int(value[0])
except (TypeError, ValueError):
return None
if isinstance(value, int):
return value
return None
def get_workcenters(
odoo: Dict[str, Any],
machine_filter: Optional[str],
) -> Tuple[Dict[str, Dict[str, Any]], List[int]]:
selected = (
{machine_filter: MACHINE_MAP[machine_filter]}
if machine_filter
else MACHINE_MAP
)
result: Dict[str, Dict[str, Any]] = {}
ids: List[int] = []
for machine_id, workcenter_name in selected.items():
rows = search_read(
odoo,
"mrp.workcenter",
[("name", "=", workcenter_name)],
["id", "name", "display_name", "code", "active"],
limit=5,
order="id asc",
)
if len(rows) != 1:
result[machine_id] = {
"machine_id": machine_id,
"expected_name": workcenter_name,
"status": "NOT_FOUND" if not rows else "AMBIGUOUS",
"matches": rows,
}
continue
row = rows[0]
row["machine_id"] = machine_id
row["status"] = "OK"
result[machine_id] = row
ids.append(int(row["id"]))
return result, ids
def discover(
odoo: Dict[str, Any],
workcenters: Dict[str, Dict[str, Any]],
) -> Dict[str, Any]:
output: Dict[str, Any] = {
"captured_at": utc_now().isoformat(),
"odoo": {
"url": odoo["url"],
"db": odoo["db"],
"user": odoo["user"],
"uid": odoo["uid"],
},
"workcenters": workcenters,
"models": {},
}
for model, wanted in RELEVANT_FIELDS.items():
all_fields = model_fields(odoo, model)
output["models"][model] = {
"accessible": bool(all_fields),
"relevant_fields_present": {
field: all_fields[field]
for field in wanted
if field in all_fields
},
"relevant_fields_missing": [
field for field in wanted if field not in all_fields
],
}
return output
def collect_snapshot(
odoo: Dict[str, Any],
workcenter_ids: List[int],
lookback_hours: float,
) -> Dict[str, List[Dict[str, Any]]]:
start = odoo_datetime(utc_now() - timedelta(hours=lookback_hours))
productivity_domain: List[Any] = [
("workcenter_id", "in", workcenter_ids),
"|",
("date_end", "=", False),
("date_start", ">=", start),
]
productivity_rows = search_read(
odoo,
"mrp.workcenter.productivity",
productivity_domain,
RELEVANT_FIELDS["mrp.workcenter.productivity"],
limit=1000,
order="id asc",
)
workorder_domain: List[Any] = [
("workcenter_id", "in", workcenter_ids),
("state", "in", ["ready", "progress", "done"]),
"|",
("write_date", ">=", start),
("state", "=", "progress"),
]
workorder_rows = search_read(
odoo,
"mrp.workorder",
workorder_domain,
RELEVANT_FIELDS["mrp.workorder"],
limit=1000,
order="id asc",
)
user_ids = sorted(
{
user_id
for row in productivity_rows
for user_id in [m2o_id(row.get("user_id"))]
if user_id
}
|
{
int(user_id)
for row in workorder_rows
for user_id in (row.get("working_user_ids") or [])
if isinstance(user_id, int)
}
)
users = (
search_read(
odoo,
"res.users",
[("id", "in", user_ids)],
RELEVANT_FIELDS["res.users"],
limit=1000,
order="id asc",
)
if user_ids
else []
)
employee_ids = sorted(
{
employee_id
for row in users
for employee_id in (
([m2o_id(row.get("employee_id"))] if row.get("employee_id") else [])
+ [
int(item)
for item in (row.get("employee_ids") or [])
if isinstance(item, int)
]
)
if employee_id
}
)
employees = (
search_read(
odoo,
"hr.employee",
[("id", "in", employee_ids)],
RELEVANT_FIELDS["hr.employee"],
limit=1000,
order="id asc",
)
if employee_ids
else []
)
loss_ids = sorted(
{
loss_id
for row in productivity_rows
for loss_id in [m2o_id(row.get("loss_id"))]
if loss_id
}
)
losses = (
search_read(
odoo,
"mrp.workcenter.productivity.loss",
[("id", "in", loss_ids)],
RELEVANT_FIELDS["mrp.workcenter.productivity.loss"],
limit=1000,
order="id asc",
)
if loss_ids
else []
)
return {
"mrp.workcenter.productivity": productivity_rows,
"mrp.workorder": workorder_rows,
"res.users": users,
"hr.employee": employees,
"mrp.workcenter.productivity.loss": losses,
}
def normalize_record(record: Dict[str, Any]) -> str:
return json.dumps(
record,
sort_keys=True,
ensure_ascii=False,
default=json_default,
)
def index_snapshot(
snapshot: Dict[str, List[Dict[str, Any]]],
) -> Dict[Tuple[str, int], str]:
indexed: Dict[Tuple[str, int], str] = {}
for model, rows in snapshot.items():
for row in rows:
row_id = row.get("id")
if isinstance(row_id, int):
indexed[(model, row_id)] = normalize_record(row)
return indexed
def append_jsonl(path: Path, payload: Dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as handle:
handle.write(
json.dumps(
payload,
ensure_ascii=False,
default=json_default,
)
+ "\n"
)
def emit_event(
event_type: str,
model: str,
record: Dict[str, Any],
output_path: Path,
) -> None:
payload = {
"captured_at": utc_now().isoformat(),
"event_type": event_type,
"model": model,
"record": record,
}
print(json.dumps(payload, ensure_ascii=False, default=json_default), flush=True)
append_jsonl(output_path, payload)
def watch(
odoo: Dict[str, Any],
workcenter_ids: List[int],
*,
lookback_hours: float,
interval_s: float,
duration_minutes: float,
output_path: Path,
) -> None:
snapshot = collect_snapshot(odoo, workcenter_ids, lookback_hours)
previous = index_snapshot(snapshot)
for model, rows in snapshot.items():
for row in rows:
emit_event("BASELINE", model, row, output_path)
print(
f"[INFO] Vigilancia iniciada. interval={interval_s}s "
f"duration={duration_minutes}min output={output_path}",
flush=True,
)
deadline = time.monotonic() + duration_minutes * 60
while time.monotonic() < deadline:
time.sleep(interval_s)
current_snapshot = collect_snapshot(odoo, workcenter_ids, lookback_hours)
current = index_snapshot(current_snapshot)
for model, rows in current_snapshot.items():
for row in rows:
key = (model, int(row["id"]))
serialized = normalize_record(row)
if key not in previous:
emit_event("CREATED", model, row, output_path)
elif previous[key] != serialized:
emit_event("UPDATED", model, row, output_path)
removed = set(previous) - set(current)
for model, row_id in sorted(removed):
emit_event(
"NO_LONGER_IN_WINDOW",
model,
{"id": row_id},
output_path,
)
previous = current
print("[INFO] Vigilancia finalizada.", flush=True)
def main() -> int:
parser = argparse.ArgumentParser(
description="Sonda read-only de sesiones Odoo Shop Floor"
)
parser.add_argument(
"--env-file",
default=".env",
help="Ruta del .env de UCEPSA",
)
parser.add_argument(
"--machine",
choices=sorted(MACHINE_MAP),
help="Limitar la prueba a una cortadora",
)
parser.add_argument(
"--discover",
action="store_true",
help="Mostrar modelos y campos disponibles",
)
parser.add_argument(
"--snapshot",
action="store_true",
help="Mostrar una instantánea de registros",
)
parser.add_argument(
"--watch",
action="store_true",
help="Vigilar cambios y escribir JSONL",
)
parser.add_argument(
"--lookback-hours",
type=float,
default=24.0,
)
parser.add_argument(
"--interval",
type=float,
default=2.0,
)
parser.add_argument(
"--duration-minutes",
type=float,
default=30.0,
)
parser.add_argument(
"--output",
default="/tmp/odoo_shopfloor_session_probe.jsonl",
)
args = parser.parse_args()
if not (args.discover or args.snapshot or args.watch):
parser.error("Indica --discover, --snapshot o --watch")
env = read_env(args.env_file)
odoo = connect_odoo(env)
workcenters, workcenter_ids = get_workcenters(odoo, args.machine)
if not workcenter_ids:
print(json.dumps(workcenters, indent=2, ensure_ascii=False))
raise RuntimeError("No se ha encontrado ningún centro de trabajo válido")
print(
f"[INFO] Odoo conectado: url={odoo['url']} db={odoo['db']} "
f"user={odoo['user']} uid={odoo['uid']}",
flush=True,
)
if args.discover:
print(
json.dumps(
discover(odoo, workcenters),
indent=2,
ensure_ascii=False,
default=json_default,
)
)
if args.snapshot:
print(
json.dumps(
{
"captured_at": utc_now().isoformat(),
"workcenters": workcenters,
"snapshot": collect_snapshot(
odoo,
workcenter_ids,
args.lookback_hours,
),
},
indent=2,
ensure_ascii=False,
default=json_default,
)
)
if args.watch:
watch(
odoo,
workcenter_ids,
lookback_hours=args.lookback_hours,
interval_s=args.interval,
duration_minutes=args.duration_minutes,
output_path=Path(args.output),
)
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except KeyboardInterrupt:
print("\n[INFO] Interrumpido por usuario.", file=sys.stderr)
raise SystemExit(130)
except Exception as exc:
print(f"[ERROR] {exc}", file=sys.stderr)
raise SystemExit(1)