run_example.py

#!/usr/bin/env python3
"""Linked two-stage example for the AI-Assisted Production Scheduling service (EnerTEF).

Default (offline): read the four example files in ./example and print a short summary
of both stages. Needs only the Python standard library.

--live: send the example inputs to the hosted service and print the same summary.
Live access is available on request through the EnerTEF project or LMS, University
of Patras (https://lms.mech.upatras.gr). An account and access instructions come with
the approval. The live mode needs the 'requests' package and an access token in the
TEF_IND_ACCESS_TOKEN environment variable; the token is never read from the command
line. See README.md, "Live access".

Python 3.8 or later.
"""

import argparse
import datetime
import json
import os
import sys
import textwrap
import time

SERVICE_TITLE = "AI-Assisted Production Scheduling"
BASE_URL = "https://api.scheduler.tef-ind.eu"

PLAN_PATH = "/api/scheduling/planFromJSONNODB"          # stage 1: MIP planning
SCHEDULE_PATH = "/api/scheduling/scheduleFromJSONNODB"  # stage 2: IMPACT heuristic
TOKEN_ENV = "TEF_IND_ACCESS_TOKEN"  # the token is read from the environment only
ACCESS_NOTE = ("Live access is available on request through the EnerTEF project or LMS, "
               "University of Patras (https://lms.mech.upatras.gr). An account and access "
               "instructions come with the approval.")
MAX_UPLOAD_BYTES = 2 * 1024 * 1024  # the service rejects inputs above 2 MB
CONNECT_TIMEOUT_S = 15
READ_TIMEOUT_S = 120  # the proxy in front of the service gives up after about 60 s
PARAM_BOUNDS = {"dh": (1, 5, 2), "mna": (1, 1000, 100), "sr": (1, 100, 10)}  # min, max, default
ROWS_SHOWN = 5

EXAMPLE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "example")
STAGE1_INPUT = "stage1_mip_input.json"
STAGE1_OUTPUT = "stage1_mip_output.json"
STAGE2_INPUT = "stage2_heuristic_input.json"
STAGE2_OUTPUT = "stage2_heuristic_output.json"
ALL_FILES = (STAGE1_INPUT, STAGE1_OUTPUT, STAGE2_INPUT, STAGE2_OUTPUT)


# ---------------------------------------------------------------------------
# Small helpers
# ---------------------------------------------------------------------------

def example_path(name):
    return os.path.join(EXAMPLE_DIR, name)


def missing_files(names):
    return [name for name in names if not os.path.isfile(example_path(name))]


def report_not_captured(missing):
    print("Example data not yet captured.")
    print("These files are missing from example/:")
    for name in missing:
        print("  - " + name)
    print("They are added after one run of the service has been captured on anonymized")
    print("demo data. See example/README.md.")


def load_json(name):
    with open(example_path(name), "r", encoding="utf-8") as handle:
        return json.load(handle)


def get_ci(obj, key, default=None):
    """Dict lookup that ignores the case of the key."""
    if not isinstance(obj, dict):
        return default
    if key in obj:
        return obj[key]
    wanted = key.lower()
    for name, value in obj.items():
        if isinstance(name, str) and name.lower() == wanted:
            return value
    return default


def as_list(value):
    if value is None:
        return []
    if isinstance(value, list):
        return value
    return [value]


def fmt_num(value):
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return str(value)
    if float(value).is_integer():
        return "{:,.0f}".format(value)
    return "{:,.2f}".format(value)


def fmt_time(moment):
    return moment.strftime("%Y-%m-%d %H:%M") if moment else "?"


def nested_get(tree, *keys):
    for key in keys:
        if not isinstance(tree, dict):
            return None
        tree = tree.get(key)
    return tree


def walk(tree, depth):
    """Yield (path, leaf) for every leaf of a dict nested exactly `depth` levels deep."""
    if depth == 0:
        yield (), tree
        return
    if isinstance(tree, dict):
        for key in sorted(tree, key=str):
            for path, leaf in walk(tree[key], depth - 1):
                yield (key,) + path, leaf


def unique_values(mapping):
    seen = set()
    if isinstance(mapping, dict):
        for value in mapping.values():
            for item in as_list(value):
                seen.add(str(item))
    return seen


# ---------------------------------------------------------------------------
# Summaries (used by both modes)
# ---------------------------------------------------------------------------

