task_mcmas.py

import itertools as it



### Template for Environment

ENV_CODE = '''
---Environment agent
Agent Environment
    Obsvars:
        reached: boolean;
    k: 2..{N};
    end Obsvars
    Vars:
        state: {{idle, detected, initiated, completed}};
    end Vars
    Actions = {{none, task_detected_, task_initiated_, task_completed_}};
    Protocol:
        state = idle: {{task_detected_}};
        state = detected: {{task_initiated_}};
        state = completed: {{task_completed_}};
    end Protocol
    Evolution:
        reached=true if {reached_code};
        state = detected if ((state = idle) and {detected_code}) and (Action=task_detected_);
        state = initiated if (state = detected) and (reached=true) and (Action=task_initiated_);
        state = completed if (Action=task_completed_) and
      {completed_code};
    end Evolution
end Agent'''


### Template for Agent

AGENT_CODE = '''
---Robot{i}
Agent Robot{i}
    Vars:
        state: {{idle, busy, ready, promised}};
    end Vars
    Actions = {{task_detected, send_confirm, send_request, send_willing, received_request, received_willing, received_engaged, received_confirm, received_not_required, task_completed}};
    Protocol:
        state = idle: {{task_detected, received_request}};
        state = ready: {{send_request, send_confirm, received_willing, received_engaged}};
        state = busy: {{task_completed}};
        state = promised: {{received_confirm, received_not_required, send_willing}};
    end Protocol
    Evolution:
        state = idle if ((state = idle) and (Action = task_detected) and ({detected_code})) or
                ((state = busy) and (Action = task_completed) and (
            {completed_code})) or
                ((state= promised) and (Action=received_not_required));

        state = ready if ((state = idle) and (Action = task_detected) and (
            {request_code})) or
                ((state=ready) and ((Action = received_willing) or (Action = received_engaged)));

        state = busy if ((state = ready) and (Action = send_confirm)) or
                ((state = promised) and (Action=received_confirm));

        state = promised if {promised_code};
   end Evolution
end Agent'''


### Template for Evaluation. InitStates, Fairness, Formulae

EVAL_CODE = '''
Evaluation
      ready_var if ( {ready_code} );

      bad_condition if {bad_cond_code};
end Evaluation


InitStates
    {init_code} and
    (Environment.reached=false) and (Environment.state=idle){init_team_size};
end InitStates


Fairness
  ready_var;
end Fairness


Formulae
    AG(!bad_condition);
end Formulae'''



#################################
# generate code for Environment #
#################################

def gen_env_code(N):
    return ENV_CODE.format(**{
        'N': N,
        'reached_code': gen_env_reached(N),
        'detected_code': gen_env_detected(N),
        'completed_code': gen_env_completed(N)
    })


def gen_env_reached(N):
    seq = range(1, N+1)
    s = []
    for k in range(2, N+1):
        c = it.combinations(seq, k-1)
        l = ' or '.join('(%s)'%' and '.join(f'(Robot{_i}.Action=received_confirm)' for _i in i) for i in c)
        s.append('((k = %d) and (%s))' % (k, l))
    return '(%s)'%' or '.join(s)


def gen_env_detected(N):
    return '(%s)'%' or '.join('(Robot%d.Action = task_detected)' % i for i in range(1, N+1))


def gen_env_completed(N):
    seq = range(1, N+1)
    s = []
    for k in range(2, N+1):
        c = it.combinations(seq, k)
        l = ' or '.join('(%s)'%' and '.join(f'(Robot{_i}.Action=task_completed)' for _i in i) for i in c)
        s.append('((k = %d) and (%s))' % (k, l))
    return '(%s)'%' or '.join(s)


################################
# generate code for i-th Agent #
################################

def gen_agent_code(N, i):
    return AGENT_CODE.format(**{
        'N': N, 'i': i,
        'detected_code': gen_agent_detected(N, i),
        'completed_code': gen_agent_completed(N, i),
        'request_code': gen_agent_request(N, i),
        'promised_code': gen_agent_promised(N, i),
    })


def gen_agent_detected(N, i):
    return ' or '.join(f'(Robot{_i}.Action = task_detected)' for _i in range(1, N+1) if _i != i)


def gen_agent_completed(N, i):
    return ' or '.join('((Environment.k = %d) and (%s))'%(k,
        ' or '.join('(%s)'%' and '.join(f'(Robot{_i}.Action = task_completed)' for _i in y) for y in it.combinations(filter(lambda x:x!=i, range(1,N+1)), k-1))
    ) for k in range(2, N+1))


def gen_agent_request(N, i):
    return ' or '.join('((Environment.k = %d) and (%s))'%(k,
        ' or '.join('(%s)'%' and '.join(f'(Robot{_i}.Action = received_request)' for _i in y) for y in it.combinations(filter(lambda x:x!=i, range(1,N+1)), k-1))
    ) for k in range(2, N+1))


def gen_agent_promised(N, i):
    return ' or '.join(
        f'((state = idle) and (Action=received_request) and (Robot{_i}.Action=send_request)) or '
        f'((state = promised) and (Action=send_willing) and (Robot{_i}.Action=received_willing))'
    for _i in range(1,N+1) if _i != i)



#############################################################
# generate Evaluation, InitStates, Fairness, Formulae, etc. #
#############################################################

def gen_eval_code(N, k):
    return EVAL_CODE.format(**{
        'N': N,
        'ready_code': gen_eval_ready(N),
        'bad_cond_code': gen_eval_bad_condition(N),
        'init_code': gen_eval_init_states(N),
        'init_team_size': gen_eval_init_team_size(N, k)
    })


def gen_eval_ready(N):
    return ' or '.join(f'(Robot{i}.state=ready)' for i in range(1, N+1))


def gen_eval_bad_condition(N):
    return ' or '.join('(%s)'%' and '.join(f'(Robot{_i}.state={"promised" if _i == i else "idle"})' for _i in range(1,N+1)) for i in range(1,N+1))


def gen_eval_init_states(N):
    return ' and '.join(f'(Robot{i}.state = idle)' for i in range(1, N+1))


def gen_eval_init_team_size(N, k):
    return f' and (Environment.k = {k})' if k else ''



# main driver

if __name__ == '__main__':
    N = int(input('Enter the number of agents: '))
    k = input('Enter the team size (leave empty for a random number every time): ')
    filename = input('Enter output file name (leave empty for `stdout`):')

    import sys

    with open(filename, 'w') if filename else sys.stdout as file:

        print(gen_env_code(N), file=file)
        print()

        for i in range(1,N+1):
            print(gen_agent_code(N, i), file=file)
            print()

        print(gen_eval_code(N, k), file=file)