Find dependency paths between a selected pair of jobs or boxes.
This is an offline Python prototype: it reads local JIL files and writes the chosen output file. It does not query or control a live AutoSys scheduler. The parser supports the patterns implemented below, not the complete JIL language or scheduler semantics. Review its output against the original files before relying on it.
import os
import re
from collections import defaultdict
class AutoSysJobVisualizer:
def __init__(self, root_folder):
self.root_folder = root_folder
self.all_jobs = {}
self.box_jobs = defaultdict(list)
self.box_names = set()
def parse_jil_file(self, file_path):
jobs = {}
current_job = None
with open(file_path, 'r') as file:
for line in file:
line = line.strip()
if line.startswith('insert_job:'):
current_job = line.split()[1]
jobs[current_job] = {'conditions': [], 'box_name': None}
elif line.startswith('condition:') and current_job:
conditions = re.findall(r'([sdfn])\((.*?)\)', line)
jobs[current_job]['conditions'].extend(conditions)
elif line.startswith('box_name:') and current_job:
box_name = line.split()[1]
jobs[current_job]['box_name'] = box_name
self.box_jobs[box_name].append(current_job)
self.box_names.add(box_name)
return jobs
def find_all_dependencies(self, jobs):
dependencies = defaultdict(list)
for job, details in jobs.items():
for condition_type, condition_job in details['conditions']:
dependencies[condition_job].append((job, f'con: {condition_type}'))
if details['box_name']:
box_name = details['box_name']
dependencies[box_name].append((job, 'box: contains'))
return dependencies
def find_paths(self, start, end, dependencies, path=None):
if path is None:
path = []
path = path + [start]
if start == end:
return [path]
if start not in dependencies:
return []
paths = []
for node, dep_type in dependencies[start]:
if node not in path:
new_paths = self.find_paths(node, end, dependencies, path)
for p in new_paths:
paths.append(p)
return paths
def find_all_relationships(self, job1, job2, dependencies):
paths = self.find_paths(job1, job2, dependencies)
return paths
def run(self, output_file, job1, job2):
for dirpath, _, filenames in os.walk(self.root_folder):
for filename in filenames:
if filename.endswith('.jil'):
file_path = os.path.join(dirpath, filename)
jobs = self.parse_jil_file(file_path)
self.all_jobs.update(jobs)
dependencies = self.find_all_dependencies(self.all_jobs)
relationships = self.find_all_relationships(job1, job2, dependencies)
output = "\nRelationships between jobs:\n"
for path in relationships:
detailed_path = []
for i in range(len(path) - 1):
node, next_node = path[i], path[i + 1]
for dep in dependencies[node]:
if dep[0] == next_node:
detailed_path.append(f"{node} ({dep[1]})")
detailed_path.append(path[-1])
output += " -> ".join(detailed_path) + "\n"
with open(output_file, 'w') as file:
file.write(output)
if __name__ == "__main__":
root_folder = '/path/to/your/jil/files' # Change to your JIL files directory
output_file = 'output.txt' # Change to your desired output file path
job1 = 'job_a' # Change to the first job/box name
job2 = 'job_b' # Change to the second job/box name
visualizer = AutoSysJobVisualizer(root_folder)
visualizer.run(output_file, job1, job2)