Class: JobWorkflow::Monitoring::DagLayout

Inherits:
Object
  • Object
show all
Defined in:
lib/job_workflow/monitoring/dag_layout.rb,
sig/generated/job_workflow/monitoring/dag_layout.rbs

Constant Summary collapse

NODE_WIDTH =

RBS:

  • @nodes: Array[Hash[Symbol, untyped]]

  • @edges: Array[Hash[Symbol, untyped]]

  • @node_positions: Hash[Symbol, Hash[Symbol, Integer]]

Returns:

  • (::Integer) 
224
NODE_HEIGHT =

Returns:

  • (::Integer) 
84
COLUMN_GAP =

Returns:

  • (::Integer) 
56
ROW_GAP =

Returns:

  • (::Integer) 
24
PADDING =

Returns:

  • (::Integer) 
16
LABEL_LIMIT =

Returns:

  • (::Integer) 
24

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(tasks:) ⇒ DagLayout

: (tasks: Array[Hash[Symbol, untyped]]) -> void

Parameters:

  • tasks: (Array[Hash[Symbol, untyped]]) 


18
19
20
21
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 18

def initialize(tasks:)
  validate_tasks!(tasks)
  @tasks = tasks
end

Instance Attribute Details

#tasks ⇒ Array[Hash[Symbol, untyped]] (readonly)

Signature:

  • Array[Hash[Symbol, untyped]]

Returns:

  • (Array[Hash[Symbol, untyped]]) 


35
36
37
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 35

def tasks
  @tasks
end

Instance Method Details

#canvas_height ⇒ Integer

: () -> Integer

Returns:

  • (Integer) 


146
147
148
149
150
151
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 146

def canvas_height
  return 0 if nodes.empty?

  max_row = node_positions.values.map { |position| position.fetch(:row) }.max || 0
  (PADDING * 2) + ((max_row + 1) * NODE_HEIGHT) + (max_row * ROW_GAP)
end

#canvas_width ⇒ Integer

: () -> Integer

Returns:

  • (Integer) 


138
139
140
141
142
143
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 138

def canvas_width
  return 0 if nodes.empty?

  max_column = node_positions.values.map { |position| position.fetch(:column) }.max || 0
  (PADDING * 2) + ((max_column + 1) * NODE_WIDTH) + (max_column * COLUMN_GAP)
end

#dependency_column(task, positions) ⇒ Integer

: (Hash[Symbol, untyped], Hash[Symbol, Hash[Symbol, Integer]]) -> Integer

Parameters:

  • (Hash[Symbol, untyped])
  • (Hash[Symbol, Hash[Symbol, Integer]])

Returns:

  • (Integer) 


103
104
105
106
107
108
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 103

def dependency_column(task, positions)
  depends_on = task.fetch(:depends_on)
  return 0 if depends_on.empty?

  depends_on.map { |dependency_name| positions.fetch(dependency_name).fetch(:column) + 1 }.max || 0
end

#each_progress_label(progress) ⇒ String

: (Hash[Symbol, Integer]) -> String

Parameters:

  • (Hash[Symbol, Integer])

Returns:

  • (String) 


185
186
187
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 185

def each_progress_label(progress)
  "each #{progress.fetch(:succeeded)}/#{progress.fetch(:total)}"
end

#edge_path(from_node, to_node) ⇒ String

: (Hash[Symbol, untyped], Hash[Symbol, untyped]) -> String

Parameters:

  • (Hash[Symbol, untyped])
  • (Hash[Symbol, untyped])

Returns:

  • (String) 


122
123
124
125
126
127
128
129
130
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 122

def edge_path(from_node, to_node)
  start_x = from_node.fetch(:x) + from_node.fetch(:width)
  start_y = from_node.fetch(:y) + (from_node.fetch(:height) / 2)
  end_x = to_node.fetch(:x)
  end_y = to_node.fetch(:y) + (to_node.fetch(:height) / 2)
  mid_x = ((start_x + end_x) / 2.0).round(2)

  "M #{start_x} #{start_y} L #{mid_x} #{start_y} L #{mid_x} #{end_y} L #{end_x} #{end_y}"
end

#edge_view(from_name, to_name) ⇒ Hash[Symbol, untyped]

: (Symbol, Symbol) -> Hash[Symbol, untyped]

Parameters:

  • (Symbol)
  • (Symbol)

Returns:

  • (Hash[Symbol, untyped]) 


111
112
113
114
115
116
117
118
119
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 111

def edge_view(from_name, to_name)
  from_node = node_view(from_name)
  to_node = node_view(to_name)
  {
    from: from_name,
    to: to_name,
    path: edge_path(from_node, to_node)
  }
end

#edges ⇒ Array[Hash[Symbol, untyped]]

: () -> Array[Hash[Symbol, untyped]]

Returns:

  • (Array[Hash[Symbol, untyped]]) 


79
80
81
82
83
84
85
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 79

