You need to do quite a bit of filtering and transformation to get the desired result. As you have learned, a good strategy for tackling larger problems is to solve the problem in small steps.
You decide to scope the first step so that the result is the raw JSON data transformed into a Table (a data structure provided by the RPA.Tables library) that is then easy to filter, sort, group, etc.:
producer.robot:
from robocorp.tasks import taskfrom RPA.HTTP import HTTPfrom RPA.JSON import JSONfrom RPA.Tables import Tableshttp = HTTP()json = JSON()table = Tables()@taskdef produce_traffic_data(): """ Inhuman Insurance, Inc. Artificial Intelligence System automation. Produces traffic data work items. """ http.download( url="https://github.com/robocorp/inhuman-insurance-inc/raw/main/RS_198.json", target_file="output/traffic.json", overwrite=True, ) traffic_data = load_traffic_data_as_table()@taskdef consume_traffic_data(): """ Inhuman Insurance, Inc. Artificial Intelligence System robot. Consumes traffic data work items. """ print("consume")def load_traffic_data_as_table(): json_data = json.load_json_from_file("output/traffic.json") table_from_json = table.create_table(json_data["value"]) return table_from_json
You add a new load_traffic_data_as_table() function that returns the JSON data in a Table format.
- The
load_json_from_file()function provided by theRPA.JSONlibrary returns the file contents (string) in JSON format. - The
create_table()function from theRPA.Tableslibrary converts the JSON format into a Table.
You run the robot and view the log. Your robot has transformed the raw JSON data into a Table with lots of rows:
Table( columns=[ 'Id', 'IndicatorCode', 'SpatialDimType', 'SpatialDim', 'TimeDimType', 'TimeDim', 'Dim1Type', 'Dim1', 'Dim2Type', 'Dim2', 'Dim3Type', 'Dim3', 'DataSourceDimType', 'DataSourceDim', 'Value', 'NumericValue', 'Low', 'High', 'Comments', 'Date', 'TimeDimensionValue', 'TimeDimensionBegin', 'TimeDimensionEnd' ], rows=11640 )
Viewing the whole table using a VS Code extensions
The table entry in the log is nice and concise but lacks the actual data.
To view the entire table, you decide to export the Table into a CSV file and view that as a human-readable representation using the Excel Viewer VS Code extension.
You call the super-handy Write table to CSV keyword:
table.write_table_to_csv(traffic_data, "output/test.csv")
After rerunning the robot, you see a test.csv file at the root. Clicking on it opens the raw CSV file. You click on the Open Preview icon at the top-right. A preview of the data is displayed as a table. What a nice way to manually inspect the intermediate results!
DRY - Don't repeat yourself
Always in search for tidy code, you see that you have broken the Don't repeat yourself principle by duplicating the path to the traffic.json file.
It's not a massive deal at the moment since you have so little code, but it still requires future maintainers to find and change the path in two places if the path needs to be changed.
Little repetition might eventually grow into lots of small repetition and make it slower to maintain your robot or even cause subtle bugs if someone forgets to replace all the instances of the duplicated things.
You decide to store the path of the traffic.json file into a variable and then refer to that variable instead. This way, you remove the duplication:
You also notice, that in load_traffic_data_as_table() you define a variable just to return it.But, you read that you can return an expression to be evaluated, not just a variable.So we are going to refactor this as well!
from robocorp.tasks import taskfrom RPA.HTTP import HTTPfrom RPA.JSON import JSONfrom RPA.Tables import Tableshttp = HTTP()json = JSON()table = Tables()TRAFFIC_JSON_FILE_PATH = "output/traffic.json"@taskdef produce_traffic_data(): """ Inhuman Insurance, Inc. Artificial Intelligence System automation. Produces traffic data work items. """ http.download( url="https://github.com/robocorp/inhuman-insurance-inc/raw/main/RS_198.json", target_file=TRAFFIC_JSON_FILE_PATH, overwrite=True, ) traffic_data = load_traffic_data_as_table() table.write_table_to_csv(traffic_data, "output/test.csv")@taskdef consume_traffic_data(): """ Inhuman Insurance, Inc. Artificial Intelligence System robot. Consumes traffic data work items. """ print("consume")def load_traffic_data_as_table(): json_data = json.load_json_from_file(TRAFFIC_JSON_FILE_PATH) return table.create_table(json_data["value"])
