Simpler revised filling mechanism

Author: akariv

datapackage_pipelines_budgetkey/pipelines/budget/national/processed/combine-immediate-children.py

@@ -6,7 +6,7 @@ def nop(x):
     return x
 
 
-FIELDS = ['code', 'title', 'net_revised']
+FIELDS = ['code', 'title', 'net_allocated', 'net_revised']
 
 
 params, dp, res_iter = ingest()
@@ -15,9 +15,13 @@ def nop(x):
 
 def combine_immediate_children(rows):
     for row in rows:
-        row['children-net_revised'] = map(int, row['children-net_revised'])
+        if row['children-net_revised'] is None:
+            logging.warning('children-net_revised is undefined, %r', row)
+            row['children-net_revised'] = row['children-net_allocated']
+        for f in ('children-net_revised', 'children-net_allocated'):
+            row[f] = map(int, row[f])
         try:
-            children = zip(*(row['children-'+x] for x in FIELDS))
+            children = zip(*(row.get('children-'+x, []) for x in FIELDS))
         except TypeError:
             logging.error('Failed to process children for row %r', row)
             raise

datapackage_pipelines_budgetkey/pipelines/budget/national/processed/join_phases.py

@@ -1,8 +1,12 @@
+import copy
+
 from datapackage_pipelines.wrapper import spew, ingest
 from decimal import Decimal
 
 parameters, datapackage, res_iter = ingest()
 
+last_completed_year = str(parameters['last-completed-year'])
+
 amounts = [
     'net',
     'gross',
@@ -69,56 +73,64 @@
 resource['schema']['fields'] = new_fields
 
 
+def process_row(row, phase_key):
+    for amount, factor in zip(amounts, factors):
+        value = row[amount]
+        if value is not None and value.strip() != '':
+            value = Decimal(row[amount]) * factor
+        row[amount + '_' + phase_key] = value
+        del row[amount]
+
+    save = {}
+    for code_key, title_key in codes_and_titles:
+        save[code_key] = row[code_key]
+        if len(save[code_key]) % 2 == 1:
+            save[code_key] = '0' + save[code_key]
+        save[title_key] = row[title_key]
+        del row[code_key]
+        del row[title_key]
+
+    if int(save[codes_and_titles[0][0]]) != 0:
+        hierarchy = [['00', 'המדינה']]
+    else:
+        hierarchy = []
+    for i, (code_key, title_key) in enumerate(codes_and_titles):
+        expected_length = i*2 + 4
+        row['code'] = '0'*(expected_length-len(save[code_key])) + save[code_key]
+        row['title'] = save[title_key] or 'לא ידוע'
+        row['hierarchy'] = hierarchy
+        row['parent'] = None if len(hierarchy) == 0 else hierarchy[-1][0]
+        yield row
+        hierarchy.append([row['code'], row['title']])
+
+    row['hierarchy'] = None
+    row['parent'] = None
+
+    row['code'] = 'C%d' % (int(row['func_cls_code_1']),)
+    row['title'] = '%s' % (row['func_cls_title_1'],)
+    yield row
+
+    row['code'] = 'C%d%02d' % (int(row['func_cls_code_1']), int(row['func_cls_code_2']))
+    row['title'] = '%s/%s' % (row['func_cls_title_1'], row['func_cls_title_2'])
+    yield row
+
+    if not row['code'].startswith('0000'):
+        row['code'] = '00'
+        row['title'] = 'המדינה'
+        row['hierarchy'] = []
+        yield row
+
+
 def process_first(rows):
-    for row in rows:
+    for i, row in enumerate(rows):
         phase_key = phases[row['phase']]
         del row['phase']
 
-        for amount, factor in zip(amounts, factors):
-            value = row[amount]
-            if value is not None and value.strip() != '':
-                value = Decimal(row[amount]) * factor
-            row[amount + '_' + phase_key] = value
-            del row[amount]
-
-        save = {}
-        for code_key, title_key in codes_and_titles:
-            save[code_key] = row[code_key]
-            if len(save[code_key]) % 2 == 1:
-                save[code_key] = '0' + save[code_key]
-            save[title_key] = row[title_key]
-            del row[code_key]
-            del row[title_key]
-
-        if int(save[codes_and_titles[0][0]]) != 0:
-            hierarchy = [['00', 'המדינה']]
-        else:
-            hierarchy = []
-        for i, (code_key, title_key) in enumerate(codes_and_titles):
-            expected_length = i*2 + 4
-            row['code'] = '0'*(expected_length-len(save[code_key])) + save[code_key]
-            row['title'] = save[title_key] or 'לא ידוע'
-            row['hierarchy'] = hierarchy
-            row['parent'] = None if len(hierarchy) == 0 else hierarchy[-1][0]
-            yield row
-            hierarchy.append([row['code'], row['title']])
-
-        row['hierarchy'] = None
-        row['parent'] = None
-
-        row['code'] = 'C%d' % (int(row['func_cls_code_1']),)
-        row['title'] = '%s' % (row['func_cls_title_1'],)
-        yield row
-
-        row['code'] = 'C%d%02d' % (int(row['func_cls_code_1']), int(row['func_cls_code_2']))
-        row['title'] = '%s/%s' % (row['func_cls_title_1'], row['func_cls_title_2'])
-        yield row
+        row_ = copy.deepcopy(row)
+        yield from process_row(row_, phase_key)
 
-        if not row['code'].startswith('0000'):
-            row['code'] = '00'
-            row['title'] = 'המדינה'
-            row['hierarchy'] = []
-            yield row
+        if row['year'] > last_completed_year:
+            yield from process_row(row, 'revised')
 
 
 def process_resources(res_iter_):

datapackage_pipelines_budgetkey/pipelines/budget/national/processed/pipeline-spec.yaml

@@ -85,6 +85,9 @@ aggregated-yearly:
           commitment_balance:
             - יתרת התחיבויות
     - run: join_phases
+      parameters:
+        last-completed-year: 2016
+
     - run: topical_sums
     - run: join_categories
     - run: set_types
@@ -235,8 +238,8 @@ aggregated-yearly:
             name: "econ_cls_title_1_\u05e9\u05db\u05e8"
             aggregate: sum
 
-    - run: fill_in_revised
-
+#    - run: fill_in_revised
+#
     - run: join
       parameters:
         source:
@@ -255,6 +258,9 @@ aggregated-yearly:
           children-title:
             name: title
             aggregate: array
+          children-net_allocated:
+            name: net_allocated
+            aggregate: array
           children-net_revised:
             name: net_revised
             aggregate: array