// Path: Chapter03/src/main/java/com/packtpub/storm/trident/model/DiagnosisEvent.java
// public class DiagnosisEvent implements Serializable {
//     private static final long serialVersionUID = 1L;
//     public double lat;
//     public double lng;
//     public long time;
//     public String diagnosisCode;
// 
//     public DiagnosisEvent(double lat, double lng, long time, String diagnosisCode) {
//         super();
//         this.time = time;
//         this.lat = lat;
//         this.lng = lng;
//         this.diagnosisCode = diagnosisCode;
//     }
// }


import com.packtpub.storm.trident.model.DiagnosisEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
package com.packtpub.storm.trident.operator;



public class CityAssignment extends BaseFunction {
    private static final long serialVersionUID = 1L;
    private static final Logger LOG = LoggerFactory.getLogger(CityAssignment.class);

    private static Map<String, double[]> CITIES = new HashMap<String, double[]>();

    { // Initialize the cities we care about.
        double[] phl = {39.875365, -75.249524};
        CITIES.put("PHL", phl);
        double[] nyc = {40.71448, -74.00598};
        CITIES.put("NYC", nyc);
        double[] sf = {-31.4250142, -62.0841809};
        CITIES.put("SF", sf);
        double[] la = {-34.05374, -118.24307};
        CITIES.put("LA", la);
    }

    @Override
    public void execute(TridentTuple tuple, TridentCollector collector) {
// Path: Chapter03/src/main/java/com/packtpub/storm/trident/model/DiagnosisEvent.java
// public class DiagnosisEvent implements Serializable {
//     private static final long serialVersionUID = 1L;
//     public double lat;
//     public double lng;
//     public long time;
//     public String diagnosisCode;
// 
//     public DiagnosisEvent(double lat, double lng, long time, String diagnosisCode) {
//         super();
//         this.time = time;
//         this.lat = lat;
//         this.lng = lng;
//         this.diagnosisCode = diagnosisCode;
//     }
// }


// Path: Chapter03/src/main/java/com/packtpub/storm/trident/operator/CityAssignment.java
import com.packtpub.storm.trident.model.DiagnosisEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

package com.packtpub.storm.trident.operator;



public class CityAssignment extends BaseFunction {
    private static final long serialVersionUID = 1L;
    private static final Logger LOG = LoggerFactory.getLogger(CityAssignment.class);

    private static Map<String, double[]> CITIES = new HashMap<String, double[]>();

    { // Initialize the cities we care about.
        double[] phl = {39.875365, -75.249524};
        CITIES.put("PHL", phl);
        double[] nyc = {40.71448, -74.00598};
        CITIES.put("NYC", nyc);
        double[] sf = {-31.4250142, -62.0841809};
        CITIES.put("SF", sf);
        double[] la = {-34.05374, -118.24307};
        CITIES.put("LA", la);
    }

    @Override
    public void execute(TridentTuple tuple, TridentCollector collector) {
        DiagnosisEvent diagnosis = (DiagnosisEvent) tuple.getValue(0);