def summarize_stage1_input(data):
    print("Stage 1 input: MIP planning instance")
    orders = as_list(data.get("order_ids"))
    machines = as_list(data.get("machine_ids"))
    products = set(str(p) for p in as_list(data.get("product_ids")))
    plans = unique_values(data.get("product_process_plans"))
    jobs = unique_values(data.get("process_plan_jobs"))
    tasks = unique_values(data.get("job_tasks"))
    items = [(order, product, qty)
             for (order, product), qty in walk(data.get("product_quantity") or {}, 2)]
    print("  orders: %d, ordered items: %d, products and materials: %d"
          % (len(orders), len(items), len(products)))
    print("  process plans: %d, jobs: %d, tasks: %d, machines: %d"
          % (len(plans), len(jobs), len(tasks), len(machines)))
    for order, product, qty in items[:ROWS_SHOWN]:
        print("  order %s asks for %s units of product %s" % (order, fmt_num(qty), product))


def summarize_stage1_output(data):
    print("Stage 1 output: MIP plan")
    status = data.get("feasible") if isinstance(data, dict) else None
    print("  status: %s" % status)
    if status != "OPTIMAL":
        print("  no plan: the model has no proven optimal solution for this input.")
        return
    print("  objective (profit): %s" % fmt_num(data.get("objective_value")))
    usage = data.get("machine_usage") or {}
    print("  machine hours used: "
          + (", ".join("%s: %s h" % (m, fmt_num(h)) for m, h in sorted(usage.items())) or "none"))

    production = data.get("process_plan_production") or {}
    used_plans = [(path, qty) for path, qty in walk(production, 3)
                  if isinstance(qty, (int, float)) and qty > 0]
    for (order, product, plan), qty in used_plans[:ROWS_SHOWN]:
        print("  order %s, product %s: %s units through process plan %s"
              % (order, product, fmt_num(qty), plan))

    durations = data.get("task_duration") or {}
    used = set(path for path, _ in used_plans)
    chosen = []
    for (order, product, plan, job, task, machine), flag in walk(data.get("task_assignment") or {}, 6):
        if flag is True and (order, product, plan) in used:
            hours = nested_get(durations, order, product, plan, job, task)
            chosen.append((order, product, plan, job, task, machine, hours))
    print("  tasks assigned to machines in the plans used: %d" % len(chosen))
    for order, product, plan, job, task, machine, hours in chosen[:ROWS_SHOWN]:
        print("    order %s / plan %s / job %s / task %s -> machine %s, %s h"
              % (order, plan, job, task, machine, fmt_num(hours)))


def plan_date(data, prefix):
    try:
        return datetime.date(int(get_ci(data, prefix + "Year")),
                             int(get_ci(data, prefix + "Month")),
                             int(get_ci(data, prefix + "Day"))).isoformat()
    except (TypeError, ValueError):
        return "?"


def summarize_stage2_input(data):
    print("Stage 2 input: heuristic instance (built by the service from the stage 1 plan)")
    jobs = as_list(get_ci(get_ci(data, "jobs"), "job"))
    tasks = as_list(get_ci(get_ci(data, "tasks"), "task"))
    resources = as_list(get_ci(get_ci(data, "resources"), "resource"))
    options = as_list(get_ci(get_ci(data, "tasksuitableresources"), "tasksuitableresource"))
    precedences = as_list(get_ci(get_ci(data, "taskprecedenceconstraints"), "taskprecedenceconstraint"))
    periods = 0
    for resource in resources:
        availability = get_ci(get_ci(resource, "resourceavailability"), "nonworkingperiods")
        periods += len(as_list(get_ci(availability, "period")))
    print("  planning window: %s to %s"
          % (plan_date(data, "planStartDate"), plan_date(data, "planEndDate")))
    print("  jobs: %d, tasks: %d, resources: %d, task-resource options: %d"
          % (len(jobs), len(tasks), len(resources), len(options)))
    print("  precedence constraints: %d, non-working periods: %d" % (len(precedences), periods))
    return len(tasks)


def dispatch_time(assignment):
    stamp = get_ci(assignment, "timeofdispatch") or {}
    try:
        return datetime.datetime(int(get_ci(stamp, "year")), int(get_ci(stamp, "month")),
                                 int(get_ci(stamp, "day")), int(get_ci(stamp, "hour") or 0),
                                 int(get_ci(stamp, "minutes") or 0), int(get_ci(stamp, "seconds") or 0))
    except (TypeError, ValueError):
        return None


