Source code

001package com.astrolabsoftware.FinkBrowser.Januser;
002
003import com.Lomikel.Utils.Pair;
004import com.Lomikel.Utils.LomikelException;
005import com.Lomikel.Januser.GremlinRecipies;
006import com.Lomikel.Januser.ModifyingGremlinClient;
007import com.astrolabsoftware.FinkBrowser.HBaser.FinkHBaseClient;
008import com.astrolabsoftware.FinkBrowser.FinkPortalClient.FPC;
009
010// Tinker Pop
011import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
012import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal;
013import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.otherV;
014import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.V;
015import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.fold;
016import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.has;
017import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.not;
018import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.unfold;
019import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.out;
020import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.repeat;
021import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.values;
022import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.count;
023import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.addV;
024import static org.apache.tinkerpop.gremlin.process.traversal.P.within;
025import org.apache.tinkerpop.gremlin.structure.Edge;
026import org.apache.tinkerpop.gremlin.structure.Vertex;
027import org.apache.tinkerpop.gremlin.structure.Direction;
028
029// HBase
030import org.apache.hadoop.hbase.client.Get;
031import org.apache.hadoop.hbase.client.Get;
032
033// org.json
034import org.json.JSONArray;
035import org.json.JSONObject;
036
037// Java
038import java.lang.Math;
039import java.util.Arrays;
040import java.util.Iterator;
041import java.util.ArrayList;
042import java.util.List;
043import java.util.Map;
044import java.util.HashMap;
045import java.util.SortedSet;
046import java.util.TreeSet;
047import java.util.Set;
048import java.util.HashSet;
049import java.util.Iterator;
050import java.util.Calendar;
051import java.util.Date;
052import java.text.SimpleDateFormat;
053
054// Log4J
055import org.apache.logging.log4j.Logger;
056import org.apache.logging.log4j.LogManager;
057
058/** <code>FinkGremlinRecipies</code> provides various recipies to handle
059  * and modify Gremlin Graphs for Fink.
060  * @opt attributes
061  * @opt operations
062  * @opt types
063  * @opt visibility
064  * @author <a href="mailto:Julius.Hrivnac@cern.ch">J.Hrivnac</a> */
065// TBD: check precodition for methods, wgich doesn't work with 'client' creation
066public class FinkGremlinRecipies extends GremlinRecipies {
067    
068  /** Create and attach to {@link GraphTraversalSource}.
069    * @param g The attached {@link GraphTraversalSource}. */
070  public FinkGremlinRecipies(GraphTraversalSource g) {
071    super(g);
072    }
073    
074  /** Create and attach to {@link ModifyingGremlinClient}.
075    * @param client The attached  {@link ModifyingGremlinClient}. */
076  public FinkGremlinRecipies(ModifyingGremlinClient client) {
077    super(client);
078    }
079    
080  /** Execute full chain of new <em>object</em> correlations analyses.
081    * @param classifiers The {@link Classifier}s to be used.
082    *                        They can contain the {@link Classifier} flavor after <em>=</em> symbol.
083    * @param filter          The HBase evaluation formula to be applied.
084    *                        Ignored if <tt>clss</tt> are specified.
085    * @param hbaseUrl        The url of HBase with alerts as <tt>ip:port:table:schema</tt>.
086    * @param nLimit          The maximal number of alerts getting from HBase or Fink Portal.
087    *                        <tt>0</tt> means no limit.
088    * @param timeLimit       How far into the past the search should search (in minutes).
089    * @param clss            An array of <em>classes</em> taken from {@link FPC},
090    *                        if contains <tt>Anomaly</tt>, get anomalies from {@link FPC},                  
091    *                        if <tt>null</tt>, analyse <em>object</em>s from HBase database.
092    * @throws LomikelException If anything fails. */
093  /*@Deprecated
094  public void processOCol(Classifier[] classifiers,
095                          String       filter,
096                          String       hbaseUrl,
097                          int          nLimit,
098                          int          timeLimit,
099                          String[]     clss) throws LomikelException {
100    fillOCol(classifiers, filter, hbaseUrl, nLimit, timeLimit, clss);
101    generateCorrelations(classifiers);
102    }*/
103        
104  /** Fill graph with <em>OCol</em>.
105    * @param classifiers The {@link Classifier}s to be used.
106    * @param filter      The HBase evaluation formula to be applied.
107    *                    Ignored if <tt>clss</tt> are specified.
108    * @param hbaseUrl    The url of HBase with alerts as <tt>ip:port:table:schema</tt>.
109    * @param nLimit      The maximal number of alerts getting from HBase or Fink Portal.
110    *                    <tt>0</tt> means no limit.
111    * @param timeLimit   How far into the past the search should search (in minutes).
112    * @param clss        An array of <em>classes</em> taken from {@link FPC},
113    *                    if contains <tt>Anomaly</tt>, get anomalies from {@link FPC},                  
114    *                    if <tt>null</tt>, analyse <em>object</em>s from HBase database.
115
116    * @throws LomikelException If anything fails. */
117  /*@Deprecated
118  public void fillOCol(Classifier[] classifiers,
119                       String       filter,
120                       String       hbaseUrl,
121                       int          nLimit,
122                       int          timeLimit,
123                       String[]     clss) throws LomikelException {
124    String clssDesc = "";
125    if (clss != null) {
126      clssDesc = "of " + Arrays.toString(clss);
127      }
128    log.info("Filling OCol " + clssDesc + " using " + Arrays.toString(classifiers) + " classifiers, nLimit = " + nLimit + ", timeLimit = " + timeLimit);
129    log.info("Importing from " + hbaseUrl + ":");
130    fhclient(hbaseUrl);
131    Set<String> oids = new HashSet<>();;
132    if (clss == null) { 
133      fhclient().setEvaluation(filter);
134      if (nLimit > 0) {
135        fhclient().setLimit(nLimit);
136        }
137      oids = fhclient().latestsT("i:objectId",
138                                 null,
139                                 timeLimit,
140                                 true);
141      fhclient().setEvaluation(null);
142      }
143    else {
144      Calendar cal;
145      Date d;
146      String sd;
147      JSONArray ja;
148      JSONObject jo;
149      for (String cls : clss) {
150        cal = Calendar.getInstance();
151        cal.add(Calendar.MINUTE, -nLimit);
152        d = cal.getTime();
153        sd = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(d);
154        if (cls.equals("*")) {
155          ja = FPC.anomaly(new JSONObject().put("n",             nLimit).
156                                            put("startdate",     sd).
157                                            put("columns",       "i:objectId").
158                                            put("output-format", "json"));
159          }
160        else {
161          ja = FPC.latests(new JSONObject().put("n",             nLimit).
162                                            put("class",         cls).
163                                            put("startdate",     sd).
164                                            put("columns",       "i:objectId").
165                                            put("output-format", "json"));
166          }
167        for (int i = 0; i < ja.length(); i++) {
168          jo = ja.getJSONObject(i);
169          oids.add(jo.getString("i:objectId"));
170          }
171        log.info("*** " + cls + "[" + ja.length() + "]:");
172        }
173      }
174    classifySources(classifiers, oids, hbaseUrl);
175    }*/
176    
177  /** Classify <em>object</em> .
178    * @param classifiers The {@link Classifier}s to be used.
179    * @param oids        The {@link Set} of <tt>objectId</tt>s of <em>object</em> to be added.
180    * @param hbaseUrl    The url of HBase with alerts as <tt>ip:port:table:schema</tt>.
181    * @throws LomikelException If anything fails. */
182  public void classifySources(Classifier[] classifiers,
183                              Set<String>  oids,
184                              String       hbaseUrl) throws LomikelException {
185    int size = oids.size();
186    int n = 0;
187    long dt;
188    double freq;
189    long startTime = System.currentTimeMillis();
190    // loop over all objects
191    for (String oid : oids) {
192      log.info(oid + " (" + n + " of " + size + "):");
193      for (Classifier classifier : classifiers) {
194        try {
195          classifySource(classifier, oid, hbaseUrl);
196          }
197        catch (LomikelException e) {
198          log.error("Cannot get classification for " + oid, e);
199          }
200        }
201      n++;
202      dt = (System.currentTimeMillis() - startTime) / 1000;
203      freq = (double)n / (double)dt;
204      log.info("\t\twith " + String.format("%.2f", freq) + " Hz");
205      }
206    }
207    
208   /** Classify <em>object</em>.
209    * @param classifier The {@link Classifier} to be used.
210    * @param objectId   The <tt>objectId</tt> of <em>object</em> to be added.
211    * @param hbaseUrl   The url of HBase with alerts as <tt>ip:port:table:schema</tt>.
212    * @throws LomikelException If anything fails. */
213  public void classifySource(Classifier classifier,
214                             String     objectId,
215                             String     hbaseUrl) throws LomikelException {  
216    fhclient(hbaseUrl);
217    classifySource(classifier, objectId);
218    }
219   
220  /** Classify <em>object</em>.
221    * @param classifier The {@link Classifier} to be used.
222    * @param objectId   The <tt>objectId</tt> of <em>object</em> to be added.
223    * @throws LomikelException If anything fails. */
224  public void classifySource(Classifier classifier,
225                             String     objectId) throws LomikelException {  
226    if (g().V().has("lbl", "object").has("objectId", objectId).hasNext()) {
227      Vertex v1 = g().V().has("lbl", "object").has("objectId", objectId).next();
228      List<Vertex> v2s = g().V(v1).in().
229                                   has("lbl",        "OCol").
230                                   has("survey",     classifier.survey()).
231                                   has("classifier", classifier.name()  ).
232                                   has("flavor",     classifier.flavor()).
233                                   toList();
234      Iterator<Edge> edges;
235      for (Vertex v2 : v2s) {
236        edges = g().V(v1).inE().
237                          has("lbl", "deepcontains").
238                          where(otherV().
239                          is(v2)).
240                          toStream().
241                          iterator();
242        while (edges.hasNext()) {
243          edges.next().remove(); 
244          }
245        }        
246      // will be commited in registration
247      }
248    classifier.classify(this, objectId);
249    }
250       
251  /** Register  <em>object</em> in <em>OCol</em>.
252    * @param classifier The {@link Classifier} to be used.
253    * @param cls        The type (class) of <em>OCol</em> {@link Vertex}.
254    *                   It will be created if not yet exists.
255    * @param objectId   The objectId of the new <em>Source</em> {@link Vertex}.
256    *                   It will be created if not yet exists.
257    * @param weight     The weight of the connection.
258    *                   Usualy the number of <em>Alerts</em> of this type. 
259    * @param instanceS  The <em>jd</em> of related <em>Alerts</em> as strings separated by comma.
260    *                   Potential square brackets are removed.
261    *                   May be <tt>null</tt> or empty. */
262  public void registerOCol(Classifier classifier,
263                           String     cls,
264                           String     objectId,
265                           double     weight,
266                           String     instancesS,
267                           String     weightsS) {   
268    List<String> instances = new ArrayList<>();
269    List<Double> weights   = new ArrayList<>();
270    if (instancesS != null && !instancesS.trim().equals("")) {
271      for (String instance : instancesS.replaceAll("\\[", "").replaceAll("]", "").split(",")) {
272        instances.add(instance);
273        }
274      }
275    if (weightsS != null && !weightsS.trim().equals("")) {
276      for (String weighs : weightsS.replaceAll("\\[", "").replaceAll("]", "").split(",")) {
277        weights.add(Double.valueOf(weight));
278        }
279      }
280    registerOCol(classifier, cls, objectId, weight, instances, weights);
281    }
282    
283  /** Register <em>object</em> in <em>OCol</em>. Replace possible existing registration.
284    * @param classifier The {@link Classifier} to be used.
285    * @param cls        The type (class) of <em>OCol</em> {@link Vertex}.
286    *                   It will be created if not yet exists.
287    * @param objectId   The objectId of the new <em>Source</em> {@link Vertex}.
288    *                   It will be created if not yet exists.
289    * @param weight     The total weight of the connection.
290    *                   Usualy the number of <em>Alerts</em> of this type. 
291    * @param instances  The <em>jd</em> of related <em>Alerts</em>.
292    * @param weights    The weights of related <em>Alerts</em>. */
293  public void registerOCol(Classifier   classifier,
294                           String       cls,
295                           String       objectId,
296                           double       weight,
297                           List<String> instances,
298                           List<Double> weights) { 
299    Map<String, String> attributes = new HashMap<>();
300    attributes.put("weight",    "" + weight);
301    attributes.put("instances", instances.toString().replaceFirst("\\[", "").replaceAll("]", ""));
302    attributes.put("weights",   weights.toString().replaceFirst("\\[", "").replaceAll("]", ""));
303    registerOCol(classifier, cls, objectId, attributes, true);
304    }
305    
306  /** Register <em>object</em> in <em>OCol</em>.
307    * @param classifier The {@link Classifier} to be used.
308    * @param cls        The type (class) of <em>OCol</em> {@link Vertex}.
309    *                   It will be created if not yet exists.
310    * @param objectId   The objectId of the new <em>Source</em> {@link Vertex}.
311    *                   It will be created if not yet exists.
312    * @param attributes The additional {@link Edge} attributes.    
313    * @param replace    Whether to replace existing resistration. */
314  public void registerOCol(Classifier          classifier,
315                           String              cls,
316                           String              objectId,
317                           Map<String, String> attributes,
318                           boolean             replace) {   
319    //log.info("\tregistering " + objectId + " as " + classifier + " / " + cls + " with attributes " + attributes + ", replace = " + replace);
320    log.info("\tregistering " + objectId + " as " + classifier + " / " + cls + " with weight = " + attributes.get("weight") + ", replace = " + replace);
321    Vertex ocol = g().V().has("lbl",        "OCol"             ).
322                          has("survey",     classifier.survey()).
323                          has("classifier", classifier.name()  ).
324                          has("flavor",     classifier.flavor()).
325                          has("cls",        cls                ).
326                          fold().
327                          coalesce(unfold(), 
328                                  addV("OCol").
329                                  property("lbl",        "OCol").
330                                  property("survey",     classifier.survey()).
331                                  property("classifier", classifier.name()  ).
332                                  property("flavor",     classifier.flavor()).
333                                  property("cls",        cls                )).
334                         next();
335    Vertex s = g().V().has("lbl",      "object").
336                       has("objectId", objectId).
337                       fold().
338                       coalesce(unfold(), 
339                                addV("object").
340                                property("lbl",      "object").
341                                property("objectId", objectId)).
342                       property("importDate", _now).
343                       next();
344    if (replace) {
345      addEdge(g().V(ocol).next(),
346              g().V(s).next(),
347              "deepcontains",
348              attributes.keySet().toArray(new String[0]),
349              attributes.values().toArray(new String[0]),
350              true);
351      }
352    else {
353      Edge e = ocol.addEdge("deepcontains", s);
354      e.property("lbl", "deepcontains");
355      for (Map.Entry<String, String> attribute : attributes.entrySet()) {
356        e.property(attribute.getKey(), attribute.getValue());
357        }
358      }
359    commit();
360    }
361   
362  /** Clean tree under <em>OCol</em>.
363    * Drop alerts. Alerts are dropped even if they have other
364    * {@link Edge}s.
365    * @param classifier The {@link Classifier} to be used.
366    *                   All its <em>flavors</em> are handled.
367    * @param cls        The type (class) of <em>OCol</em>.
368    * @throws LomikelException If anything goes wrong. */
369  public void cleanOCol(Classifier classifier,
370                        String     cls) throws LomikelException {
371    log.info("Cleaning " + cls + " OCol");
372    g().V().has("lbl",        "OCol").
373            has("survey",     classifier.survey()).
374            has("classifier", classifier.name()  ).
375            has("flavor",     classifier.flavor()).
376            has("cls",        cls                ).
377            out().
378            out().
379            drop().
380            iterate();
381    commit();
382    }
383        
384  /** Generate <em>overlaps</em> Edges between <em>OCol</em>.
385    * Possibly between two {@link Classifier}s.
386    * @param classifier The {@link Classifier}s to be used. */
387  public void generateCorrelations(Classifier... classifiers) {
388    log.info("Generating correlations for OCol of " + Arrays.asList(classifiers));
389    // Clean all correlations 
390    g().E().has("lbl", "overlaps").
391            drop().
392            iterate();
393    List<String> surveysL = new ArrayList<>();
394    List<String> namesL   = new ArrayList<>();
395    List<String> flavorsL = new ArrayList<>();
396    for (Classifier classifier : classifiers) {
397      surveysL.add(classifier.survey());
398      namesL.add(  classifier.name()  );
399      flavorsL.add(classifier.flavor());
400      // Remove wrong OCol
401      g().V().has("lbl",        "OCol"             ).
402              has("classifier", classifier.name()  ).
403              has("flavor",     classifier.flavor()).
404              not(has("cls")).
405              drop().
406              iterate();
407      }
408    String[] surveys = surveysL.toArray(String[]::new);
409    String[] names   = namesL.toArray(  String[]::new);
410    String[] flavors = flavorsL.toArray(String[]::new);
411    commit();
412    // Accumulate correlations and sizes
413    Map<OCol, Double>             weights0 = new HashMap<>(); // cls -> weight (for one objext)
414    Map<Pair<OCol, OCol>, Double> corrS    = new HashMap<>(); // [cls1, cls2] -> weight (for all object between OCol-OCol)
415    Map<OCol, Double>             sizeS    = new HashMap<>(); // cls -> total (for all objects of OCol)
416    SortedSet<OCol>               types0   = new TreeSet<>(); // [cls] (for one object)
417    SortedSet<OCol>               types    = new TreeSet<>(); // [cls] (for all objects)
418    Vertex object;
419    Iterator<Edge> deepcontainsIt;
420    Edge deepcontains;
421    Edge overlaps;
422    double weight;
423    double weight1;
424    double weight2;
425    double cor;
426    Vertex ocol;
427    Vertex ocol1;
428    Vertex ocol2;
429    OCol cls;
430    Pair<OCol, OCol> rel;
431    // Loop over objets and accumulated weights to each object
432    GraphTraversal<Vertex, Vertex> objectT = g().V().has("lbl", "object");
433    while (objectT.hasNext()) {
434      weights0.clear();
435      types0.clear();
436      object = objectT.next();
437      deepcontainsIt = object.edges(Direction.IN);
438      // Get all weights to this object
439      while (deepcontainsIt.hasNext()) {
440        deepcontains = deepcontainsIt.next();
441        weight = Double.parseDouble(deepcontains.property("weight").value().toString());
442        ocol1 = deepcontains.outVertex();
443        cls = new OCol(ocol1);
444        types0.add(cls);
445        types.add(cls);
446        weights0.put(cls, weight);
447        }
448      // Double loop over accumulated weights and fill weights between OCols
449      for (OCol cls1 : types0) {
450        weight1 = weights0.get(cls1);
451        for (OCol cls2 : types0) {
452          weight2 = weights0.get(cls2);
453          rel = Pair.of(cls1, cls2);
454          // OCol-OCol
455          if (!corrS.containsKey(rel)) {
456            corrS.put(rel, 0.0);
457            }
458          cor = corrS.get(rel);
459          //corrS.put(rel, cor + 1.0);
460          corrS.put(rel, cor + Math.sqrt(weight1 * weight2));
461          }
462        }
463      }
464    // Fill total sizes
465    double sizeS0;
466    for (OCol cls1 : types) {
467      sizeS0 = 0.0;
468      for (OCol cls2 : types) {
469        if (corrS.containsKey(Pair.of(cls1, cls2))) {
470          sizeS0 += corrS.get(Pair.of(cls1, cls2));
471          }
472        }
473      sizeS.put(cls1, sizeS0);
474      }
475    // Create overlaps
476    int ns = 0;
477    // Loop over OCol-OCol and create overlaps Edge OCol-OCol if non empty 
478    OCol cls1;
479    OCol cls2;
480    for (Map.Entry<Pair<OCol, OCol>, Double> entry : corrS.entrySet()) {
481      rel = entry.getKey();
482      cls1 = rel.first();
483      cls2 = rel.second();
484      weight = entry.getValue();
485      ocol1 = g().V().has("lbl",        "OCol"          ).
486                      has("survey",     cls1.survey()    ).
487                      has("classifier", cls1.classifier()).
488                      has("flavor",     cls1.flavor()    ).
489                      has("cls",        cls1.cls()       ).
490                      next();
491      ocol2 = g().V().has("lbl",        "OCol"          ).
492                      has("survey",     cls2.survey()    ).
493                      has("classifier", cls2.classifier()).
494                      has("flavor",     cls2.flavor()    ).
495                      has("cls",        cls2.cls()       ).
496                      next();
497      overlaps = ocol1.addEdge("overlaps", ocol2);
498      overlaps.property("lbl",          "overlaps");
499      overlaps.property("intersection", weight);
500      overlaps.property("sizeIn",       sizeS.get(cls1));
501      overlaps.property("sizeOut",      sizeS.get(cls2));
502      ns++;
503      }
504    commit();
505    log.info("" + ns + " object-object correlations generated");
506    }
507    
508  /** Create a new {@link FinkHBaseClient}. Singleton when url unchanged.
509    * @param hbaseUrl The HBase url as <tt>ip:port:table[:schema]</tt>.
510    * @return         The corresponding {@link FinkHBaseClient}, created and initialised if needed.
511    * @throws LomikelException If cannot be created. */
512  public FinkHBaseClient fhclient(String hbaseUrl) throws LomikelException {
513    if (hbaseUrl == null || hbaseUrl.equals(_fhclientUrl)) {
514      return _fhclient;
515      }
516    _fhclientUrl = hbaseUrl;
517    String[] url = hbaseUrl.split(":");
518    String ip     = url[0];
519    String port   = url[1];
520    String table  = url[2];
521    String schema = "";
522    if (url.length >= 4) {
523      schema = url[3];
524      }
525    _fhclient = new FinkHBaseClient(ip, port);
526    _fhclient.connect(table, schema);
527    return _fhclient;
528    }
529    
530  /** Get existing {@link FinkHBaseClient}.
531    * @return The corresponding {@link FinkHBaseClient}.
532    * @throws LomikelException If not yet created. */
533  public FinkHBaseClient fhclient() throws LomikelException {
534    if (_fhclient == null) {
535      throw new LomikelException("FinkHBaseClient not initialised");
536      }
537    return _fhclient;
538    }
539    
540  /** Give HBase url.
541    * @return The HBase url as <tt>ip:port:table[:schema]</tt>. */
542  public String hbaseUrl() {
543    return _fhclientUrl;
544    }
545    
546  private FinkHBaseClient _fhclient;
547  
548  private String _fhclientUrl;
549   
550  private String _now = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()).toString();
551 
552  private static String FINK_OBJECTS_WS = "https://api.ztf.fink-portal.org/api/v1/objects";
553  private static String FINK_LATESTS_WS = "https://api.ztf.fink-portal.org/api/v1/latests";
554  private static String FINK_ANOMALY_WS = "https://api.ztf.fink-portal.org/api/v1/anomaly";
555
556  /** Logging . */
557  private static Logger log = LogManager.getLogger(FinkGremlinRecipies.class);
558
559  }