def edges
  @edges ||= tasks.flat_map do |task|
    task.fetch(:depends_on).map do |dependency_name|
      edge_view(dependency_name, task.fetch(:name))
    end
  end
end

#node_meta_label(task) ⇒ String?

: (Hash[Symbol, untyped]) -> String?

Parameters:

  • (Hash[Symbol, untyped])

Returns:

  • (String, nil) 


172
173
174
175
176
177
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 172

def node_meta_label(task)
  return root_task_label if task.fetch(:depends_on).empty?
  return each_progress_label(task.fetch(:each_progress)) if task.fetch(:each)

  nil
end

#node_positions ⇒ Hash[Symbol, Hash[Symbol, Integer]]

: () -> Hash[Symbol, Hash[Symbol, Integer]]

Returns:

  • (Hash[Symbol, Hash[Symbol, Integer]]) 


88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 88

def node_positions
  @node_positions ||= begin
    column_rows = Hash.new(0) #: Hash[Integer, Integer]
    positions = {} #: Hash[Symbol, Hash[Symbol, Integer]]

    tasks.each_with_object(positions) do |task, current_positions|
      column = dependency_column(task, current_positions)
      row = column_rows[column]
      column_rows[column] += 1
      current_positions[task.fetch(:name)] = { column:, row: }
    end
  end
end

#node_view(task_name) ⇒ Hash[Symbol, untyped]

: (Symbol) -> Hash[Symbol, untyped]

Parameters:

  • (Symbol)

Returns:

  • (Hash[Symbol, untyped]) 


133
134
135
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 133

def node_view(task_name)
  nodes.find { |task| task.fetch(:name) == task_name } || raise(KeyError, task_name.to_s)
end

#nodes ⇒ Array[Hash[Symbol, untyped]]

: () -> Array[Hash[Symbol, untyped]]

Returns:

  • (Array[Hash[Symbol, untyped]]) 


63
64
65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 63

def nodes
  @nodes ||= tasks.map do |task|
    position = node_positions.fetch(task.fetch(:name))
    task.merge(
      x: x_for(position.fetch(:column)),
      y: y_for(position.fetch(:row)),
      width: NODE_WIDTH,
      height: NODE_HEIGHT,
      label: task.fetch(:name).to_s,
      truncated_label: truncate_label(task.fetch(:name)),
      meta_label: node_meta_label(task)
    )
  end
end

#root_task_label ⇒ String

: () -> String

Returns:

  • (String) 


180
181
182
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 180

def root_task_label
  "root task"
end

#to_h ⇒ Hash[Symbol, untyped]

: () -> Hash[Symbol, untyped]

Returns:

  • (Hash[Symbol, untyped]) 


24
25
26
27
28
29
30
31
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 24

def to_h
  {
    width: canvas_width,
    height: canvas_height,
    nodes:,
    edges:
  }
end

#truncate_label(task_name) ⇒ String

: (Symbol) -> String

Parameters:

  • (Symbol)

Returns:

  • (String) 


164
165
166
167
168
169
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 164

def truncate_label(task_name)
  label = task_name.to_s
  return label if label.length <= LABEL_LIMIT

  "#{label[0, LABEL_LIMIT - 1]}…"
end

#validate_dependencies!(task, seen_names) ⇒ void

This method returns an undefined value.

: (Hash[Symbol, untyped], Hash[Symbol, bool]) -> void

Parameters:

  • (Hash[Symbol, untyped])
  • (Hash[Symbol, bool]) 


48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 48

def validate_dependencies!(task, seen_names)
  missing_dependencies = task.fetch(:depends_on).reject { |dependency_name| seen_names[dependency_name] }
  return if missing_dependencies.empty?

  task_name = task.fetch(:name)
  dependency_names = missing_dependencies.join(", ")

  raise(
    ArgumentError,
    "DagLayout tasks must be topologically sorted; " \
    "#{task_name} depends on unavailable prior tasks: #{dependency_names}"
  )
end

#validate_tasks!(tasks) ⇒ void

This method returns an undefined value.

: (Array[Hash[Symbol, untyped]]) -> void

Parameters:

  • (Array[Hash[Symbol, untyped]]) 


38
39
40
41
42
43
44
45
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 38

def validate_tasks!(tasks)
  seen_names = {} #: Hash[Symbol, bool]

  tasks.each do |task|
    validate_dependencies!(task, seen_names)
    seen_names[task.fetch(:name)] = true
  end
end

#x_for(column) ⇒ Integer

: (Integer) -> Integer

Parameters:

  • (Integer)

Returns:

  • (Integer) 


154
155
156
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 154

def x_for(column)
  PADDING + (column * (NODE_WIDTH + COLUMN_GAP))
end

#y_for(row) ⇒ Integer

: (Integer) -> Integer

Parameters:

  • (Integer)

Returns:

  • (Integer) 


159
160
161
 
# File 'lib/job_workflow/monitoring/dag_layout.rb', line 159

def y_for(row)
  PADDING + (row * (NODE_HEIGHT + ROW_GAP))
end