def summarize_stage2_output(data, expected_tasks=None):
    print("Stage 2 output: timed schedule")
    schedule = data
    if isinstance(data, dict) and "schedule" in data:
        # Shape returned by the live endpoint; the captured file holds the bare schedule.
        print("  parameters: dh=%s, mna=%s, sr=%s" % (data.get("dh"), data.get("mna"), data.get("sr")))
        schedule = data.get("schedule")
    rows = []
    for assignment in as_list(get_ci(get_ci(schedule, "assignments"), "assignment")):
        start = dispatch_time(assignment)
        millis = get_ci(assignment, "durationinmilliseconds") or 0
        end = start + datetime.timedelta(milliseconds=millis) if start else None
        rows.append((start, end, get_ci(get_ci(assignment, "task"), "id"),
                     get_ci(get_ci(assignment, "resource"), "id")))
    rows.sort(key=lambda row: (row[0] is None, row[0] or datetime.datetime.min))
    line = "  assignments: %d" % len(rows)
    if expected_tasks is not None:
        line += (" (all %d tasks scheduled)" % expected_tasks if len(rows) == expected_tasks
                 else " (the input has %d tasks)" % expected_tasks)
    print(line)
    starts = [row[0] for row in rows if row[0]]
    ends = [row[1] for row in rows if row[1]]
    if starts and ends:
        print("  first start: %s, last end: %s" % (fmt_time(min(starts)), fmt_time(max(ends))))
    for start, end, task, resource in rows[:ROWS_SHOWN]:
        print("    task %s on resource %s: %s -> %s" % (task, resource, fmt_time(start), fmt_time(end)))


# ---------------------------------------------------------------------------
# Offline mode
# ---------------------------------------------------------------------------

def run_offline():
    missing = missing_files(ALL_FILES)
    if missing:
        report_not_captured(missing)
        return 1
    print("%s: linked example (offline)" % SERVICE_TITLE)
    print()
    try:
        summarize_stage1_input(load_json(STAGE1_INPUT))
        print()
        summarize_stage1_output(load_json(STAGE1_OUTPUT))
        print()
        task_count = summarize_stage2_input(load_json(STAGE2_INPUT))
        print()
        summarize_stage2_output(load_json(STAGE2_OUTPUT), task_count)
    except (OSError, ValueError) as exc:
        print("Could not read the example files: %s" % exc)
        return 1
    return 0


# ---------------------------------------------------------------------------
# Live mode
# ---------------------------------------------------------------------------

def server_message(response):
    try:
        body = response.json()
    except ValueError:
        text = (response.text or "").strip()
        return "" if text.startswith("<") else text[:200]
    if isinstance(body, dict):
        for key in ("message", "error", "detail"):
            value = body.get(key)
            if isinstance(value, str) and value.strip():
                return value.strip()
    return ""


def explain_error(response, stage):
    status = response.status_code
    message = server_message(response)
    print("%s failed with HTTP %d." % (stage, status))
    if status == 400:
        print("The service rejected the input." + (" It said: " + message if message else ""))
    elif status == 401:
        print("Not signed in, or the token has expired. Get a fresh token as your access")
        print("instructions describe, and set %s again." % TOKEN_ENV)
    elif status == 403:
        print("You are signed in, but your account may not run optimizations.")
        print("Live access is granted on request: see README.md, 'Live access'.")
    elif status == 404:
        print("Endpoint not found. This deployment may not offer it yet.")
    elif status in (406, 415):
        print("The service did not accept the request format. This is likely a problem in this script.")
    elif status == 413:
        print("The input is larger than the service accepts (2 MB).")
    elif status == 429:
        wait = response.headers.get("Retry-After", "30")
        print("Another optimization is already running on this service, which runs one at a time.")
        print("Try again in about %s seconds. This script does not retry by itself." % wait)
    elif status == 504:
        print("The run took longer than the proxy allows (about 60 s). It may still be running on")
        print("the server and holding the single optimization slot, so the next request may get 429")
        print("until it ends. Use a smaller instance or lower dh/mna/sr.")
    elif status in (502, 503):
        print("The service is not reachable right now. Try again later.")
    elif message:
        print("The service said: " + message)


