A Clojure simulation of the Passport Seva Kendra

psk.clj

(ns challenges.psk
  "A Simulation of the Passport Seva Kendra"
  (:require [clj-time.core :as ct]
            [clojure.tools.logging :as ctl])
  (:import [java.util.concurrent LinkedBlockingQueue PriorityBlockingQueue]))

(def working-hours? (atom false)) ; use this to control agents.
                                        ; Turning this off will
                                        ; shutdown agents.

(def stages
  "The various stages in the PSK, and transition from one stage to the other.

  - `processing-time-range` represents the amount of time spent at the
  counter in this stage (in mins).
  - `next` represents the next stage for the person.
  - `counters` represent the number of counters/agents serving this stage."
  ;; Actual Values:
  ;; 4 Doc verification Counters
  ;; 33 counters for Biometrics
  ;; 12 for form checking
  ;; 10 for final checking
  ;; 3 for corrections
  ;; Using different values here so that the display board is
  ;; human-readable.
  {::enter {:next ::doc-verification}
   ::doc-verification {:next ::biometrics
                       :counters 10
                       :display-str "0-"
                       :processing-time-range [1 2]}
   ::biometrics {:next ::form-check
                 :counters 3
                 :display-str "A-"
                 :processing-time-range [3 15]}
   ::form-check {:next ::final-check
                 :counters 2
                 :display-str "B-"
                 :processing-time-range [2 4]
                 :failure ::corrections}
   ::final-check {:next ::exit
                  :counters 1
                  :display-str "C-"
                  :processing-time-range [2 4]}
   ::corrections {:next ::form-check
                  :counters 1
                  :display-str "D-"
                  :processing-time-range [5 15]}})

(def waiting-room-capacity 500) ; max number of waiting people

(def total-capacity
  "number of people that can be in the PSK at max"
  (apply + waiting-room-capacity (keep :counters (vals stages))))

(def processing-batch-size
  "no of people entering the center at one time."
  ;; Actual capacity: 25
  25)

(def new-batch-in-mins
  "time between one batch and the next"
  15)

(def guide-people-to-next-stage-ms 500) ; every x millis, guide people
                                        ; to the next stage.

(def stage-status
  "For the given stage, the map of possible states that a person can be
  in."
  [;; Person is in the waiting area, looking at the display to see
   ;; when he's called to a counter.
   ::waiting
   ;; Counter is assigned to the person
   ::in-process
   ;; Person is done with the counter and heading to the next stage.
   ::done])

(def token-generator
  "Give the next token number to the applicant."
  (atom {:normal 0
         :senior 0
         :tatkal 0
         :police-clearance 0}))

(defrecord Token [type val])

(def ^:private token-type-distribution
  "Represents which % of population falls under which passport type.

  This distribution is just my guess! :P"
  {:normal 90
   :senior 5
   :tatkal 3
   :police-clearance 2})

(defn- get-token-type-by-distribution
  "Returns a random token type, skewed by weights defined above (in
  `token-type-distribution`)."
  []
  ;; The idea is to use numbers to represent distribution. We take
  ;; a distribution of:

  ;; {:normal 90
  ;;  :senior 5
  ;;  :tatkal 3
  ;;  :police-clearance 2}

  ;; and say that a number
  ;; 1. from 1  to 90 represents normal,
  ;; 2. from 91 to 95 represents senior,
  ;; 3. from 96 to 98 represents tatkal and
  ;; 4. 99-100 represents police-clearance.

  ;; Then, we use the built in `rand-int` function to generate any
  ;; number between 1 and 100. We use the generated number to identify
  ;; the category.
  (let [[total weighted-seq]
        (reduce (fn [[n ws] m]
                  [(+ n (:dist m))
                   (conj ws
                         (assoc m :weight (+ n (:dist m))))])
                [0 []]
                (sort-by :dist
                         (map (fn [[k v]] {:type k :dist v})
                              token-type-distribution)))
        w (rand-int total)]
    (loop [elem (first weighted-seq)
           ws (rest weighted-seq)]
      (if (<= w (:weight elem))
        (:type elem)
        (recur (first ws) (rest ws))))))

;;; `enter-time` and `exit-time` are vanity metrics to
;;; calculate the amount of time the person spent in the PSK.
(defrecord Person
    [token stage stage-status enter-time exit-time total-time psk-agent stage-history])

(let [token-type->char {:normal "N-"
                        :senior "S-"
                        :tatkal "T-"
                        :police-clearance "P-"}]
  (defn- person-representation
    "The visual representation of the applicant. eg: N-300, T-50 etc."
    [person]
    (let [token (:token person)]
      (str (token-type->char (:type token)) (:val token)))))

(defrecord StageHistory
    [name status change-time])

