Setting the file. One moment.
Score Units · Agent Advisor · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page Add Capabilities
81
Creating Amazon Aurora Db Cluster With Instances
104
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
scripts/score_units.py
scripts/ score_units.py
Python · 88 lines · 3 KB
sys
16 from pathlib import Path
17
18 sys.path.insert( 0 , str (Path( __file__ ).resolve().parent))
19
20 import scoring # noqa: E402
21
22 RUN_EVIDENCE_FILENAME = "current-run-verifications.json"
23
24
25 def _load_run_evidence (answers_path: Path):
26 """Load evidence created for this run, or return no evidence when absent."""
27 evidence_path = answers_path.parent / RUN_EVIDENCE_FILENAME
28 if not evidence_path.exists():
29 return None
30 evidence = json.loads(evidence_path.read_text( encoding = "utf-8" ))
31 if not scoring.is_run_materialized_evidence(evidence):
32 raise ValueError ( f " { evidence_path } : invalid run-materialized evidence" )
33 if evidence[ "run_id" ] != answers_path.parent.name:
34 raise ValueError (
35 f " { evidence_path } : run_id must match the answers.json run directory"
36 )
37 return evidence
38
39
40 def main (argv = None ) -> int :
41 if argv is None :
42 if len (sys.argv) != 2 :
43 print ( __doc__ .strip().splitlines()[ - 1 ], file = sys.stderr)
44 return 2
45 answers_path = Path(sys.argv[ 1 ])
46 else :
47 if len (argv) != 1 :
48 print ( __doc__ .strip().splitlines()[ - 1 ], file = sys.stderr)
49 return 2
50 answers_path = Path(argv[ 0 ])
51 answers = json.loads(answers_path.read_text( encoding = "utf-8" ))
52 run_evidence = _load_run_evidence(answers_path)
53 entry_point = answers.get( "entry_point" , "build_scratch" )
54 system_answers = {
55 key: value
56 for key, value in answers[ "system" ].items()
57 if key not in ( "current_run_verifications" , "provenance" )
58 }
59 units = {
60 unit_id: scoring.score(
61 {
62 "entry_point" : entry_point,
63 "answers" : {
64 ** system_answers,
65 ** {
66 k: v
67 for k, v in info.items()
68 if k not in ( "workload_class" , "provenance" )
69 },
70 },
71 },
72 run_evidence = run_evidence,
73 )
74 for unit_id, info in answers[ "units" ].items()
75 if info.get( "workload_class" ) == "agent_session"
76 }
77 out = { "units" : units}
78 # Collapse mirror: the scope gate guarantees >=1 agent_session unit, so
79 # `units` is non-empty and a mirror always exists.
80 primary = answers.get( "primary_unit" )
81 mirror = units.get(primary) or next ( iter (units.values()))
82 out.update(mirror)
83 print (json.dumps(out, indent = 2 ))
84 return 0
85
86
87 if __name__ == "__main__" :
88 sys.exit(main())