def post(session, url, **kwargs):
    import requests  # imported here so that the offline mode needs only the standard library
    try:
        return session.post(url, timeout=(CONNECT_TIMEOUT_S, READ_TIMEOUT_S), **kwargs)
    except requests.exceptions.SSLError:
        print("Could not verify the TLS certificate of %s." % url)
    except requests.exceptions.ConnectTimeout:
        print("Could not connect to %s within %d s." % (url, CONNECT_TIMEOUT_S))
    except requests.exceptions.ReadTimeout:
        print("No answer within %d s. The run may still be going on the server and holding its" % READ_TIMEOUT_S)
        print("single optimization slot. Wait a few minutes before trying again.")
    except requests.exceptions.ConnectionError:
        print("Could not connect to %s." % url)
    except requests.exceptions.RequestException as exc:
        print("The request failed: %s" % exc.__class__.__name__)
    return None


def call(session, stage, url, **kwargs):
    """POST and return the parsed JSON body, or None after printing what went wrong."""
    started = time.monotonic()
    response = post(session, url, **kwargs)
    if response is None:
        return None
    if not response.ok:
        explain_error(response, stage)
        return None
    try:
        body = response.json()
    except ValueError:
        print("%s returned a body that is not JSON." % stage)
        return None
    print("(%s answered in %.1f s)" % (stage, time.monotonic() - started))
    return body


def run_live(args):
    token = os.environ.get(TOKEN_ENV, "").strip()
    if not token:
        print("No access token. Set the %s environment variable first." % TOKEN_ENV)
        print(textwrap.fill(ACCESS_NOTE, 88))
        print("See README.md, section 'Live access'.")
        return 2
    missing = missing_files((STAGE1_INPUT, STAGE2_INPUT))
    if missing:
        report_not_captured(missing)
        return 1
    try:
        import requests
    except ImportError:
        print("The live mode needs the 'requests' package: python -m pip install requests")
        return 2
    for name in (STAGE1_INPUT, STAGE2_INPUT):
        if os.path.getsize(example_path(name)) > MAX_UPLOAD_BYTES:
            print("%s is larger than the 2 MB the service accepts." % name)
            return 2

    session = requests.Session()
    session.headers["Authorization"] = "Bearer " + token
    session.headers["Accept"] = "application/json"
    print("%s: linked example (live, %s)" % (SERVICE_TITLE, BASE_URL))
    print()

    with open(example_path(STAGE1_INPUT), "rb") as handle:
        stage1_body = handle.read()
    plan = call(session, "Stage 1", BASE_URL + PLAN_PATH, data=stage1_body,
                headers={"Content-Type": "application/json"})
    if plan is None:
        return 3
    summarize_stage1_output(plan)
    print()

    print("Stage 2 sends the example's heuristic input, not one built from the plan above:")
    print("the service converts a plan into heuristic input internally and does not expose that step.")
    with open(example_path(STAGE2_INPUT), "rb") as handle:
        stage2_bytes = handle.read()
    try:
        expected_tasks = len(as_list(get_ci(get_ci(json.loads(stage2_bytes.decode("utf-8")), "tasks"), "task")))
    except ValueError:
        expected_tasks = None
    params = {"dh": args.dh, "mna": args.mna, "sr": args.sr}
    schedule = call(session, "Stage 2", BASE_URL + SCHEDULE_PATH, params=params,
                    files={"file": (STAGE2_INPUT, stage2_bytes, "application/json")})
    if schedule is None:
        return 3
    summarize_stage2_output(schedule, expected_tasks)
    return 0


# ---------------------------------------------------------------------------
# Command line
# ---------------------------------------------------------------------------

def bounded(name):
    low, high, _ = PARAM_BOUNDS[name]

    def parse(text):
        try:
            value = int(text)
        except ValueError:
            raise argparse.ArgumentTypeError("%s must be an integer" % name)
        if not low <= value <= high:
            raise argparse.ArgumentTypeError("%s must be between %d and %d" % (name, low, high))
        return value
    return parse


def main(argv=None):
    parser = argparse.ArgumentParser(
        description="Linked two-stage example for %s. Offline by default." % SERVICE_TITLE,
        epilog=ACCESS_NOTE)
    parser.add_argument("--live", action="store_true",
                        help="send the example inputs to %s; needs approved live access and a token "
                             "in the %s environment variable" % (BASE_URL, TOKEN_ENV))
    for name, label in (("dh", "decision horizon"), ("mna", "maximum number of alternatives"),
                        ("sr", "sampling rate")):
        low, high, default = PARAM_BOUNDS[name]
        parser.add_argument("--" + name, type=bounded(name), default=default,
                            help="%s for stage 2, %d-%d (live mode only, default %d)"
                                 % (label, low, high, default))
    args = parser.parse_args(argv)
    return run_live(args) if args.live else run_offline()


if __name__ == "__main__":
    sys.exit(main())