(defn create-person!
  "Create a new person for the PSK. Token numbers are always
  monotonically increasing. This function modifies the global atom
  `token-generator` as a side-effect. This atom is used to generate
  the next token number."
  []
  (let [person-type (get-token-type-by-distribution)
        person-number (get (swap! token-generator
                                  update
                                  person-type
                                  inc)
                           person-type)
        time-instant (ct/now)]
    (Person. (Token. person-type person-number)
             ::enter
             ::done
             time-instant
             nil
             nil
             nil
             [(StageHistory. ::enter ::done time-instant)])))

(defn- get-processing-time-for-stage
  "Given a `stage-config`, get the processing time range for the stage.
  Calculate a random amount of time within this range. Note that we
  treat mins as secs in this simulation."
  [stage-config]
  (let [[min-time max-time] (:processing-time-range stage-config)]
    (* 3000 (+ min-time (rand-int (- max-time min-time))))))

(defn- agent-representation
  "The visual representation of the psk-agent. eg: A-300, B-50 etc."
  [psk-agent]
  (str (get-in psk-agent [:config :display-str]) (:id psk-agent)))

(defrecord PSKAgent
    [id type config])

(defn- create-agents
  "For the given `agent-type`, create the given `num` of agents."
  [agent-type stage-config num]
  (map (comp agent (fn [i] (PSKAgent. (inc i) agent-type stage-config)))
       (range num)))

(defn create-kendra-agents
  "Given the `stages` and their config for the kendra, create the
  appropriate agents to work these counters."
  [kendra-stages]
  (let [stages-with-counters (-> kendra-stages
                                 keys
                                 set
                                 ;; Remove the stages where no counter
                                 ;; of agents is needed.
                                 (disj ::enter ::exit))]
    (mapcat (fn [s]
              (let [config (get kendra-stages s)]
                (create-agents s config (:counters config))))
            stages-with-counters)))

(defn- store-stage-change
  "For the given `Person` ref, store the change to their stage for later analysis."
  ([person new-stage new-status]
   (store-stage-change person new-stage new-status (ct/now)))
  ([person new-stage new-status time-instant]
   (let [stage-log (->StageHistory new-stage new-status time-instant)]
     (dosync
      (alter person
             update
             :stage-history
             conj
             stage-log)))))

(defn- call-person-to-counter
  "Announce that person should come to the processing counter. Takes
  `person` and `notice-board` refs, performs a transactional update."
  [stage notice-board psk-agent person]
  (dosync
   (alter person
          assoc
          :stage stage
          :stage-status ::in-process
          :psk-agent (agent-representation psk-agent))
   (store-stage-change person stage ::in-process)
   (alter notice-board
          assoc
          (person-representation @person)
          (agent-representation psk-agent))))

