"""AGM results module."""
import json
import re
from html import escape
from pathlib import Path
from urllib import request
from ..datetime import dt
from ..html import HTMLCollapsible, html
from ..misc import SSL_CONTEXT
[docs]class AGMResults:
"""AGM results object.
Parameters
----------
results: dict or pathlib.Path
AGM formatted results.
"""
def __init__(self, results):
if isinstance(results, Path):
with results.open('r') as fp:
self.data = json.load(fp)
elif isinstance(results, dict):
self.data = results
else:
raise TypeError('Results must be a pathlib.Path or a dict. '
f'`{results.__class__.__name__}` provided.')
# Parse the data content
self.ptr = AGMResultsPTR(self['ptr'])
self.status = AGMResultsStatus(self['results']['success'])
session_id = self['results'].get('session_id')
if self.status.success:
self.log = AGMResultsLog(self['results']['output'], session_id=session_id)
self.ck = AGMResultsCK(self)
self.ptr_resolved = AGMResultsResolvedPTR(self)
self.quaternions = AGMResultsQuaternions(self['results']['quaternions'])
else:
self.log = AGMResultsLog(self['results']['errors'], session_id=session_id)
self.ck = None
self.ptr_resolved = None
self.quaternions = None
def __str__(self):
return json.dumps(self.data, indent=4)
def __repr__(self):
return f'<{self.__class__.__name__}> {self}'
def _repr_html_(self):
return '\n'.join([
html(self.status),
'<hr/>',
html(HTMLCollapsible(
('Input Parameters', AGMInputParameters(self)),
('Output Results', AGMOutputResults(self) if self.status.success else ''),
('Log', self.log, '' if self.status.success else 'expand'),
)),
])
def __iter__(self):
if self.status.success:
return iter(self.ck) # Load-able by spiceypy
raise IOError('Simulation failed. No CK available')
def __getitem__(self, item):
try:
return self.data[item]
except KeyError:
raise KeyError(f'`{item}` is not present in the results.') from None
@property
def success(self) -> bool:
"""Results status success flag."""
return self.status.success
[docs]class AGMOutputResults:
"""AGM output results."""
def __init__(self, res):
self.ck = res.ck
self.ptr_resolved = res.ptr_resolved
self.quaternions = res.quaternions
def __repr__(self):
return str({
'ck': self.ck,
'ptr_resolved': self.ptr_resolved,
'quaternions': self.quaternions,
})
def _repr_html_(self):
return '\n'.join([
'<ul style="color:#777">',
f'<li><b>CK:</b> <code>{self.ck}</code></li>',
f'<li><b>Resolved PTR:</b> <code>{self.ptr_resolved}</code></li>',
'</ul>',
HTMLCollapsible(('Quaternions', self.quaternions)).html,
])
[docs]class AGMResultsPTR:
"""AGM PTR object."""
def __init__(self, ptr):
self.ptr = ptr
def __repr__(self):
return self.ptr
def _repr_html_(self):
return f'<pre><code>{escape(self.ptr)}</code></pre>'
[docs]class AGMResultsQuaternions:
"""AGM results quaternions object."""
HTML_MAX_LINES = 25
def __init__(self, quaternions):
self._raw = quaternions
self.data = [
[dt(time), (float(qx), float(qy), float(qz), float(qw))]
for line in quaternions.splitlines()[1:] # Skip header
for time, qx, qy, qz, qw in [line.split(',')]
]
def __repr__(self):
return '\n'.join(
f'{t} | {x:.16f} | {y:0.16f} | {z:.16f} | {w:.16f}'
for t, (x, y, z, w) in self.data
)
def __len__(self):
return len(self.data)
def __iter__(self):
return iter(self.data)
def __getitem__(self, item):
return self.data[item]
def _repr_html_(self):
header = '<tr><th>time</th><th>qx</th><th>qy</th><th>qz</th><th>qw</tr>'
lines = ''.join([
f'<tr><td><em>{t}</em></td><td>{x}</td>'
f'<td>{y}</td><td>{z}</td><td>{w}</td></tr>'
for t, (x, y, z, w) in self.data[:self.HTML_MAX_LINES]
])
if len(self) > self.HTML_MAX_LINES:
lines += (
'<tfoot>'
'<tr><td>…</td><td>…</td><td>…</td>'
'<td>…</td><td>…</td></tr>'
'<tr><td colspan="5" style="text-align: center;">'
'<em>Use <code>print()</code> to display all elements.</em>'
'</td></tr></tfoot>'
)
return '<table>' + header + lines + '</table>'
class AGMResultsFile:
"""AGM results file object (CK or resolved PTR)."""
EXT = None # File extension
PATH = None # Results path key
def __init__(self, res):
if cache := res['cache']:
self.fname = Path(cache['location']) / (cache['md5_hash'] + self.EXT)
else:
key = res['results'][self.PATH].split('/')[1]
self.fname = Path(f'AGM_{key}{self.EXT}')
if '://' in (url := res['endpoint']):
protocol, uri = url.split('://')
self.url = protocol + '://' + uri.split('/')[0] + res['results'][self.PATH]
else:
self.url = None
def __str__(self):
return str(self.fname) if self.fname.exists() else self.url
def __repr__(self):
return f'<{self.__class__.__name__}> {self}'
def __iter__(self):
if not self.fname.exists():
self.download()
yield str(self.fname) # Load-able by spiceypy
def download(self):
"""Download the file."""
if self.url is None:
raise ValueError('Endpoint is not a remote URL.')
# Download the file
with request.urlopen(self.url, context=SSL_CONTEXT) as resp:
self.fname.write_bytes(resp.read())
return self.fname
def save(self, fout, overwrite=False):
"""Save the file into an new location."""
fout = Path(fout)
if not overwrite and fout.exists():
raise FileExistsError(fout)
for f in self:
content = Path(f).read_bytes()
fout.write_bytes(content)
self.fname = fout
return self.fname
[docs]class AGMResultsCK(AGMResultsFile):
"""AGM results CK object."""
EXT = '.ck'
PATH = 'ck_path'
class AGMResultsResolvedPTR(AGMResultsFile):
"""AGM results resolved PTR object."""
EXT = '.ptx'
PATH = 'ptr_resolved_path'
def __repr__(self):
return f'<{self.__class__.__name__}> {self}\n\n{self.content}'
@property
def content(self):
"""Resolved PTR file content."""
if not self.fname.exists():
self.download()
return self.fname.read_text(encoding='utf-8')
[docs]class AGMResultsLog:
"""AGM results log object."""
HTML_MAX_LINES = 25
line = re.compile(r'\[(\w+)\]\s*<([\w\s><]+)>\s*([\w\s\/\.\:+\-_"]+)')
def __init__(self, log, session_id=None):
self.log = [
(msg['severity'], msg['module'], msg['text'])
for msg in log
if isinstance(msg, dict)
]
self.session_id = session_id
def __repr__(self):
return '\n'.join(' | '.join(line) for line in self.log)
def _repr_html_(self):
session_id = (
'<ul style="color:#777">'
f'<li><b>Session ID:</b> <code>{self.session_id}</code></li>'
'</ul>\n'
) if self.session_id is not None else ''
lines = ''.join([
f'<tr><td>{flag}</td><td>{tag}</td>'
f'<td style="text-align: left;"><em>{escape(msg)}</em></td></tr>'
for flag, tag, msg in self.log[:self.HTML_MAX_LINES]
])
if len(self) > self.HTML_MAX_LINES:
lines += (
'<tfoot>'
'<tr><td>…</td><td>…</td>'
'<td style="text-align: left;">…</td></tr>'
'<tr><td colspan="3" style="text-align: center;">'
'<em>Use <code>print()</code> to display all elements.</em>'
'</td></tr></tfoot>'
)
return session_id + '<table>' + lines + '</table>'
def __len__(self):
return len(self.log)
def __iter__(self):
return iter(self.log)
def __getitem__(self, item):
return self.log[item]
[docs]class AGMResultsStatus:
"""AGM results status object"""
def __init__(self, success):
self.success = success
def __repr__(self):
return 'Success' if self.success else 'Failure'
def _repr_html_(self):
return (
f'<p><span style="color: {self.color}">{self.symbol}</span> <b>{self}</b></p>'
)
@property
def failure(self):
"""Failure flag."""
return not self.success
@property
def symbol(self):
"""Status symbol."""
return '✔' if self.success else '✘'
@property
def color(self):
"""Status color."""
return '#2ca02c' if self.success else '#d62728'
