Bunch of implementation

This commit is contained in:
Jamie Hardt
2021-05-31 23:22:16 -07:00
parent 3502eaddfd
commit c6be2ba404
4 changed files with 206 additions and 13 deletions

View File

@@ -1,9 +1,19 @@
from typing import Generator, Callable, Iterator
from enum import Enum
def apply_appends(source: Iterator,
should_append: Callable,
do_append: Callable) -> Generator:
"""
:param source:
:param should_append: Called with two variables a and b, your
function should return true if b should be
appended to a
:param do_append: Called with two variables a and b, your function
should return
:returns: A Generator
"""
this_element = next(source)
for element in source:
if should_append(this_element, element):

View File

@@ -1,14 +1,11 @@
from .doc_entity import SessionDescriptor, TrackDescriptor, TrackClipDescriptor
from typing import Optional, Generator, List, Callable
from tagged_string_parser_visitor import parse_tags
from itertools import chain
from functools import reduce
from typing import Optional, Generator
from fractions import Fraction
# field_map maps tags in the text export to fields in FMPXMLRESULT
# - tuple field 0 is a list of tags, the first tag with contents will be used as source
# - tuple field 1 is the field in FMPXMLRESULT
# - tuple field 2 the constructor/type of the field
from .tag_mapping import TagMapping
adr_field_map = ((['Title', 'PT.Session.Name'], 'Title', str),
(['Supv'], 'Supervisor', str),
@@ -46,6 +43,7 @@ adr_field_map = ((['Title', 'PT.Session.Name'], 'Title', str),
)
class ADRLine:
title: str
supervisor: str
@@ -58,9 +56,12 @@ class ADRLine:
priority: int
cue_number: str
character_id: str
character_name: str
actor_name: str
prompt: str
reason: str
requested_by: str
time_budget_mins: float
note: str
spot: str
shot: str
@@ -70,7 +71,40 @@ class ADRLine:
omitted: bool
adlib: bool
optional: bool
done: bool
adr_tag_to_line_map = (
TagMapping(source='Title', target="title", alt=TagMapping.ContentSource.Session),
TagMapping(source="Supv", target="supervisor"),
TagMapping(source="Client", target="client"),
TagMapping(source="Sc", target="scene"),
TagMapping(source="Ver", target="version"),
TagMapping(source="Reel", target="reel"),
TagMapping(source="P", target="priority"),
TagMapping(source="QN", target="cue_number"),
TagMapping(source="CN", target="character_id"),
TagMapping(source="Char", target="character_name", alt=TagMapping.ContentSource.Track),
TagMapping(source="Actor", target="actor_name"),
TagMapping(source="Line", target="prompt", alt=TagMapping.ContentSource.Clip),
TagMapping(source="R", target="reason"),
TagMapping(source="Rq", target="requested_by"),
TagMapping(source="Mins", target="time_budget_mins",
formatter=(lambda n: float(n))),
TagMapping(source="Note", target="note"),
TagMapping(source="Spot", target="spot"),
TagMapping(source="Shot", target="shot"),
TagMapping(source="EFF", target="effort",
formatter=(lambda x: len(x) > 0)),
TagMapping(source="TV", target="tv",
formatter=(lambda x: len(x) > 0)),
TagMapping(source="TBW", target="tbw",
formatter=(lambda x: len(x) > 0)),
TagMapping(source="OMIT", target="omitted",
formatter=(lambda x: len(x) > 0)),
TagMapping(source="ADLIB", target="adlib",
formatter=(lambda x: len(x) > 0)),
TagMapping(source="OPT", target="optional",
formatter=(lambda x: len(x) > 0))
)
@staticmethod
def from_clip(clip: TrackClipDescriptor,
@@ -85,5 +119,3 @@ class ADRLine:
line = ADRLine.from_clip(track_clip, track, session)
if line is not None:
yield line

View File

@@ -1,6 +1,7 @@
from fractions import Fraction
from ptulsconv.broadcast_timecode import smpte_to_frame_count
from typing import Tuple, List, Generator
from collections import namedtuple
from . import apply_appends
from .tagged_string_parser_visitor import parse_tags
@@ -21,6 +22,29 @@ class SessionDescriptor:
self.tracks = kwargs['tracks']
self.markers = kwargs['markers']
def markers_timed(self):
for marker in self.markers:
marker_time = self.header.convert_timecode(marker.location)
yield marker, marker_time
def tracks_clips(self):
for track_idx, track in enumerate(self.tracks):
for clip in track.clips:
yield track_idx, track, clip
def track_clips_timed(self) -> Generator[Tuple[int, "TrackDescriptor", "TrackClipDescriptor",
Fraction, Fraction, Fraction]]:
"""
:return: A Generator that yields track, clip, start time, finish time, and timestamp
"""
for track_idx, track, clip in self.tracks_clips():
start_time = self.header.convert_timecode(clip.start_timecode)
finish_time = self.header.convert_timecode(clip.finish_timecode)
timestamp_time = self.header.convert_timecode(clip.timestamp) \
if clip.timestamp is not None else None
yield track_idx, track, clip, start_time, finish_time, timestamp_time
class HeaderDescriptor:
session_name: str
@@ -44,7 +68,7 @@ class HeaderDescriptor:
self.count_clips = kwargs['count_clips']
self.count_files = kwargs['count_files']
def convert_timecode(self, tc_string) -> Fraction:
def convert_timecode(self, tc_string: str) -> Fraction:
frame_count = smpte_to_frame_count(tc_string,
self.logical_fps,
self.timecode_drop_frame)
@@ -108,8 +132,8 @@ class TrackClipDescriptor:
channel: int
event: int
clip_name: str
start_time: str
finish_time: str
start_timecode: str
finish_timecode: str
duration: str
timestamp: str
state: str
@@ -118,8 +142,8 @@ class TrackClipDescriptor:
self.channel = kwargs['channel']
self.event = kwargs['event']
self.clip_name = kwargs['clip_name']
self.start_time = kwargs['start_time']
self.finish_time = kwargs['finish_time']
self.start_timecode = kwargs['start_time']
self.finish_timecode = kwargs['finish_time']
self.duration = kwargs['duration']
self.timestamp = kwargs['timestamp']
self.state = kwargs['state']

View File

@@ -0,0 +1,127 @@
from enum import Enum
from typing import Optional, Callable, Any, List
from .doc_entity import SessionDescriptor, TrackClipDescriptor
from .tagged_string_parser_visitor import parse_tags, TagPreModes
from . import apply_appends
from fractions import Fraction
class TagCompiler:
session: SessionDescriptor
def timespan_tags(self, at: Fraction, track_index: int):
retval = dict()
for this_track_idx, _, clip, start, finish, _ in self.session.track_clips_timed():
if this_track_idx > track_index:
break
clip_parsed = parse_tags(clip)
if clip_parsed.mode == TagPreModes.TIMESPAN and start <= at < finish:
retval.update(clip_parsed.tag_dict)
return retval
def marker_tags(self, at):
retval = dict()
return retval
def combined_clips(self):
def should_append(_, rhs):
parsed = parse_tags(rhs[0][2].clip_name)
return parsed.mode == TagPreModes.APPEND
def do_append(lhs: List, rhs: List):
return lhs + rhs
source = ([x] for x in self.session.track_clips_timed())
yield from apply_appends(source, should_append, do_append)
def coalesce_tags(self, clip_tags: dict, track_tags: dict, timespan_tags: dict,
marker_tags: dict, session_tags: dict):
effective_tags = session_tags
effective_tags.update(marker_tags)
effective_tags.update(timespan_tags)
effective_tags.update(track_tags)
effective_tags.update(clip_tags)
return effective_tags
def compiled_clips(self):
session_parsed = parse_tags(self.session.header.session_name)
for track_idx, track, clip, start, finish, _ in self.session.track_clips_timed():
clip_parsed = parse_tags(clip.clip_name)
track_parsed = parse_tags(track.name)
timespan_tags = self.timespan_tags(start, track_idx)
marker_tags = self.marker_tags(start)
tags = self.coalesce_tags(clip_parsed.tag_dict, track_parsed.tag_dict,
timespan_tags, marker_tags, session_parsed.tag_dict)
yield track, clip, tags, start, finish
class TagMapping:
class ContentSource(Enum):
Session = 1,
Track = 2,
Clip = 3,
source: str
alternate_source: Optional[ContentSource]
formatter: Callable[[str], Any]
@staticmethod
def apply_rules(rules: List['TagMapping'],
tags: dict,
clip_content: str,
track_content: str,
session_content: str,
to: object):
done = set()
for rule in rules:
if rule.target in done:
continue
if rule.apply(tags, clip_content, track_content, session_content, to):
done.update(rule.target)
def __init__(self, source: str,
target: str,
alt: Optional[ContentSource] = None,
formatter=None):
self.source = source
self.target = target
self.alternate_source = alt
self.formatter = formatter or (lambda x: x)
def apply(self, tags: dict,
clip_content: str,
track_content: str,
session_content: str, to: object) -> bool:
setter = getattr(to, self.target)
new_value = None
if self.source in tags.keys():
new_value = tags[self.source]
elif self.alternate_source == 1:
new_value = session_content
elif self.alternate_source == 2:
new_value = track_content
elif self.alternate_source == 3:
new_value = clip_content
if new_value is not None:
setter(self.formatter(new_value))
return True
else:
return False