(def unlucky-applicant?
  "Introduce a little anarchy!"
  (atom #{"N-3" "S-2"}))

(defn- process-person
  "Do the work for processing the given person. Takes a `person`
  object and not a ref."
  [stage stage-config psk-agent person]
  (let [processing-time* (get-processing-time-for-stage stage-config)
        processing-time (if (@unlucky-applicant? (person-representation person))
                          ;; You will need more time because the gods
                          ;; are against you.
                          (* 10 processing-time*)
                          processing-time*)]
    (ctl/debug (format "[Agent: %s] [Stage: %s] Doing %s ms of work for %s person"
                       (agent-representation psk-agent)
                       stage
                       processing-time
                       (person-representation person)))
    (Thread/sleep processing-time)))

(defn- mark-processing-as-complete
  "For this `person`, this `stage` is complete. Update accordingly.
  Takes `person` and `notice-board` refs, performs a transactional
  update."
  [stage notice-board psk-agent person]
  (let [person-str (person-representation @person)]
    (dosync
     (alter person assoc :stage stage :stage-status ::done :psk-agent nil)
     (store-stage-change person stage ::done)
     (alter notice-board dissoc person-str))

    (ctl/debug "Processing Complete!"
               (assoc psk-agent :current-applicant person-str))))

(defn process-applicant
  "Get a person from the queue. Process this person as per the rules
  of the counter."
  [psk-agent my-queue notice-board]
  (ctl/debug (format "[Agent: %s] Waiting for next person!"
                     (agent-representation psk-agent)))
  (if @working-hours?

    (if-let [person (.poll my-queue 1 java.util.concurrent.TimeUnit/SECONDS)]
      (let [stage (:type psk-agent)
            stage-config (:config psk-agent)]
        (ctl/debug (format "[Agent: %s] We have a person: %s"
                           (agent-representation psk-agent)
                           person))
        (call-person-to-counter stage notice-board psk-agent person)
        (process-person stage stage-config psk-agent @person)
        (mark-processing-as-complete stage notice-board psk-agent person)
        ;; Next!
        (ctl/debug "Repeating Agent")
        (send-off *agent* process-applicant my-queue notice-board)
        ;; Set the new state of the agent.
        (assoc psk-agent :last-processed (person-representation @person)))
      (do (send-off *agent* process-applicant my-queue notice-board)
          psk-agent))

    (ctl/info (format "[Agent: %s] Working hours are over! Closing Shop! Come back later!"
                      (agent-representation psk-agent)))))

(defn- book-keeping-for-applicants
  "Remove all applicants who are completely done from
  `active-applicants`. Store them in `done-applicants` for
  book-keeping.

  *NOTE* : Since this goes through the entire collection, it is slow.
  Hence we run it when sending in new batches of people."
  [active-applicants done-applicants]
  (ctl/debug "Removing completed applications for book-keeping!")
  (dosync
   (let [[active-people done-people] (reduce (fn [[aa da] p]
                                               (if (and (= (:stage @p) ::exit)
                                                        (= (:stage-status @p) ::done))
                                                 [aa (conj da p)]
                                                 [(conj aa p) da]))
                                             [[] []]
                                             @active-applicants)]
     (ref-set active-applicants active-people)
     (alter done-applicants into done-people))))

(defn let-people-through
  "Send people into the PSK in batches as defined by
  `processing-batch-size` and `new-batch-in-mins`. Note that in our
  code we use seconds to represent minutes."
  [active-applicants done-applicants]
  (future
    (loop []
      (if @working-hours?

        (if (> (count @active-applicants)
               (- total-capacity processing-batch-size))

          (do (ctl/info "[Entry] PSK at max capacity! We need to cancel this batch!")
              (Thread/sleep (* 1000 new-batch-in-mins))
              (recur))

          (let [new-people (->> create-person!
                                (repeatedly processing-batch-size)
                                (map ref))]
            (dosync
             (alter active-applicants into new-people))
            (ctl/debug (format "[Entry] %s new people entered into the PSK."
                               (count new-people)))

            (book-keeping-for-applicants active-applicants done-applicants)

            (Thread/sleep (* 1000 new-batch-in-mins))
            (recur)))

        (ctl/info "[Entry] Working hours are over! Closing Shop! Come back later!")))))

(defn- mark-applicant-process-as-complete
  "Takes a `person` ref object and marks its processing as complete."
  [person]
  (let [time-instant (ct/now)]
    (dosync
     (alter person
            assoc
            :stage ::exit
            :exit-time time-instant
            :total-time (-> @person
                            :enter-time
                            (ct/interval time-instant)
                            ct/in-seconds))
     (store-stage-change person ::exit ::done time-instant)))

  (ctl/debug (format "Dear %s, Thank you for visiting the Passport Seva Kendra! Your entire experience took: %s mins!"
                     (person-representation @person)
                     (:total-time @person))))

(defn- move-applicant-to-next-stage
  "Given a `person` ref and the next stage they should go to, move
  them to the stage. Does a transactional update."
  [stage->queue next-stage person]
  (ctl/debug (format "[Guide] %s from %s to %s"
                     (person-representation @person)
                     (:stage @person)
                     next-stage))
  (dosync
   (alter person
          assoc
          :stage next-stage
          :stage-status ::waiting)
   (store-stage-change person next-stage ::waiting))
  (.put (stage->queue next-stage) person))

(defn move-people-through
  "Review all the active applicants and move them into appropriate stages."
  [kendra-stages stage->queue active-applicants]
  (future
    (loop []
      (if @working-hours?

        (let [people @active-applicants]
          (doseq [person (->> people
                              (group-by (comp :stage-status deref))
                              ::done)]
            (if (= ::exit (get-in kendra-stages [(:stage @person) :next]))
              (mark-applicant-process-as-complete person)
              (when-let [next-stage (get-in kendra-stages
                                            [(:stage @person) :next])]
                (move-applicant-to-next-stage stage->queue next-stage person))))

          (Thread/sleep guide-people-to-next-stage-ms)
          (recur))

        (ctl/info "[Guide] Working hours are over! Closing Shop! Come back later!")))))

(let [comp-score {:normal 1
                  :tatkal 2
                  :senior 3
                  :police-clearance 4}]
  (defn- person-comparator
    "A `comparator` function for comparing 2 `person` ref objects. This
  allows us to use a `PriorityBlockingQueue` instead of a simple
  `LinkedBlockingQueue`."
    [p1 p2]
    (let [t1 (:token @p1)
          t2 (:token @p2)]
      (if (= (:type t1) (:type t2))
        (< (:val t1) (:val t2))
        (> (comp-score (:type t1))
           (comp-score (:type t2)))))))

(defn- create-queue
  "Create a queue for the kendra"
  ([q-capacity]
   (create-queue :lbq q-capacity))
  ([q-type q-capacity]
   (case q-type
     :lbq (LinkedBlockingQueue. q-capacity)
     ;; ^ The PSK implementation is based on a
     ;; `LinkedBlockingQueue`. Using a `PriorityBlockingQueue`
     ;; gives a much better experience to the people who are in
     ;; the Kendra.
     :pbq (PriorityBlockingQueue. q-capacity person-comparator))))

(defn create-kendra-queues
  "Given the counter-types / `stages` in the kendra, create the
  appropriate queues."
  [kendra-stages q-capacity]
  (let [queues-we-need (-> kendra-stages
                           keys
                           set
                           ;; Remove the stages where no queue of
                           ;; people is needed.
                           (disj ::enter ::exit))]
    (reduce (fn [m s]
              (assoc m
                     s (create-queue :pbq q-capacity)))
            {}
            queues-we-need)))

(defn- show-applicant-info
  "Helper function to print overall applicant information."
  [applicants]
  (let [people @applicants]
    (reduce (fn [m [stage ps]]
              (assoc m
                     stage (merge (reduce (fn [m [st p]] (assoc m st (count p)))
                                          {}
                                          (group-by (comp :stage-status deref) ps))
                                  {::total (count ps)})))
            {:absolute-total (count people)}
            (group-by (comp :stage deref)
                      people))))

(defn- get-processing-time-info
  "Print interesting information about "
  [done-applicants]
  (let [header ["Total Processing Time (mins): "
                "Total People Processed: "
                "Avg Processing Time (mins): "
                "Max Processing Time (mins): "
                "Person who took the most time: "]
        numbers (reduce (fn [[tot n avg max max-name] p]
                          [(+ tot (:total-time @p))
                           (inc n)
                           (int (/ (+ tot (:total-time @p)) (inc n)))
                           (if (> max (:total-time @p))
                             max
                             (:total-time @p))
                           (if (> max (:total-time @p))
                             max-name
                             (person-representation @p))])
                        [0 0 0 0 ""]
                        @done-applicants)]
    (interleave header numbers)))

(defn- show-processing-log
  "Given an identifier for a person, show their progress so far."
  [identifier applicants]
  (when-let [person (first (filter (comp (partial = identifier)
                                         person-representation
                                         deref)
                                   @applicants))]
    (last
     (reduce (fn [[prev-time prev-log stream] {:keys [change-time] :as h}]
               [change-time
                h
                (->> change-time
                     (ct/interval prev-time)
                     ct/in-seconds
                     (assoc (dissoc prev-log :change-time) :processing-time)
                     (conj stream))])
             [(:enter-time @person)
              (first (:stage-history @person))
              []]
             (rest (:stage-history @person))))))

(defn start-the-kendra!
  "Setup our Passport Seva Kendra."
  []
  (let [;; Create queues for the various stages, returns a map of
        ;; stage-name -> queue
        stage->queue (create-kendra-queues stages total-capacity)
        ;; Create all the agents
        list-of-agents (create-kendra-agents stages)
        ;; Create a display board for waiting members
        notice-board (ref (sorted-map))
        ;; Track all the active applicants
        active-applicants (ref [])
        ;; Track all the completed applicants (for debugging /
        ;; historical data purpose)
        done-applicants (ref [])]
    (ctl/info "[PSK] Welcome, today is a good day.")
    ;; For each agent at each counter, start processing!
    (doseq [a list-of-agents]
      ;; Get the stage this agent is working at, and the queue of
      ;; people for that stage.
      (let [s (:type @a)
            q (stage->queue s)]
        ;; Start processing people from the queue concurrently in
        ;; independent threads.
        (send-off a process-applicant q notice-board)))
    ;; Start a continuous future for applicants to periodically enter
    ;; the PSK.
    (let-people-through active-applicants done-applicants)
    ;; Start a helper process to move people from one stage to the
    ;; other.
    (move-people-through stages stage->queue active-applicants)
    ;; Return the data. We'll use this to monitor our system.
    [notice-board active-applicants done-applicants]))


(comment
  (reset! working-hours? true)
  (def state (start-the-kendra!))
  (clojure.pprint/pprint @(first state))
  (def f
    (future (loop []
              (clojure.pprint/pprint "=========DISPLAY BOARD========")
              (clojure.pprint/pprint @(first state))
              (Thread/sleep 2000)
              (recur))))
  (future-cancel f)
  (clojure.pprint/pprint (show-applicant-info (second state)))
  (clojure.pprint/pprint (show-applicant-info (last state)))
  (clojure.pprint/pprint (get-processing-time-info (last state)))
  (clojure.pprint/pprint (show-processing-log "N-18" (last state)))
  (reset! working-hours? false))